fix(crdt): make flush cost proportional to what changed

nodedb-lite/src/engine/crdt/engine/checkpoint.rs

@@ -0,0 +1,177 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! What each flush writes for the CRDT layer: an incremental update in the
+//! common case, a full snapshot only when the updates have earned one.
+//!
+//! A Loro snapshot export costs O(document). Writing one per flush therefore
+//! costs the size of the whole collection every `auto_flush_ms`, whatever the
+//! write rate was — the term that produced both unbounded file growth and a
+//! flush that held the reader-visible lock for all wall time. An update export
+//! costs O(new operations), so the same tick under the same load writes what
+//! actually changed.
+//!
+//! The base snapshot is still rewritten periodically. Restore replays every
+//! delta written since the base, so letting deltas accumulate without bound
+//! moves the cost from flush to open. Rewriting once the deltas reach a
+//! fraction of the base keeps that replay bounded by roughly the same
+//! fraction, and amortises the O(document) export over the writes that made it
+//! necessary.
+
+use super::types::{CrdtEngine, DELTA_CHECKPOINT_MIN_BYTES, DELTA_CHECKPOINT_RATIO};
+use crate::error::LiteError;
+
+/// One CRDT write a flush must perform, with the frontier it covers.
+pub struct CrdtWrite {
+    /// Collection this write belongs to.
+    pub collection: String,
+    /// Whether this is a fresh base snapshot or an update on top of one.
+    pub kind: CrdtWriteKind,
+    /// Payload, unwrapped — the caller adds whatever framing storage needs.
+    pub bytes: Vec<u8>,
+    /// Frontier the payload was exported at. Report it back through
+    /// [`CrdtEngine::mark_persisted`] once the write has committed.
+    pub version: loro::VersionVector,
+}
+
+/// Which of the two shapes a [`CrdtWrite`] carries.
+#[derive(Clone, Copy, Debug)]
+pub enum CrdtWriteKind {
+    /// A full snapshot that supersedes the base and the `superseded_deltas`
+    /// deltas written on top of it. Those must be deleted in the same batch,
+    /// or a later restore replays updates the new base already contains.
+    Checkpoint { superseded_deltas: u64 },
+    /// An update from the previously persisted frontier, stored under `seq`.
+    Delta { seq: u64 },
+}
+
+/// A committed [`CrdtWrite`], reported back so the engine can advance its
+/// bookkeeping. Carries the payload's length rather than the payload.
+pub struct CrdtPersisted {
+    pub collection: String,
+    pub kind: CrdtWriteKind,
+    pub bytes: usize,
+    pub version: loro::VersionVector,
+}
+
+impl CrdtWrite {
+    /// Describe this write without holding on to its payload.
+    pub fn persisted(&self) -> CrdtPersisted {
+        CrdtPersisted {
+            collection: self.collection.clone(),
+            kind: self.kind,
+            bytes: self.bytes.len(),
+            version: self.version.clone(),
+        }
+    }
+}
+
+impl CrdtEngine {
+    /// Decide what every collection needs written and export it.
+    ///
+    /// A collection whose oplog frontier has not moved since its last
+    /// persisted write is absent from the result entirely, so an idle store
+    /// exports nothing. A collection that has moved yields either an update
+    /// since that frontier, or — when it has no base yet, or its accumulated
+    /// deltas have reached [`DELTA_CHECKPOINT_RATIO`] of the base — a fresh
+    /// full snapshot.
+    ///
+    /// Pass the committed writes back through [`Self::mark_persisted`]; until
+    /// then the engine still considers them outstanding, so a failed batch is
+    /// retried rather than silently skipped.
+    pub fn plan_persistence(&self) -> Result<Vec<CrdtWrite>, LiteError> {
+        let mut out = Vec::new();
+        for (collection, state) in &self.states {
+            let version = state.oplog_version_vector();
+            let persisted = self.flushed_versions.get(collection);
+            if persisted == Some(&version) {
+                continue;
+            }
+
+            let (kind, bytes) = match persisted {
+                Some(from) if !self.checkpoint_is_due(collection) => {
+                    let bytes =
+                        state
+                            .export_updates_since(from)
+                            .map_err(|e| LiteError::Storage {
+                                detail: format!("delta export for '{collection}' failed: {e}"),
+                            })?;
+                    let seq = self.next_delta_seq.get(collection).copied().unwrap_or(0);
+                    (CrdtWriteKind::Delta { seq }, bytes)
+                }
+                _ => {
+                    let superseded_deltas =
+                        self.next_delta_seq.get(collection).copied().unwrap_or(0);
+                    let bytes = self.export_one(collection, state)?;
+                    (CrdtWriteKind::Checkpoint { superseded_deltas }, bytes)
+                }
+            };
+
+            out.push(CrdtWrite {
+                collection: collection.clone(),
+                kind,
+                bytes,
+                version,
+            });
+        }
+        Ok(out)
+    }
+
+    /// Advance the bookkeeping for writes that are now durable.
+    ///
+    /// Call only after the batch has committed — see [`Self::plan_persistence`].
+    pub fn mark_persisted(&mut self, persisted: impl IntoIterator<Item = CrdtPersisted>) {
+        for entry in persisted {
+            match entry.kind {
+                CrdtWriteKind::Checkpoint { .. } => {
+                    self.checkpoint_bytes
+                        .insert(entry.collection.clone(), entry.bytes);
+                    self.delta_bytes.insert(entry.collection.clone(), 0);
+                    self.next_delta_seq.insert(entry.collection.clone(), 0);
+                }
+                CrdtWriteKind::Delta { seq } => {
+                    *self
+                        .delta_bytes
+                        .entry(entry.collection.clone())
+                        .or_insert(0) += entry.bytes;
+                    self.next_delta_seq
+                        .insert(entry.collection.clone(), seq + 1);
+                }
+            }
+            self.flushed_versions
+                .insert(entry.collection, entry.version);
+        }
+    }
+
+    /// Seed the bookkeeping from what restore found on disk, so the first
+    /// flush after an open does not rewrite a base that is already current.
+    pub fn adopt_persisted_state(
+        &mut self,
+        collection: &str,
+        version: loro::VersionVector,
+        checkpoint_bytes: usize,
+        delta_bytes: usize,
+        next_delta_seq: u64,
+    ) {
+        self.flushed_versions
+            .insert(collection.to_string(), version);
+        self.checkpoint_bytes
+            .insert(collection.to_string(), checkpoint_bytes);
+        self.delta_bytes.insert(collection.to_string(), delta_bytes);
+        self.next_delta_seq
+            .insert(collection.to_string(), next_delta_seq);
+    }
+
+    /// Whether this collection's accumulated deltas have grown enough to be
+    /// worth folding back into the base.
+    ///
+    /// The floor matters as much as the ratio: a fraction of a small document
+    /// is a few hundred bytes, which would put us back to a full rewrite per
+    /// flush for exactly the collections where the delta path is cheapest.
+    fn checkpoint_is_due(&self, collection: &str) -> bool {
+        let Some(&base) = self.checkpoint_bytes.get(collection) else {
+            return true;
+        };
+        let accumulated = self.delta_bytes.get(collection).copied().unwrap_or(0);
+        accumulated >= (base / DELTA_CHECKPOINT_RATIO).max(DELTA_CHECKPOINT_MIN_BYTES)
+    }
+}

