Add delta-compression using an operation log

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Diff-based replication for components via `replicate_diff`.
 - `ReplicationStorage` resource for storing arbitrary serialization/deserialization state.
 - `SerializeCtx::entity` and `WriteCtx::entity` with the current entity.
 - `AllExcept` filter scope and `VisibilityScope::AllExcept` as a counterpart to `Components`. When a `VisibilityFilter` denies visibility, every component except the listed ones is hidden. Useful for replicating a stripped-down entity (e.g. only its transform and light) to clients outside its full visibility range.

Cargo.toml

@@ -113,6 +113,10 @@ required-features = ["client", "server"]
 name = "despawn"
 required-features = ["client", "server"]
 
+[[test]]
+name = "diff"
+required-features = ["client", "server"]
+
 [[test]]
 name = "fns"
 required-features = ["client"]

src/lib.rs

@@ -750,6 +750,7 @@ pub mod prelude {
             protocol::{ProtocolHash, ProtocolHasher, ProtocolMismatch},
             replication::{
                 Replicated,
+                diff::{Diffable, EntityCommandsDiffExt, EntityDiffExt, diff_index::DiffIndex},
                 receive_markers::AppMarkerExt,
                 registry::rule_fns::RuleFns,
                 rules::{AppRuleExt, component::ReplicationMode},

src/server.rs

@@ -530,7 +530,7 @@ fn collect_removals(
                 }
                 let fns_id_range = serialized.write_cached_fns_id(&mut fns_id_range, fns_id)?;
                 message.add_removal(fns_id_range);
-                entity_ticks.components.remove(component_index);
+                entity_ticks.remove_component(component_index);
             }
         }
     }
@@ -582,7 +582,7 @@ fn collect_removals(
                 }
                 let fns_id_range = serialized.write_fns_id(rule.fns_id)?;
                 message.add_removal(fns_id_range);
-                entity_ticks.components.remove(component_index);
+                entity_ticks.remove_component(component_index);
 
                 Ok(())
             };
@@ -675,7 +675,9 @@ fn collect_changes(
                 let mut ctx = SerializeCtx {
                     entity: entity.id(),
                     component_id,
+                    last_changed: ticks.changed,
                     server_tick: **server_tick,
+                    diff_cursor: None,
                     type_registry: &type_registry,
                     storage: &mut replication_storage,
                 };
@@ -713,11 +715,23 @@ fn collect_changes(
                                     .write_cached_entity(&mut entity_range, entity.id())?;
                                 mutations.add_entity(entity.id(), graph_index, entity_range);
                             }
-                            let component_range = serialized.write_cached_component(
-                                &mut ctx,
-                                &mut component_range,
-                                &mut component,
-                            )?;
+
+                            let diff_cursor = entity_ticks.diff_cursor(component_index);
+                            let component_range = if diff_cursor.is_none() {
+                                // Cache only full component snapshots.
+                                serialized.write_cached_component(
+                                    &mut ctx,
+                                    &mut component_range,
+                                    &mut component,
+                                )?
+                            } else {
+                                ctx.diff_cursor = diff_cursor;
+                                let range = serialized.write_component(&mut ctx, &mut component)?;
+                                if let Some(cursor) = ctx.diff_cursor.take() {
+                                    mutations.add_diff_cursor(component_index, cursor);
+                                }
+                                range
+                            };
                             mutations.add_component(component_range);
                         }
                     } else {
@@ -802,11 +816,7 @@ fn update_ticks(
             entity_ticks.components |= &components;
         }
         Entry::Vacant(entry) => {
-            entry.insert(EntityTicks {
-                server_tick,
-                system_tick,
-                components,
-            });
+            entry.insert(EntityTicks::new(server_tick, system_tick, components));
         }
     }
 }

src/server/replication_messages/mutations.rs

