Fix system param validation for piped systems (#18785)

# Objective

- Piped systems are an edge case that we missed when reworking system
parameter validation.
- Fixes #18755.

## Solution

- Validate the parameters for both systems, ~~combining the errors if
both failed validation~~ by simply using an early out.
- ~~Also fix the same bug for combinator systems while we're here.~~

## Testing

I've added a large number of tests checking the behavior under various
permutations. These are separate tests, rather than one mega test
because a) it's easier to track down bugs that way and b) many of these
are `should_panic` tests, which will halt the evaluation of the rest of
the test!

I've also added a test for exclusive systems being pipeable because we
don't have one and I was very surprised that that works!

---------

Co-authored-by: Chris Russell <8494645+chescock@users.noreply.github.com>

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,26 @@ 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.
+    // See https://github.com/bevyengine/bevy/issues/18796
     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 +486,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);
+    }
+}