nodedb-lite/src/engine/crdt/engine/lifecycle.rs

@@ -46,6 +46,13 @@ impl CrdtEngine {
             policies: nodedb_crdt::PolicyRegistry::new(),
             registered_collections: std::collections::HashSet::new(),
             deferred: Vec::new(),
+            unpersisted_deltas: std::collections::HashSet::new(),
+            flushed_versions: HashMap::new(),
+            checkpoint_bytes: HashMap::new(),
+            delta_bytes: HashMap::new(),
+            next_delta_seq: HashMap::new(),
+            delta_writes: AtomicU64::new(0),
+            snapshot_exports: AtomicU64::new(0),
         })
     }
 
@@ -176,24 +183,38 @@ impl CrdtEngine {
         let Some(state) = self.states.get(collection) else {
             return Ok(Vec::new());
         };
-        state.export_snapshot().map_err(|e| LiteError::Storage {
-            detail: format!("snapshot export for '{collection}' failed: {e}"),
-        })
+        self.export_one(collection, state)
     }
 
     /// Export every collection's snapshot as `(collection, snapshot_bytes)`,
     /// in deterministic collection order.
     pub fn export_all_snapshots(&self) -> Result<Vec<(String, Vec<u8>)>, LiteError> {
         let mut out = Vec::with_capacity(self.states.len());
         for (collection, state) in &self.states {
-            let bytes = state.export_snapshot().map_err(|e| LiteError::Storage {
-                detail: format!("snapshot export for '{collection}' failed: {e}"),
-            })?;
+            let bytes = self.export_one(collection, state)?;
             out.push((collection.clone(), bytes));
         }
         Ok(out)
     }
 
