Follow up to cached run_system (#15410)

# Objective

- Fixes #15373
- Fixes
https://github.com/bevyengine/bevy/pull/14920#issuecomment-2370428013

## Solution

- Make `IntoSystem::pipe` and `IntoSystem::map` return two new
(possibly-ZST) types that implement `IntoSystem` and whose `into_system`
method return the systems that were previously being returned by
`IntoSystem::pipe` and `IntoSystem::map`
- Don't eagerly call `IntoSystem::into_system` on the argument given to
`RunSystemCachedWith::new` to avoid losing its ZST-ness

## Testing

- Added a regression test for each issue

## Migration Guide

- `IntoSystem::pipe` and `IntoSystem::map` now return `IntoPipeSystem`
and `IntoAdapterSystem` instead of `PipeSystem` and `AdapterSystem`.
Most notably these types don't implement `System` but rather only
`IntoSystem`.

crates/bevy_ecs/src/schedule/condition.rs

@@ -1115,11 +1115,11 @@ pub mod common_conditions {
         CIn: SystemInput,
         C: Condition<Marker, CIn>,
     {
-        condition.pipe(|In(new): In<bool>, mut prev: Local<bool>| {
+        IntoSystem::into_system(condition.pipe(|In(new): In<bool>, mut prev: Local<bool>| {
             let changed = *prev != new;
             *prev = new;
             changed
-        })
+        }))
     }
 
     /// Generates a [`Condition`] that returns true when the result of
@@ -1171,11 +1171,13 @@ pub mod common_conditions {
         CIn: SystemInput,
         C: Condition<Marker, CIn>,
     {
-        condition.pipe(move |In(new): In<bool>, mut prev: Local<bool>| -> bool {
-            let now_true = *prev != new && new == to;
-            *prev = new;
-            now_true
-        })
+        IntoSystem::into_system(condition.pipe(
+            move |In(new): In<bool>, mut prev: Local<bool>| -> bool {
+                let now_true = *prev != new && new == to;
+                *prev = new;
+                now_true
+            },
+        ))
     }
 }
 

crates/bevy_ecs/src/system/adapter_system.rs

@@ -1,6 +1,6 @@
 use std::borrow::Cow;
 
-use super::{ReadOnlySystem, System};
+use super::{IntoSystem, ReadOnlySystem, System};
 use crate::{
     schedule::InternedSystemSet,
     system::{input::SystemInput, SystemIn},
@@ -62,6 +62,40 @@ pub trait Adapt<S: System>: Send + Sync + 'static {
     ) -> Self::Out;
 }
 
+/// An [`IntoSystem`] creating an instance of [`AdapterSystem`].
+#[derive(Clone)]
+pub struct IntoAdapterSystem<Func, S> {
+    func: Func,
+    system: S,
+}
+
+impl<Func, S> IntoAdapterSystem<Func, S> {
+    /// Creates a new [`IntoSystem`] that uses `func` to adapt `system`, via the [`Adapt`] trait.
+    pub const fn new(func: Func, system: S) -> Self {
+        Self { func, system }
+    }
+}
+
+#[doc(hidden)]
+pub struct IsAdapterSystemMarker;
+
+impl<Func, S, I, O, M> IntoSystem<Func::In, Func::Out, (IsAdapterSystemMarker, I, O, M)>
+    for IntoAdapterSystem<Func, S>
+where
+    Func: Adapt<S::System>,
+    I: SystemInput,
+    S: IntoSystem<I, O, M>,
+{
+    type System = AdapterSystem<Func, S::System>;
+
+    // Required method
+    fn into_system(this: Self) -> Self::System {
+        let system = IntoSystem::into_system(this.system);
+        let name = system.name();
+        AdapterSystem::new(this.func, system, name)
+    }
+}
+
 /// A [`System`] that takes the output of `S` and transforms it by applying `Func` to it.
 #[derive(Clone)]
 pub struct AdapterSystem<Func, S> {

crates/bevy_ecs/src/system/combinator.rs

@@ -10,7 +10,7 @@ use crate::{
     world::unsafe_world_cell::UnsafeWorldCell,
 };
 
-use super::{ReadOnlySystem, System};
+use super::{IntoSystem, ReadOnlySystem, System};
 
 /// Customizes the behavior of a [`CombinatorSystem`].
 ///
@@ -273,6 +273,40 @@ where
     }
 }
 
