fix(inputs): simplify BEI input replication (#1537)

Author: cBournhonesque

crates/inputs/input_bei/src/marker.rs

@@ -1,12 +1,11 @@
 //! Add an [`InputMarker<C>`] component automatically to [`Action`] entities that need it
 
-use crate::setup::NetworkActionOf;
 use bevy_ecs::prelude::*;
 use bevy_ecs::relationship::Relationship;
 use bevy_enhanced_input::prelude::*;
 use bevy_replicon::client::confirm_history::ConfirmHistory;
 use lightyear_connection::client::Client;
-use lightyear_replication::prelude::{Controlled, ControlledBy, HasAuthority};
+use lightyear_replication::prelude::{Controlled, ControlledBy};
 
 /// Marker component that indicates that the entity is actively listening for physical user inputs.
 ///
@@ -40,132 +39,83 @@ fn action_targets_local_client<C: Component>(
 
 /// Propagate the InputMarker component from the Context entity to the Action entities
 /// whenever an InputMarker is added to a Context entity.
-/// Skip replicated action entities (those received from remote clients).
+///
+/// `InputMarker<C>` on the context is the explicit local-input signal, so
+/// confirmed prespawned owner actions should still receive it. Remote
+/// rebroadcasted actions should be attached to contexts without this marker.
 pub(crate) fn propagate_input_marker<C: Component>(
     trigger: On<Add, InputMarker<C>>,
     actions: Query<&Actions<C>>,
-    confirm: Query<(), With<ConfirmHistory>>,
     mut commands: Commands,
 ) {
     if let Ok(actions) = actions.get(trigger.entity) {
         actions.iter().for_each(|action| {
-            if confirm.get(action).is_ok() {
-                return;
-            }
             commands.entity(action).insert(InputMarker::<C>::default());
         });
     }
 }
 
 /// When an Action entity is added to a Context entity that has an InputMarker,
 /// add the InputMarker to the Action entity as well.
-/// Skip replicated entities — those are received from remote clients and should not
-/// be marked as local input sources.
 pub(crate) fn add_input_marker_from_parent<C: Component>(
     trigger: On<Add, ActionOf<C>>,
-    action_of: Query<&ActionOf<C>, Without<ConfirmHistory>>,
+    action_of: Query<&ActionOf<C>>,
     context: Query<(), With<InputMarker<C>>>,
     mut commands: Commands,
 ) {
-    if let Ok(action_of) = action_of.get(trigger.entity)
-        && context.get(action_of.get()).is_ok()
-    {
+    let Ok(action_of) = action_of.get(trigger.entity) else {
+        return;
+    };
+    if context.get(action_of.get()).is_ok() {
         commands
             .entity(trigger.entity)
             .insert(InputMarker::<C>::default());
     }
 }
 
-/// If Bindings or ActionMock is added to an Action entity, add the InputMarker to that Action entity.
-/// Only add the marker on locally controlled action entities that already have a network-facing
-/// action mapping. This avoids emitting inputs for entities that the server cannot resolve yet.
+/// If Bindings or ActionMock is added to an Action entity, add the InputMarker
+/// to that Action entity once the action has a network-resolvable identity.
+///
+/// Replicated/prespawned actions become resolvable when [`ConfirmHistory`] is
+/// present, because the client's action entity can then be mapped back to the
+/// server action entity when an input message is sent.
 pub(crate) fn add_input_marker_from_binding<C: Component>(
     trigger: On<Add, (Bindings, ActionMock)>,
-    action: Query<
-        &ActionOf<C>,
-        (
-            With<ActionOf<C>>,
-            With<NetworkActionOf<C>>,
-            Without<ConfirmHistory>,
-        ),
-    >,
+    action: Query<&ActionOf<C>, (With<ConfirmHistory>, Without<InputMarker<C>>)>,
     contexts: Query<Option<&ControlledBy>, With<Controlled>>,
     clients: Query<(), With<Client>>,
     mut commands: Commands,
 ) {
-    if let Ok(action_of) = action.get(trigger.entity)
-        && action_targets_local_client(action_of, &contexts, &clients)
-    {
+    let Ok(action_of) = action.get(trigger.entity) else {
+        return;
+    };
+    if action_targets_local_client(action_of, &contexts, &clients) {
         commands
             .entity(trigger.entity)
             .insert(InputMarker::<C>::default());
     }
 }
 
-/// If authority is granted after the action entity already exists, add the InputMarker
-/// once the entity becomes locally controlled.
-pub(crate) fn add_input_marker_from_authority<C: Component>(
-    trigger: On<Add, HasAuthority>,
-    action: Query<
-        &ActionOf<C>,
-        (
-            With<ActionOf<C>>,
-            With<NetworkActionOf<C>>,
-            Or<(With<Bindings>, With<ActionMock>)>,
-            Without<ConfirmHistory>,
-        ),
-    >,
-    contexts: Query<Option<&ControlledBy>, With<Controlled>>,
-    clients: Query<(), With<Client>>,
-    mut commands: Commands,
-) {
-    if let Ok(action_of) = action.get(trigger.entity)
-        && action_targets_local_client(action_of, &contexts, &clients)
-    {
-        commands
-            .entity(trigger.entity)
-            .insert(InputMarker::<C>::default());
-    }
-}
-
-/// If the network-facing action mapping becomes available after the entity already has bindings,
-/// start treating it as a local input source at that point.
-pub(crate) fn add_input_marker_from_network_action<C: Component>(
-    trigger: On<Add, NetworkActionOf<C>>,
+/// If an existing bound action becomes network-resolvable, add the InputMarker
+/// once it targets the local client.
+pub(crate) fn add_input_marker_when_action_becomes_ready<C: Component>(
+    trigger: On<Add, ConfirmHistory>,
     action: Query<
         &ActionOf<C>,
         (
-            With<ActionOf<C>>,
-            With<NetworkActionOf<C>>,
+            With<ConfirmHistory>,
             Or<(With<Bindings>, With<ActionMock>)>,
-            Without<ConfirmHistory>,
+            Without<InputMarker<C>>,
         ),
     >,
     contexts: Query<Option<&ControlledBy>, With<Controlled>>,
     clients: Query<(), With<Client>>,
     mut commands: Commands,
 ) {
-    if let Ok(action_of) = action.get(trigger.entity)
-        && action_targets_local_client(action_of, &contexts, &clients)
-    {
-        commands
-            .entity(trigger.entity)
-            .insert(InputMarker::<C>::default());
-    }
-}
-
-/// If a prespawned action entity is confirmed but still targets a locally controlled context,
-/// keep using it as a local input source.
-pub(crate) fn add_input_marker_from_confirmed_controlled_action<C: Component>(
-    trigger: On<Add, ConfirmHistory>,
-    action: Query<&ActionOf<C>, (With<ConfirmHistory>, Or<(With<Bindings>, With<ActionMock>)>)>,
-    contexts: Query<Option<&ControlledBy>, With<Controlled>>,
-    clients: Query<(), With<Client>>,
-    mut commands: Commands,
-) {
-    if let Ok(action_of) = action.get(trigger.entity)
-        && action_targets_local_client(action_of, &contexts, &clients)
-    {
+    let Ok(action_of) = action.get(trigger.entity) else {
+        return;
+    };
+    if action_targets_local_client(action_of, &contexts, &clients) {
         commands
             .entity(trigger.entity)
             .insert(InputMarker::<C>::default());

crates/inputs/input_bei/src/plugin.rs

@@ -1,9 +1,16 @@
 #[cfg(any(feature = "client", feature = "server"))]
 use crate::input_message::{BEIBuffer, BEIStateSequence};
 
-use crate::setup::InputRegistryPlugin;
-use bevy_app::{PreUpdate, prelude::*};
+#[cfg(feature = "client")]
+use crate::setup::resolve_pending_action_of;
+#[cfg(any(feature = "client", feature = "server"))]
+use crate::setup::{
+    InputRegistryPlugin, deserialize_action_of, remove_action_of, serialize_action_of,
+    write_action_of,
+};
+use bevy_app::prelude::*;
 use bevy_ecs::prelude::*;
+#[cfg(any(feature = "client", feature = "server"))]
 use bevy_ecs::schedule::IntoScheduleConfigs;
 #[cfg(all(feature = "client", feature = "server"))]
 use bevy_ecs::schedule::common_conditions::not;
@@ -12,19 +19,16 @@ use bevy_enhanced_input::EnhancedInputSystems;
 #[cfg(feature = "client")]
 use bevy_enhanced_input::action::TriggerState;
 use bevy_enhanced_input::context::InputContextAppExt;
-#[cfg(any(feature = "client", feature = "server"))]
 use bevy_enhanced_input::prelude::ActionOf;
 use bevy_reflect::TypePath;
-use bevy_replicon::prelude::{AppRuleExt, ReplicationMode, RuleFns};
+use bevy_replicon::prelude::{AppMarkerExt, AppRuleExt, ReplicationMode, RuleFns};
 use bevy_replicon::shared::replication::registry::receive_fns::MutWrite;
 use core::fmt::Debug;
 #[cfg(feature = "client")]
 use lightyear_core::prelude::is_in_rollback;
 #[cfg(feature = "client")]
 use lightyear_inputs::client::InputSystems;
 use lightyear_inputs::config::InputConfig;
-use lightyear_messages::plugin::MessageSystems;
-use lightyear_replication::ReplicationSystems;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 
@@ -45,14 +49,37 @@ use serde::de::DeserializeOwned;
 /// # Action entities
 ///
 /// BEI uses separate "action entities" with [`ActionOf<C>`] to represent
-/// individual actions. These entities need to exist on both client and server.
-/// The recommended approach is to use [`PreSpawned`] so both sides spawn them
-/// independently and match via a deterministic hash — this avoids the need
-/// for client-to-server entity replication.
+/// individual actions. In the server-authoritative flow, spawn those action
+/// entities on the server and replicate them to clients along with the context
+/// entity. The owning client should add local-only [`Bindings`] once its local
+/// controlled context has the replicated [`Action`] entity in its
+/// [`ActionOf<C>`]/`Actions<C>` relationship.
+///
+/// Replicating the action entity is also what lets remote clients receive
+/// rebroadcasted BEI input. Rebroadcasted [`BEIStateSequence`] messages target
+/// action entities, so a remote client needs a corresponding replicated action
+/// entity to resolve the target and buffer the remote player's input state.
 ///
+/// The replicated [`Action`] component is structural: it recreates the typed BEI
+/// action entity on the receiver, but does not carry runtime input state. The
+/// action relationship is replicated directly through [`ActionOf<C>`].
+/// Lightyear uses a custom receive path for [`ActionOf<C>`] so the context
+/// entity is only inserted into Bevy's relationship component once Replicon's
+/// server-to-client entity map already contains it. If the action arrives first,
+/// the relationship is held as a local pending marker and resolved after the
+/// context mapping is available.
+/// Live action state is sent by [`BEIStateSequence`] input messages. The owning
+/// client adds [`InputMarker`] to local action entities, buffers BEI trigger
+/// state/value/time each tick, and sends those snapshots to the server. If input
+/// rebroadcasting is enabled, the server forwards those input messages to other
+/// clients so they can update remote action buffers for prediction.
+///
+/// [`Action`]: bevy_enhanced_input::prelude::Action
+/// [`Bindings`]: bevy_enhanced_input::prelude::Bindings
 /// [`BEIStateSequence`]: crate::input_message::BEIStateSequence
 /// [`ActionOf<C>`]: bevy_enhanced_input::prelude::ActionOf
-/// [`PreSpawned`]: lightyear_replication::prelude::PreSpawned
+/// [`InputMarker`]: crate::marker::InputMarker
+/// [`Replicate`]: lightyear_replication::prelude::Replicate
 pub struct InputPlugin<C> {
     pub config: InputConfig<C>,
 }
@@ -91,37 +118,16 @@ impl<
         app.add_input_context_to::<FixedPreUpdate, C>();
         // we register the context C entity so that it can be replicated from the server to the client
         app.replicate::<C>();
-
-        // We mirror ActionOf<C> into a separate component that stores the authoritative
-        // remote entity. That avoids depending on sender-side entity mapping in replicon's
-        // SerializeCtx.
         app.replicate_with((
-            RuleFns::new(
-                crate::setup::serialize_network_action_of::<C>,
-                crate::setup::deserialize_network_action_of::<C>,
-            ),
-            ReplicationMode::default(),
-        ));
-        app.add_observer(InputRegistryPlugin::mirror_action_of_for_replication::<C>);
-        app.add_observer(InputRegistryPlugin::insert_action_of_from_network::<C>);
-        app.add_systems(
-            PreUpdate,
-            (
-                InputRegistryPlugin::resolve_pending_network_action_of::<C>,
-                InputRegistryPlugin::resolve_pending_action_of::<C>,
-            )
-                .chain()
-                .after(ReplicationSystems::Receive)
-                .before(MessageSystems::Receive),
-        );
-
+            RuleFns::new(serialize_action_of::<C>, deserialize_action_of::<C>),
+            ReplicationMode::Once,
+        ))
+        .set_receive_fns::<ActionOf<C>>(write_action_of::<C>, remove_action_of::<C>);
         #[cfg(feature = "client")]
         {
             use crate::marker::{
-                add_input_marker_from_authority, add_input_marker_from_binding,
-                add_input_marker_from_confirmed_controlled_action,
-                add_input_marker_from_network_action, add_input_marker_from_parent,
-                propagate_input_marker,
+                add_input_marker_from_binding, add_input_marker_from_parent,
+                add_input_marker_when_action_becomes_ready, propagate_input_marker,
             };
             // for rebroadcasting inputs, we insert TriggerState (which inserts the InputBuffer) when ActionOf<C> is added
             // on an entity
@@ -130,9 +136,8 @@ impl<
             app.add_observer(propagate_input_marker::<C>);
             app.add_observer(add_input_marker_from_parent::<C>);
             app.add_observer(add_input_marker_from_binding::<C>);
-            app.add_observer(add_input_marker_from_authority::<C>);
-            app.add_observer(add_input_marker_from_network_action::<C>);
-            app.add_observer(add_input_marker_from_confirmed_controlled_action::<C>);
+            app.add_observer(add_input_marker_when_action_becomes_ready::<C>);
+            app.add_systems(PreUpdate, resolve_pending_action_of::<C>);
 
             if self.config.rebroadcast_inputs {
                 app.add_observer(InputRegistryPlugin::on_rebroadcast_action_received::<C>);
@@ -147,8 +152,6 @@ impl<
                 );
             }
 
-            app.add_observer(InputRegistryPlugin::add_action_of_replicate::<C>);
-
             app.add_plugins(lightyear_inputs::client::ClientInputPlugin::<
                 BEIStateSequence<C>,
             >::new(self.config));