+    /// Number of full snapshot exports performed since this engine was created.
+    pub fn snapshot_export_count(&self) -> u64 {
+        self.snapshot_exports
+            .load(std::sync::atomic::Ordering::Relaxed)
+    }
+
+    pub(in crate::engine::crdt) fn export_one(
+        &self,
+        collection: &str,
+        state: &CrdtState,
+    ) -> Result<Vec<u8>, LiteError> {
+        self.snapshot_exports
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        state.export_snapshot().map_err(|e| LiteError::Storage {
+            detail: format!("snapshot export for '{collection}' failed: {e}"),
+        })
+    }
+
     /// Compact Loro history on every collection to prevent unbounded growth.
     ///
     /// Replaces each internal LoroDoc with a shallow snapshot. Historical
@@ -204,6 +225,16 @@ impl CrdtEngine {
                 detail: format!("history compaction for '{collection}' failed: {e}"),
             })?;
         }
+        // Compaction rewrites the document without advancing its frontier, so
+        // neither the persisted base nor the updates on top of it describe the
+        // document any more, and the discarded history means an update export
+        // from the old frontier may not even be possible. Dropping both marks
+        // forces a fresh checkpoint, which also deletes the stale updates —
+        // `next_delta_seq` is deliberately kept, since it is the count of the
+        // entries that checkpoint has to delete.
+        self.flushed_versions.clear();
+        self.checkpoint_bytes.clear();
+        self.delta_bytes.clear();
         Ok(())
     }
 
@@ -267,6 +298,14 @@ impl CrdtEngine {
             .compact_at_version(version)
             .map_err(|e| LiteError::Storage {
                 detail: format!("compact_at_version '{collection}': {e}"),
-            })
+            })?;
+        // See `compact_history`: the frontier is unchanged but the exported
+        // bytes are not, so the collection must be re-persisted — as a fresh
+        // checkpoint, since the updates on top of the old base no longer
+        // describe this document.
+        self.flushed_versions.remove(collection);
+        self.checkpoint_bytes.remove(collection);
+        self.delta_bytes.remove(collection);
+        Ok(())
     }
 }

nodedb-lite/src/engine/crdt/engine/mod.rs

@@ -13,6 +13,7 @@
 //! through this engine. It wraps each as a Loro operation, generating a
 //! delta that will eventually sync to Origin.
 
+mod checkpoint;
 mod lifecycle;
 mod list_ops;
 mod mutate;
@@ -25,4 +26,5 @@ pub mod types;
 #[cfg(test)]
 mod tests;
 