@@ -11,9 +11,9 @@ use crate::{
     shared::{
         backend::channels::ServerChannel,
         replication::{
-            client_ticks::{ClientTicks, MutateInfo},
+            client_ticks::{ClientTicks, DiffCursors, MutateInfo, MutatedEntityInfo},
             mutate_index::MutateIndex,
-            registry::component_mask::ComponentMask,
+            registry::{ComponentIndex, component_mask::ComponentMask},
         },
     },
 };
@@ -70,6 +70,7 @@ impl Mutations {
                 data: Default::default(),
             },
             components: Default::default(),
+            diff_cursors: Default::default(),
         };
 
         match graph_index {
@@ -97,6 +98,19 @@ impl Mutations {
         mutations.ranges.add_data(component);
     }
 
+    /// Adds the diff cursor serialized for component.
+    pub(crate) fn add_diff_cursor(&mut self, component: ComponentIndex, cursor: DiffIndex) {
+        let mutations = self
+            .entity_location
+            .and_then(|location| match location {
+                EntityLocation::Related { index } => self.related[index].last_mut(),
+                EntityLocation::Standalone => self.standalone.last_mut(),
+            })
+            .expect("entity should be written before adding diff cursors");
+
+        mutations.diff_cursors.push((component, cursor));
+    }
+
     /// Removes last added entity from [`Self::add_entity`] and returns it.
     pub(super) fn pop(&mut self) -> Option<EntityMutations> {
         self.entity_location
@@ -187,11 +201,13 @@ impl Mutations {
                 body_size = 0;
             }
 
-            mutate_info.entities.extend(
-                chunk
-                    .iter_mut()
-                    .map(|mutations| (mutations.entity, mem::take(&mut mutations.components))),
-            );
+            mutate_info
+                .entities
+                .extend(chunk.iter_mut().map(|mutations| MutatedEntityInfo {
+                    entity: mutations.entity,
+                    components: mem::take(&mut mutations.components),
+                    diff_cursors: mem::take(&mut mutations.diff_cursors),
+                }));
             chunks_range.end += 1;
             body_size += mutations_size;
         }
@@ -278,6 +294,13 @@ pub(crate) struct EntityMutations {
     ///
     /// Like [`Self::entity`], used for later component acknowledgement.
     pub(super) components: ComponentMask,
+
+    /// Diff cursors represented by the serialized component ranges.
+    ///
+    /// When the client ACKs this mutation message, these cursors become the last
+    /// acknowledged diff indices for their components. Future mutations can then
+    /// include only diffs after these indices.
+    pub(super) diff_cursors: DiffCursors,
 }
 
 #[derive(Clone, Copy)]

src/server/replication_messages/serialized_data.rs

@@ -85,7 +85,7 @@ impl SerializedData {
         Ok(range)
     }
 
-    fn write_component(
+    pub(crate) fn write_component(
         &mut self,
         ctx: &mut SerializeCtx,
         component: &mut ErasedComponent,

src/shared/replication.rs

@@ -1,5 +1,6 @@
 pub mod client_ticks;
 pub mod deferred_entity;
+pub mod diff;
 pub(crate) mod mutate_index;
 pub mod receive_markers;
 pub mod registry;

src/shared/replication/client_ticks.rs

@@ -6,9 +6,19 @@ use bevy::{
     prelude::*,
 };
 use log::{debug, trace};
+use smallvec::SmallVec;
 
 use super::mutate_index::MutateIndex;
-use crate::{prelude::*, shared::replication::registry::component_mask::ComponentMask};
+use crate::{
+    prelude::*,
+    shared::replication::registry::{ComponentIndex, component_mask::ComponentMask},
+};
+
+/// Alias for cursors associated with components.
+///
+/// We use a [`SmallVec`] because entities usually don't have more than a few
+/// components with diff replication enabled.
+pub(crate) type DiffCursors = SmallVec<[(ComponentIndex, DiffIndex); 3]>;
 
 /// Tracks replication ticks for a client.
 #[derive(Component, Default)]
@@ -59,8 +69,8 @@ impl ClientTicks {
             return;
         };
 
-        for (entity, components) in &mutate_info.entities {
-            let Some(entity_ticks) = self.entities.get_mut(entity) else {
+        for info in mutate_info.entities {
+            let Some(entity_ticks) = self.entities.get_mut(&info.entity) else {
                 // We ignore missing entities, since they were probably despawned.
                 continue;
             };
@@ -70,7 +80,13 @@ impl ClientTicks {
             if entity_ticks.server_tick < mutate_info.server_tick {
                 entity_ticks.server_tick = mutate_info.server_tick;
                 entity_ticks.system_tick = mutate_info.system_tick;
-                entity_ticks.components |= components;
+                entity_ticks.components |= &info.components;
+
+                for (component, cursor) in info.diff_cursors {
+                    if entity_ticks.components.contains(component) {
+                        entity_ticks.set_diff_cursor(component, cursor);
+                    }
+                }
             }
         }
         trace!(
@@ -102,12 +118,81 @@ pub(crate) struct EntityTicks {
 
     /// The list of components that were replicated on this tick.
     pub(crate) components: ComponentMask,
+
+    /// Last acknowledged diff cursor for components.
+    ///
+    /// This is separate from [`Self::server_tick`]: the server tick controls change
+    /// detection for the component as a whole, while the cursor controls the
+    /// acknowledged base used to serialize only diffs that the client has not yet
+    /// acknowledged.
+    ///
+    /// Absence means the client has never acknowledged the base value, or
+    /// diff replication is not enabled for it.
+    ///
+    /// Cursors are pruned when the component is removed.
+    diff_cursors: DiffCursors,
+}
+
+impl EntityTicks {
+    pub(crate) fn new(
+        server_tick: RepliconTick,
+        system_tick: Tick,
+        components: ComponentMask,
+    ) -> Self {
+        Self {
+            server_tick,
+            system_tick,
+            components,
+            diff_cursors: Default::default(),
+        }
+    }
+
+    pub(crate) fn diff_cursor(&self, component: ComponentIndex) -> Option<DiffIndex> {
+        self.diff_cursors
+            .iter()
+            .find_map(|&(index, cursor)| (index == component).then_some(cursor))
+    }
+
+    /// Sets the acknowledged diff cursor for a component.
+    ///
+    /// If ACKs arrive out of order, older ACKs must be filtered out by the caller.
+    fn set_diff_cursor(&mut self, component: ComponentIndex, cursor: DiffIndex) {
+        if let Some((_, existing)) = self
+            .diff_cursors
+            .iter_mut()
+            .find(|(index, _)| *index == component)
+        {
+            *existing = cursor;
+        } else {
+            self.diff_cursors.push((component, cursor));
+        }
+    }
+
+    pub(crate) fn remove_component(&mut self, component: ComponentIndex) {
+        self.components.remove(component);
+        // Component removal resets the entity's state for this component, so
+        // its diff cursor becomes stale too.
+        if let Some(index) = self
+            .diff_cursors
+            .iter()
+            .position(|(index, _)| *index == component)
+        {
+            self.diff_cursors.remove(index);
+        }
+    }
 }
 
 /// Information about a mutation message.
 pub(crate) struct MutateInfo {
     pub(crate) server_tick: RepliconTick,
     pub(crate) system_tick: Tick,
     pub(crate) timestamp: Duration,
-    pub(crate) entities: Vec<(Entity, ComponentMask)>,
+    pub(crate) entities: Vec<MutatedEntityInfo>,
+}
+
+/// Entity data acknowledged by a mutation message.
+pub(crate) struct MutatedEntityInfo {
+    pub(crate) entity: Entity,
+    pub(crate) components: ComponentMask,
+    pub(crate) diff_cursors: DiffCursors,
 }