Fix system param validation for piped systems

crates/bevy_ecs/src/schedule/executor/mod.rs

@@ -315,7 +315,7 @@ mod __rust_begin_short_backtrace {
 #[cfg(test)]
 mod tests {
     use crate::{
-        prelude::{Component, Resource, Schedule},
+        prelude::{Component, In, IntoSystem, Resource, Schedule},
         schedule::ExecutorKind,
         system::{Populated, Res, ResMut, Single},
         world::World,
@@ -336,6 +336,9 @@ mod tests {
         single_ran: bool,
     }
 
+    #[derive(Resource, Default)]
+    struct Counter(u8);
+
     fn set_single_state(mut _single: Single<&TestComponent>, mut state: ResMut<TestState>) {
         state.single_ran = true;
     }
@@ -408,4 +411,162 @@ mod tests {
         schedule.add_systems(look_for_missing_resource);
         schedule.run(&mut world);
     }
+
+    #[test]
+    fn piped_systems_first_system_skipped() {
+        // This system should be skipped when run due to no matching entity
+        fn pipe_out(_single: Single<&TestComponent>) -> u8 {
+            42
+        }
+
+        fn pipe_in(_input: In<u8>, mut counter: ResMut<Counter>) {
+            counter.0 += 1;
+        }
+
+        let mut world = World::new();
+        world.init_resource::<Counter>();
+        let mut schedule = Schedule::default();
+
+        schedule.add_systems(pipe_out.pipe(pipe_in));
+        schedule.run(&mut world);
+
+        let counter = world.resource::<Counter>();
+        assert_eq!(counter.0, 0);
+    }
+
+    #[test]
+    fn piped_system_second_system_skipped() {
+        fn pipe_out(mut counter: ResMut<Counter>) -> u8 {
+            counter.0 += 1;
+            42
+        }
+
+        // This system should be skipped when run due to no matching entity
+        fn pipe_in(_input: In<u8>, _single: Single<&TestComponent>) {}
+
+        let mut world = World::new();
+        world.init_resource::<Counter>();
+        let mut schedule = Schedule::default();
+
+        schedule.add_systems(pipe_out.pipe(pipe_in));
+        schedule.run(&mut world);
+        let counter = world.resource::<Counter>();
+        assert_eq!(counter.0, 0);
+    }
+
+    #[test]
+    #[should_panic]
+    fn piped_system_first_system_panics() {
+        // This system should panic when run because the resource is missing
+        fn pipe_out(_res: Res<TestState>) -> u8 {
+            42
+        }
+
+        fn pipe_in(_input: In<u8>) {}
+
+        let mut world = World::new();
+        let mut schedule = Schedule::default();
+
+        schedule.add_systems(pipe_out.pipe(pipe_in));
+        schedule.run(&mut world);
+    }
+
+    #[test]
+    #[should_panic]
+    fn piped_system_second_system_panics() {
+        fn pipe_out() -> u8 {
+            42
+        }
+
+        // This system should panic when run because the resource is missing
+        fn pipe_in(_input: In<u8>, _res: Res<TestState>) {}
+
+        let mut world = World::new();
+        let mut schedule = Schedule::default();
+
+        schedule.add_systems(pipe_out.pipe(pipe_in));
+        schedule.run(&mut world);
+    }
+
+    // This test runs without panicking because we've
+    // decided to use early-out behavior for piped systems
+    #[test]
+    fn piped_system_skip_and_panic() {
+        // This system should be skipped when run due to no matching entity
+        fn pipe_out(_single: Single<&TestComponent>) -> u8 {
+            42
+        }
+
+        // This system should panic when run because the resource is missing
+        fn pipe_in(_input: In<u8>, _res: Res<TestState>) {}
+
+        let mut world = World::new();
+        let mut schedule = Schedule::default();
+
+        schedule.add_systems(pipe_out.pipe(pipe_in));
+        schedule.run(&mut world);
+    }
+
+    #[test]
+    #[should_panic]
+    fn piped_system_panic_and_skip() {
+        // This system should panic when run because the resource is missing
+
+        fn pipe_out(_res: Res<TestState>) -> u8 {
+            42
+        }
+
+        // This system should be skipped when run due to no matching entity
+        fn pipe_in(_input: In<u8>, _single: Single<&TestComponent>) {}
+
+        let mut world = World::new();
+        let mut schedule = Schedule::default();
+
+        schedule.add_systems(pipe_out.pipe(pipe_in));
+        schedule.run(&mut world);
+    }
+
+    #[test]
+    #[should_panic]
+    fn piped_system_panic_and_panic() {
+        // This system should panic when run because the resource is missing
+
+        fn pipe_out(_res: Res<TestState>) -> u8 {
+            42
+        }
+
+        // This system should panic when run because the resource is missing
+        fn pipe_in(_input: In<u8>, _res: Res<TestState>) {}
+
+        let mut world = World::new();
+        let mut schedule = Schedule::default();
+
+        schedule.add_systems(pipe_out.pipe(pipe_in));
+        schedule.run(&mut world);
+    }
+
+    #[test]
+    fn piped_system_skip_and_skip() {
+        // This system should be skipped when run due to no matching entity
+
+        fn pipe_out(_single: Single<&TestComponent>, mut counter: ResMut<Counter>) -> u8 {
+            counter.0 += 1;
+            42
+        }
+
+        // This system should be skipped when run due to no matching entity
+        fn pipe_in(_input: In<u8>, _single: Single<&TestComponent>, mut counter: ResMut<Counter>) {
+            counter.0 += 1;
+        }
+
+        let mut world = World::new();
+        world.init_resource::<Counter>();
+        let mut schedule = Schedule::default();
+
+        schedule.add_systems(pipe_out.pipe(pipe_in));
+        schedule.run(&mut world);
+
+        let counter = world.resource::<Counter>();
+        assert_eq!(counter.0, 0);
+    }
 }

crates/bevy_ecs/src/system/combinator.rs

@@ -417,17 +417,25 @@ where
         self.b.queue_deferred(world);
     }
 
+    /// This method uses "early out" logic: if the first system fails validation,
+    /// the second system is not validated.
+    ///
+    /// Because the system validation is performed upfront, this can lead to situations
+    /// where later systems pass validation, but fail at runtime due to changes made earlier
+    /// in the piped systems.
+    // TODO: ensure that systems are only validated just before they are run.
+    // Fixing this will require fundamentally rethinking how piped systems work:
+    // they're currently treated as a single system from the perspective of the scheduler.
     unsafe fn validate_param_unsafe(
         &mut self,
         world: UnsafeWorldCell,
     ) -> Result<(), SystemParamValidationError> {
-        // SAFETY: Delegate to other `System` implementations.
-        unsafe { self.a.validate_param_unsafe(world) }
-    }
+        // SAFETY: Delegate to the `System` implementation for `a`.
+        unsafe { self.a.validate_param_unsafe(world) }?;
+
+        // SAFETY: Delegate to the `System` implementation for `b`.
+        unsafe { self.b.validate_param_unsafe(world) }?;
 
-    fn validate_param(&mut self, world: &World) -> Result<(), SystemParamValidationError> {
-        self.a.validate_param(world)?;
-        self.b.validate_param(world)?;
         Ok(())
     }
 
@@ -477,3 +485,27 @@ where
     for<'a> B::In: SystemInput<Inner<'a> = A::Out>,
 {
 }
+
+#[cfg(test)]
+mod tests {
+
+    #[test]
+    fn exclusive_system_piping_is_possible() {
+        use crate::prelude::*;
+
+        fn my_exclusive_system(_world: &mut World) -> u32 {
+            1
+        }
+
+        fn out_pipe(input: In<u32>) {
+            assert!(input.0 == 1);
+        }
+
+        let mut world = World::new();
+
+        let mut schedule = Schedule::default();
+        schedule.add_systems(my_exclusive_system.pipe(out_pipe));
+
+        schedule.run(&mut world);
+    }
+}