+pub use checkpoint::{CrdtPersisted, CrdtWrite, CrdtWriteKind};
 pub use types::{CrdtBatchOp, CrdtEngine, CrdtField, PendingDelta};

nodedb-lite/src/engine/crdt/engine/mutate.rs

@@ -183,6 +183,7 @@ impl CrdtEngine {
                 delta_bytes,
                 seq: 0,
             });
+            self.unpersisted_deltas.insert(mutation_id);
         }
 
         Ok(count)
@@ -255,6 +256,7 @@ impl CrdtEngine {
             delta_bytes,
             seq: 0,
         });
+        self.unpersisted_deltas.insert(mutation_id);
         Ok((value, mutation_id))
     }
 }

nodedb-lite/src/engine/crdt/engine/pending.rs

@@ -24,6 +24,42 @@ impl CrdtEngine {
     /// The CRDT state is authoritative — pending deltas are regenerated on next mutation.
     pub fn clear_pending_deltas(&mut self) {
         self.pending_deltas.clear();
+        self.unpersisted_deltas.clear();
+    }
+
+    /// The pending deltas whose stored form may not match the queue.
+    ///
+    /// Entries already written under their own key are absent: the queue is
+    /// append-only, so an unchanged entry does not need rewriting. Report the
+    /// write back with [`Self::mark_pending_deltas_persisted`] once it has
+    /// committed.
+    pub fn pending_deltas_needing_write(&self) -> impl Iterator<Item = &PendingDelta> {
+        self.pending_deltas
+            .iter()
+            .filter(|d| self.unpersisted_deltas.contains(&d.mutation_id))
+            .inspect(|_| {
+                self.delta_writes
+                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            })
+    }
+
+    /// Number of queue entries handed out for writing since this engine was
+    /// created.
+    pub fn pending_delta_write_count(&self) -> u64 {
+        self.delta_writes.load(std::sync::atomic::Ordering::Relaxed)
+    }
+
+    /// Whether any pending delta needs writing.
+    pub fn has_unpersisted_deltas(&self) -> bool {
+        !self.unpersisted_deltas.is_empty()
+    }
+
+    /// Record that every pending delta is now stored as written.
+    ///
+    /// Call only after the batch has committed — see
+    /// [`Self::pending_deltas_needing_write`].
+    pub fn mark_pending_deltas_persisted(&mut self) {
+        self.unpersisted_deltas.clear();
     }
 
     /// Drop a single pending delta by `mutation_id` without touching CRDT state.
@@ -34,6 +70,7 @@ impl CrdtEngine {
     /// when the host's `SyncGate` rejects it for sync.
     pub fn drop_pending(&mut self, mutation_id: u64) {
         self.pending_deltas.retain(|d| d.mutation_id != mutation_id);
+        self.unpersisted_deltas.remove(&mutation_id);
     }
 
     /// Assign a stable stream seq to a pending delta the first time it is sent.
@@ -49,6 +86,8 @@ impl CrdtEngine {
             && d.seq == 0
         {
             d.seq = seq;
+            // The stored entry now carries a stale seq.
+            self.unpersisted_deltas.insert(mutation_id);
         }
     }
 
@@ -63,6 +102,7 @@ impl CrdtEngine {
     /// acks arrive.
     pub fn acknowledge(&mut self, acked_id: u64) {
         self.pending_deltas.retain(|d| d.mutation_id != acked_id);
+        self.unpersisted_deltas.remove(&acked_id);
     }
 
     /// Roll back a specific pending delta (after DeltaReject with CompensationHint).
@@ -79,6 +119,7 @@ impl CrdtEngine {
             .position(|d| d.mutation_id == mutation_id)
         {
             let delta = self.pending_deltas.remove(pos);
+            self.unpersisted_deltas.remove(&mutation_id);
             // Best-effort rollback: delete the affected document from its own
             // collection's document. The application should handle the
             // CompensationHint and re-create with corrected values.