+/// An [`IntoSystem`] creating an instance of [`PipeSystem`].
+pub struct IntoPipeSystem<A, B> {
+    a: A,
+    b: B,
+}
+
+impl<A, B> IntoPipeSystem<A, B> {
+    /// Creates a new [`IntoSystem`] that pipes two inner systems.
+    pub const fn new(a: A, b: B) -> Self {
+        Self { a, b }
+    }
+}
+
+#[doc(hidden)]
+pub struct IsPipeSystemMarker;
+
+impl<A, B, IA, OA, IB, OB, MA, MB> IntoSystem<IA, OB, (IsPipeSystemMarker, OA, IB, MA, MB)>
+    for IntoPipeSystem<A, B>
+where
+    IA: SystemInput,
+    A: IntoSystem<IA, OA, MA>,
+    B: IntoSystem<IB, OB, MB>,
+    for<'a> IB: SystemInput<Inner<'a> = OA>,
+{
+    type System = PipeSystem<A::System, B::System>;
+
+    fn into_system(this: Self) -> Self::System {
+        let system_a = IntoSystem::into_system(this.a);
+        let system_b = IntoSystem::into_system(this.b);
+        let name = format!("Pipe({}, {})", system_a.name(), system_b.name());
+        PipeSystem::new(system_a, system_b, Cow::Owned(name))
+    }
+}
+
 /// A [`System`] created by piping the output of the first system into the input of the second.
 ///
 /// This can be repeated indefinitely, but system pipes cannot branch: the output is consumed by the receiving system.
@@ -296,7 +330,7 @@ where
 ///     world.insert_resource(Message("42".to_string()));
 ///
 ///     // pipe the `parse_message_system`'s output into the `filter_system`s input
-///     let mut piped_system = parse_message_system.pipe(filter_system);
+///     let mut piped_system = IntoSystem::into_system(parse_message_system.pipe(filter_system));
 ///     piped_system.initialize(&mut world);
 ///     assert_eq!(piped_system.run((), &mut world), Some(42));
 /// }

crates/bevy_ecs/src/system/commands/mod.rs

