Ensure that events are updated even when using a bare-bones Bevy App (#13808)

# Objective

As discovered in
https://github.com/Leafwing-Studios/leafwing-input-manager/issues/538,
there appears to be some real weirdness going on in how event updates
are processed between Bevy 0.13 and Bevy 0.14.

To identify the cause and prevent regression, I've added tests to
validate the intended behavior.
My initial suspicion was that this would be fixed by
https://github.com/bevyengine/bevy/pull/13762, but that doesn't seem to
be the case.

Instead, events appear to never be updated at all when using `bevy_app`
by itself. This is part of the problem resolved by
https://github.com/bevyengine/bevy/pull/11528, and introduced by
https://github.com/bevyengine/bevy/pull/10077.

After some investigation, it appears that `signal_event_update_system`
is never added using a bare-bones `App`, and so event updates are always
skipped.

This can be worked around by adding your own copy to a
later-in-the-frame schedule, but that's not a very good fix.

## Solution

Ensure that if we're not using a `FixedUpdate` schedule, events are
always updated every frame.

To do this, I've modified the logic of `event_update_condition` and
`event_update_system` to clearly and correctly differentiate between the
two cases: where we're waiting for a "you should update now" signal and
where we simply don't care.

To encode this, I've added the `ShouldUpdateEvents` enum, replacing a
simple `bool` in `EventRegistry`'s `needs_update` field.

Now, both tests pass as expected, without having to manually add a
system!

## Testing

I've written two parallel unit tests to cover the intended behavior:

1. Test that `iter_current_update_events` works as expected in
`bevy_ecs`.
2. Test that `iter_current_update_events` works as expected in
`bevy_app`

I've also added a test to verify that event updating works correctly in
the presence of a fixed main schedule, and a second test to verify that
fixed updating works at all to help future authors narrow down failures.

## Outstanding

- [x] figure out why the `bevy_app` version of this test fails but the
`bevy_ecs` version does not
- [x] figure out why `EventRegistry::run_updates` isn't working properly
- [x] figure out why `EventRegistry::run_updates` is never getting
called
- [x] figure out why `event_update_condition` is always returning false
- [x] figure out why `EventRegistry::needs_update` is always false
- [x] verify that the problem is a missing `signal_events_update_system`

---------

Co-authored-by: Mike <mike.hsu@gmail.com>

crates/bevy_app/src/app.rs

@@ -925,7 +925,7 @@ mod tests {
         change_detection::{DetectChanges, ResMut},
         component::Component,
         entity::Entity,
-        event::EventWriter,
+        event::{Event, EventWriter, Events},
         query::With,
         removal_detection::RemovedComponents,
         schedule::{IntoSystemConfigs, ScheduleLabel},
@@ -1274,4 +1274,41 @@ mod tests {
             .init_non_send_resource::<NonSendTestResource>()
             .init_resource::<TestResource>();
     }
+
+    #[test]
+    fn events_should_be_updated_once_per_update() {
+        #[derive(Event, Clone)]
+        struct TestEvent;
+
+        let mut app = App::new();
+        app.add_event::<TestEvent>();
+
+        // Starts empty
+        let test_events = app.world().resource::<Events<TestEvent>>();
+        assert_eq!(test_events.len(), 0);
+        assert_eq!(test_events.iter_current_update_events().count(), 0);
+        app.update();
+
+        // Sending one event
+        app.world_mut().send_event(TestEvent);
+
+        let test_events = app.world().resource::<Events<TestEvent>>();
+        assert_eq!(test_events.len(), 1);
+        assert_eq!(test_events.iter_current_update_events().count(), 1);
+        app.update();
+
+        // Sending two events on the next frame
+        app.world_mut().send_event(TestEvent);
+        app.world_mut().send_event(TestEvent);
+
+        let test_events = app.world().resource::<Events<TestEvent>>();
+        assert_eq!(test_events.len(), 3); // Events are double-buffered, so we see 1 + 2 = 3
+        assert_eq!(test_events.iter_current_update_events().count(), 2);
+        app.update();
+
+        // Sending zero events
+        let test_events = app.world().resource::<Events<TestEvent>>();
+        assert_eq!(test_events.len(), 2); // Events are double-buffered, so we see 2 + 0 = 2
+        assert_eq!(test_events.iter_current_update_events().count(), 0);
+    }
 }

crates/bevy_ecs/src/event/collections.rs

@@ -366,3 +366,41 @@ impl<E: Event> ExactSizeIterator for SendBatchIds<E> {
         self.event_count.saturating_sub(self.last_count)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::{self as bevy_ecs, event::Events};
+    use bevy_ecs_macros::Event;
+
+    #[test]
+    fn iter_current_update_events_iterates_over_current_events() {
+        #[derive(Event, Clone)]
+        struct TestEvent;
+
+        let mut test_events = Events::<TestEvent>::default();
+
+        // Starting empty
+        assert_eq!(test_events.len(), 0);
+        assert_eq!(test_events.iter_current_update_events().count(), 0);
+        test_events.update();
+
+        // Sending one event
+        test_events.send(TestEvent);
+
+        assert_eq!(test_events.len(), 1);
+        assert_eq!(test_events.iter_current_update_events().count(), 1);
+        test_events.update();
+
+        // Sending two events on the next frame
+        test_events.send(TestEvent);
+        test_events.send(TestEvent);
+
+        assert_eq!(test_events.len(), 3); // Events are double-buffered, so we see 1 + 2 = 3
+        assert_eq!(test_events.iter_current_update_events().count(), 2);
+        test_events.update();
+
+        // Sending zero events
+        assert_eq!(test_events.len(), 2); // Events are double-buffered, so we see 2 + 0 = 2
+        assert_eq!(test_events.iter_current_update_events().count(), 0);
+    }
+}

crates/bevy_ecs/src/event/mod.rs

@@ -13,7 +13,7 @@ pub use bevy_ecs_macros::Event;
 pub use collections::{Events, SendBatchIds};
 pub use iterators::{EventIterator, EventIteratorWithId, EventParIter};
 pub use reader::{EventReader, ManualEventReader};
-pub use registry::EventRegistry;
+pub use registry::{EventRegistry, ShouldUpdateEvents};
 pub use update::{
     event_update_condition, event_update_system, signal_event_update_system, EventUpdates,
 };

crates/bevy_ecs/src/event/registry.rs

@@ -17,14 +17,29 @@ struct RegisteredEvent {
     update: unsafe fn(MutUntyped),
 }
 
-/// A registry of all of the [`Events`] in the [`World`], used by [`event_update_system`]
+/// A registry of all of the [`Events`] in the [`World`], used by [`event_update_system`](crate::event::update::event_update_system)
 /// to update all of the events.
 #[derive(Resource, Default)]
 pub struct EventRegistry {
-    pub(super) needs_update: bool,
+    /// Should the events be updated?
+    ///
+    /// This field is generally automatically updated by the [`signal_event_update_system`](crate::event::update::signal_event_update_system).
+    pub should_update: ShouldUpdateEvents,
     event_updates: Vec<RegisteredEvent>,
 }
 
+/// Controls whether or not the events in an [`EventRegistry`] should be updated.
+#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ShouldUpdateEvents {
+    /// Without any fixed timestep, events should always be updated each frame.
+    #[default]
+    Always,
+    /// We need to wait until at least one pass of the fixed update schedules to update the events.
+    Waiting,
+    /// At least one pass of the fixed update schedules has occurred, and the events are ready to be updated.
+    Ready,
+}
+
 impl EventRegistry {
     /// Registers an event type to be updated.
     pub fn register_event<T: Event>(world: &mut World) {

crates/bevy_ecs/src/event/update.rs

@@ -10,14 +10,19 @@ use bevy_ecs_macros::SystemSet;
 #[cfg(feature = "bevy_reflect")]
 use std::hash::Hash;
 
+use super::registry::ShouldUpdateEvents;
+
 #[doc(hidden)]
 #[derive(SystemSet, Clone, Debug, PartialEq, Eq, Hash)]
 pub struct EventUpdates;
 
 /// Signals the [`event_update_system`] to run after `FixedUpdate` systems.
+///
+/// This will change the behavior of the [`EventRegistry`] to only run after a fixed update cycle has passed.
+/// Normally, this will simply run every frame.
 pub fn signal_event_update_system(signal: Option<ResMut<EventRegistry>>) {
     if let Some(mut registry) = signal {
-        registry.needs_update = true;
+        registry.should_update = ShouldUpdateEvents::Ready;
     }
 }
 
@@ -26,16 +31,32 @@ pub fn event_update_system(world: &mut World, mut last_change_tick: Local<Tick>)
     if world.contains_resource::<EventRegistry>() {
         world.resource_scope(|world, mut registry: Mut<EventRegistry>| {
             registry.run_updates(world, *last_change_tick);
-            // Disable the system until signal_event_update_system runs again.
-            registry.needs_update = false;
+
+            registry.should_update = match registry.should_update {
+                // If we're always updating, keep doing so.
+                ShouldUpdateEvents::Always => ShouldUpdateEvents::Always,
+                // Disable the system until signal_event_update_system runs again.
+                ShouldUpdateEvents::Waiting | ShouldUpdateEvents::Ready => {
+                    ShouldUpdateEvents::Waiting
+                }
+            };
         });
     }
     *last_change_tick = world.change_tick();
 }
 
 /// A run condition for [`event_update_system`].
-pub fn event_update_condition(signal: Option<Res<EventRegistry>>) -> bool {
-    // If we haven't got a signal to update the events, but we *could* get such a signal
-    // return early and update the events later.
-    signal.map_or(false, |signal| signal.needs_update)
+///
+/// If [`signal_event_update_system`] has been run at least once,
+/// we will wait for it to be run again before updating the events.
+///
+/// Otherwise, we will always update the events.
+pub fn event_update_condition(maybe_signal: Option<Res<EventRegistry>>) -> bool {
+    match maybe_signal {
+        Some(signal) => match signal.should_update {
+            ShouldUpdateEvents::Always | ShouldUpdateEvents::Ready => true,
+            ShouldUpdateEvents::Waiting => false,
+        },
+        None => true,
+    }
 }

crates/bevy_time/src/lib.rs

@@ -30,7 +30,7 @@ pub mod prelude {
 }
 
 use bevy_app::{prelude::*, RunFixedMainLoop};
-use bevy_ecs::event::signal_event_update_system;
+use bevy_ecs::event::{signal_event_update_system, EventRegistry, ShouldUpdateEvents};
 use bevy_ecs::prelude::*;
 use bevy_utils::{tracing::warn, Duration, Instant};
 pub use crossbeam_channel::TrySendError;
@@ -65,8 +65,11 @@ impl Plugin for TimePlugin {
         app.add_systems(First, time_system.in_set(TimeSystem))
             .add_systems(RunFixedMainLoop, run_fixed_main_schedule);
 
-        // ensure the events are not dropped until `FixedMain` systems can observe them
+        // Ensure the events are not dropped until `FixedMain` systems can observe them
         app.add_systems(FixedPostUpdate, signal_event_update_system);
+        let mut event_registry = app.world_mut().resource_mut::<EventRegistry>();
+        // We need to start in a waiting state so that the events are not updated until the first fixed update
+        event_registry.should_update = ShouldUpdateEvents::Waiting;
     }
 }
 
@@ -141,9 +144,13 @@ fn time_system(
 
 #[cfg(test)]
 mod tests {
-    use crate::{Fixed, Time, TimePlugin, TimeUpdateStrategy};
-    use bevy_app::{App, Startup, Update};
-    use bevy_ecs::event::{Event, EventReader, EventWriter};
+    use crate::{Fixed, Time, TimePlugin, TimeUpdateStrategy, Virtual};
+    use bevy_app::{App, FixedUpdate, Startup, Update};
+    use bevy_ecs::{
+        event::{Event, EventReader, EventRegistry, EventWriter, Events, ShouldUpdateEvents},
+        system::{Local, Res, ResMut, Resource},
+    };
+    use bevy_utils::Duration;
     use std::error::Error;
 
     #[derive(Event)]
@@ -159,6 +166,91 @@ mod tests {
         }
     }
 
+    #[derive(Event)]
+    struct DummyEvent;
+
+    #[derive(Resource, Default)]
+    struct FixedUpdateCounter(u8);
+
+    fn count_fixed_updates(mut counter: ResMut<FixedUpdateCounter>) {
+        counter.0 += 1;
+    }
+
+    fn report_time(
+        mut frame_count: Local<u64>,
+        virtual_time: Res<Time<Virtual>>,
+        fixed_time: Res<Time<Fixed>>,
+    ) {
+        println!(
+            "Virtual time on frame {}: {:?}",
+            *frame_count,
+            virtual_time.elapsed()
+        );
+        println!(
+            "Fixed time on frame {}: {:?}",
+            *frame_count,
+            fixed_time.elapsed()
+        );
+
+        *frame_count += 1;
+    }
+
+    #[test]
+    fn fixed_main_schedule_should_run_with_time_plugin_enabled() {
+        // Set the time step to just over half the fixed update timestep
+        // This way, it will have not accumulated enough time to run the fixed update after one update
+        // But will definitely have enough time after two updates
+        let fixed_update_timestep = Time::<Fixed>::default().timestep();
+        let time_step = fixed_update_timestep / 2 + Duration::from_millis(1);
+
+        let mut app = App::new();
+        app.add_plugins(TimePlugin)
+            .add_systems(FixedUpdate, count_fixed_updates)
+            .add_systems(Update, report_time)
+            .init_resource::<FixedUpdateCounter>()
+            .insert_resource(TimeUpdateStrategy::ManualDuration(time_step));
+
+        // Frame 0
+        // Fixed update should not have run yet
+        app.update();
+
+        assert!(Duration::ZERO < fixed_update_timestep);
+        let counter = app.world().resource::<FixedUpdateCounter>();
+        assert_eq!(counter.0, 0, "Fixed update should not have run yet");
+
+        // Frame 1
+        // Fixed update should not have run yet
+        app.update();
+
+        assert!(time_step < fixed_update_timestep);
+        let counter = app.world().resource::<FixedUpdateCounter>();
+        assert_eq!(counter.0, 0, "Fixed update should not have run yet");
+
+        // Frame 2
+        // Fixed update should have run now
+        app.update();
+
+        assert!(2 * time_step > fixed_update_timestep);
+        let counter = app.world().resource::<FixedUpdateCounter>();
+        assert_eq!(counter.0, 1, "Fixed update should have run once");
+
+        // Frame 3
+        // Fixed update should have run exactly once still
+        app.update();
+
+        assert!(3 * time_step < 2 * fixed_update_timestep);
+        let counter = app.world().resource::<FixedUpdateCounter>();
+        assert_eq!(counter.0, 1, "Fixed update should have run once");
+
+        // Frame 4
+        // Fixed update should have run twice now
+        app.update();
+
+        assert!(4 * time_step > 2 * fixed_update_timestep);
+        let counter = app.world().resource::<FixedUpdateCounter>();
+        assert_eq!(counter.0, 2, "Fixed update should have run twice");
+    }
+
     #[test]
     fn events_get_dropped_regression_test_11528() -> Result<(), impl Error> {
         let (tx1, rx1) = std::sync::mpsc::channel();
@@ -199,4 +291,75 @@ mod tests {
         // Check event type 2 has been dropped
         rx2.try_recv()
     }
+
+    #[test]
+    fn event_update_should_wait_for_fixed_main() {
+        // Set the time step to just over half the fixed update timestep
+        // This way, it will have not accumulated enough time to run the fixed update after one update
+        // But will definitely have enough time after two updates
+        let fixed_update_timestep = Time::<Fixed>::default().timestep();
+        let time_step = fixed_update_timestep / 2 + Duration::from_millis(1);
+
+        fn send_event(mut events: ResMut<Events<DummyEvent>>) {
+            events.send(DummyEvent);
+        }
+
+        let mut app = App::new();
+        app.add_plugins(TimePlugin)
+            .add_event::<DummyEvent>()
+            .init_resource::<FixedUpdateCounter>()
+            .add_systems(Startup, send_event)
+            .add_systems(FixedUpdate, count_fixed_updates)
+            .insert_resource(TimeUpdateStrategy::ManualDuration(time_step));
+
+        for frame in 0..10 {
+            app.update();
+            let fixed_updates_seen = app.world().resource::<FixedUpdateCounter>().0;
+            let events = app.world().resource::<Events<DummyEvent>>();
+            let n_total_events = events.len();
+            let n_current_events = events.iter_current_update_events().count();
+            let event_registry = app.world().resource::<EventRegistry>();
+            let should_update = event_registry.should_update;
+
+            println!("Frame {frame}, {fixed_updates_seen} fixed updates seen. Should update: {should_update:?}");
+            println!("Total events: {n_total_events} | Current events: {n_current_events}",);
+
+            match frame {
+                0 | 1 => {
+                    assert_eq!(fixed_updates_seen, 0);
+                    assert_eq!(n_total_events, 1);
+                    assert_eq!(n_current_events, 1);
+                    assert_eq!(should_update, ShouldUpdateEvents::Waiting);
+                }
+                2 => {
+                    assert_eq!(fixed_updates_seen, 1); // Time to trigger event updates
+                    assert_eq!(n_total_events, 1);
+                    assert_eq!(n_current_events, 1);
+                    assert_eq!(should_update, ShouldUpdateEvents::Ready); // Prepping first update
+                }
+                3 => {
+                    assert_eq!(fixed_updates_seen, 1);
+                    assert_eq!(n_total_events, 1);
+                    assert_eq!(n_current_events, 0); // First update has occurred
+                    assert_eq!(should_update, ShouldUpdateEvents::Waiting);
+                }
+                4 => {
+                    assert_eq!(fixed_updates_seen, 2); // Time to trigger the second update
+                    assert_eq!(n_total_events, 1);
+                    assert_eq!(n_current_events, 0);
+                    assert_eq!(should_update, ShouldUpdateEvents::Ready); // Prepping second update
+                }
+                5 => {
+                    assert_eq!(fixed_updates_seen, 2);
+                    assert_eq!(n_total_events, 0); // Second update has occurred
+                    assert_eq!(n_current_events, 0);
+                    assert_eq!(should_update, ShouldUpdateEvents::Waiting);
+                }
+                _ => {
+                    assert_eq!(n_total_events, 0); // No more events are sent
+                    assert_eq!(n_current_events, 0);
+                }
+            }
+        }
+    }
 }