@@ -786,7 +786,10 @@ impl<'w, 's> Commands<'w, 's> {
     /// [`CachedSystemId`](crate::system::CachedSystemId) resource.
     ///
     /// See [`World::register_system_cached`] for more information.
-    pub fn run_system_cached<M: 'static, S: IntoSystem<(), (), M> + 'static>(&mut self, system: S) {
+    pub fn run_system_cached<M: 'static, S: IntoSystem<(), (), M> + Send + 'static>(
+        &mut self,
+        system: S,
+    ) {
         self.run_system_cached_with(system, ());
     }
 
@@ -798,7 +801,7 @@ impl<'w, 's> Commands<'w, 's> {
     where
         I: SystemInput<Inner<'static>: Send> + Send + 'static,
         M: 'static,
-        S: IntoSystem<I, (), M> + 'static,
+        S: IntoSystem<I, (), M> + Send + 'static,
     {
         self.queue(RunSystemCachedWith::new(system, input));
     }

crates/bevy_ecs/src/system/mod.rs

@@ -117,7 +117,7 @@ mod system_name;
 mod system_param;
 mod system_registry;
 
-use std::{any::TypeId, borrow::Cow};
+use std::any::TypeId;
 
 pub use adapter_system::*;
 pub use builder::*;
@@ -168,16 +168,13 @@ pub trait IntoSystem<In: SystemInput, Out, Marker>: Sized {
     ///
     /// The second system must have [`In<T>`](crate::system::In) as its first parameter,
     /// where `T` is the return type of the first system.
-    fn pipe<B, BIn, BOut, MarkerB>(self, system: B) -> PipeSystem<Self::System, B::System>
+    fn pipe<B, BIn, BOut, MarkerB>(self, system: B) -> IntoPipeSystem<Self, B>
     where
         Out: 'static,
         B: IntoSystem<BIn, BOut, MarkerB>,
         for<'a> BIn: SystemInput<Inner<'a> = Out>,
     {
-        let system_a = IntoSystem::into_system(self);
-        let system_b = IntoSystem::into_system(system);
-        let name = format!("Pipe({}, {})", system_a.name(), system_b.name());
-        PipeSystem::new(system_a, system_b, Cow::Owned(name))
+        IntoPipeSystem::new(self, system)
     }
 
     /// Pass the output of this system into the passed function `f`, creating a new system that
@@ -199,13 +196,11 @@ pub trait IntoSystem<In: SystemInput, Out, Marker>: Sized {
     ///     # Err(())
     /// }
     /// ```
-    fn map<T, F>(self, f: F) -> AdapterSystem<F, Self::System>
+    fn map<T, F>(self, f: F) -> IntoAdapterSystem<F, Self>
     where
         F: Send + Sync + 'static + FnMut(Out) -> T,
     {
-        let system = Self::into_system(self);
-        let name = system.name();
-        AdapterSystem::new(f, system, name)
+        IntoAdapterSystem::new(f, self)
     }
 
     /// Get the [`TypeId`] of the [`System`] produced after calling [`into_system`](`IntoSystem::into_system`).
@@ -1680,7 +1675,7 @@ mod tests {
 
         let mut world = World::new();
         world.init_resource::<Flag>();
-        let mut sys = first.pipe(second);
+        let mut sys = IntoSystem::into_system(first.pipe(second));
         sys.initialize(&mut world);
 
         sys.run(default(), &mut world);

crates/bevy_ecs/src/system/system_registry.rs

@@ -1,8 +1,10 @@
+use std::marker::PhantomData;
+
 use crate::{
     bundle::Bundle,
     change_detection::Mut,
     entity::Entity,
-    system::{input::SystemInput, BoxedSystem, IntoSystem, System, SystemIn},
+    system::{input::SystemInput, BoxedSystem, IntoSystem, System},
     world::{Command, World},
     {self as bevy_ecs},
 };
@@ -48,7 +50,7 @@ impl<I, O> RemovedSystem<I, O> {
 /// and are created via [`World::register_system`].
 pub struct SystemId<I: SystemInput = (), O = ()> {
     pub(crate) entity: Entity,
-    pub(crate) marker: std::marker::PhantomData<fn(I) -> O>,
+    pub(crate) marker: PhantomData<fn(I) -> O>,
 }
 
 impl<I: SystemInput, O> SystemId<I, O> {
@@ -69,7 +71,7 @@ impl<I: SystemInput, O> SystemId<I, O> {
     pub fn from_entity(entity: Entity) -> Self {
         Self {
             entity,
-            marker: std::marker::PhantomData,
+            marker: PhantomData,
         }
     }
 }
@@ -536,28 +538,38 @@ where
 /// [`Commands`](crate::system::Commands).
 ///
 /// See [`World::register_system_cached`] for more information.
-pub struct RunSystemCachedWith<S: System<Out = ()>> {
+pub struct RunSystemCachedWith<S, I, O, M>
+where
+    I: SystemInput,
+    S: IntoSystem<I, O, M>,
+{
     system: S,
-    input: SystemIn<'static, S>,
+    input: I::Inner<'static>,
+    _phantom: PhantomData<(fn() -> O, fn() -> M)>,
 }
 
-impl<S: System<Out = ()>> RunSystemCachedWith<S> {
+impl<S, I, O, M> RunSystemCachedWith<S, I, O, M>
+where
+    I: SystemInput,
+    S: IntoSystem<I, O, M>,
+{
     /// Creates a new [`Command`] struct, which can be added to
     /// [`Commands`](crate::system::Commands).
-    pub fn new<M>(
-        system: impl IntoSystem<S::In, (), M, System = S>,
-        input: SystemIn<'static, S>,
-    ) -> Self {
+    pub fn new(system: S, input: I::Inner<'static>) -> Self {
         Self {
-            system: IntoSystem::into_system(system),
+            system,
             input,
+            _phantom: PhantomData,
         }
     }
 }
 
-impl<S: System<Out = ()>> Command for RunSystemCachedWith<S>
+impl<S, I, O, M> Command for RunSystemCachedWith<S, I, O, M>
 where
-    S::In: SystemInput<Inner<'static>: Send>,
+    I: SystemInput<Inner<'static>: Send> + Send + 'static,
+    O: Send + 'static,
+    S: IntoSystem<I, O, M> + Send + 'static,
+    M: 'static,
 {
     fn apply(self, world: &mut World) {
         let _ = world.run_system_cached_with(self.system, self.input);
@@ -824,6 +836,40 @@ mod tests {
         assert!(matches!(output, Ok(x) if x == four()));
     }
 
+    #[test]
+    fn cached_system_commands() {
+        fn sys(mut counter: ResMut<Counter>) {
+            counter.0 = 1;
+        }
+
+        let mut world = World::new();
+        world.insert_resource(Counter(0));
+
+        world.commands().run_system_cached(sys);
+        world.flush_commands();
+
+        assert_eq!(world.resource::<Counter>().0, 1);
+    }
+
+    #[test]
+    fn cached_system_adapters() {
+        fn four() -> i32 {
+            4
+        }
+
+        fn double(In(i): In<i32>) -> i32 {
+            i * 2
+        }
+
+        let mut world = World::new();
+
+        let output = world.run_system_cached(four.pipe(double));
+        assert!(matches!(output, Ok(8)));
+
+        let output = world.run_system_cached(four.map(|i| i * 2));
+        assert!(matches!(output, Ok(8)));
+    }
+
     #[test]
     fn system_with_input_ref() {
         fn with_ref(InRef(input): InRef<u8>, mut counter: ResMut<Counter>) {