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

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

@@ -0,0 +1,211 @@
+// 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,
+    /// Compaction epoch the collection was at when this was planned. A
+    /// compaction landing before the write commits makes the payload stale in
+    /// a way the frontier cannot show, since compaction leaves the frontier
+    /// where it was; the epoch is what distinguishes the two.
+    pub epoch: u64,
+}
+
+/// 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 {
+    /// Collection the committed write belongs to.
+    pub collection: String,
+    /// Whether it was a fresh base snapshot or an update on top of one.
+    pub kind: CrdtWriteKind,
+    /// Length of the payload that was written.
+    pub bytes: usize,
+    /// Frontier the payload was exported at.
+    pub version: loro::VersionVector,
+    /// Compaction epoch the write was planned at — see [`CrdtWrite::epoch`].
+    pub epoch: u64,
+}
+
+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(),
+            epoch: self.epoch,
+        }
+    }
+}
+
+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,
+                epoch: self.state_epoch(collection),
+            });
+        }
+        Ok(out)
+    }
+
+    /// Compaction epoch a collection is currently at. Absent means never
+    /// compacted, which is epoch zero.
+    pub(in crate::engine::crdt) fn state_epoch(&self, collection: &str) -> u64 {
+        self.state_epochs.get(collection).copied().unwrap_or(0)
+    }
+
+    /// Record that a collection's document has been structurally rewritten, so
+    /// any write planned against the previous form is no longer describable by
+    /// its frontier. See [`CrdtEngine::state_epochs`].
+    pub(in crate::engine::crdt) fn advance_state_epoch(&mut self, collection: &str) {
+        *self.state_epochs.entry(collection.to_string()).or_insert(0) += 1;
+    }
+
+    /// Advance the bookkeeping for writes that are now durable.
+    ///
+    /// A write whose collection was compacted between the plan and the commit
+    /// is discarded rather than recorded: the bytes on disk describe a document
+    /// that no longer exists, and the marks compaction dropped must stay
+    /// dropped so the next flush writes a fresh checkpoint.
+    ///
+    /// 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 {
+            if self.state_epoch(&entry.collection) != entry.epoch {
+                continue;
+            }
+            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/flush_ack_tests.rs

@@ -0,0 +1,140 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! What a flush may still claim once it has let go of the engine.
+//!
+//! A flush plans and exports under the engine lock, releases it while its batch
+//! commits, then re-takes it to record what is now durable. Anything that
+//! happens in that window changed state the batch does not carry, so the
+//! acknowledgement has to be able to tell the two apart. These assert that it
+//! does, for each way the window can be used.
+
+use loro::LoroValue;
+
+use super::types::CrdtEngine;
+
+/// A write queued while the batch was committing was not in it. Retiring its
+/// dirty mark strands it: an append-only queue is only revisited when it
+/// changes, so the entry would sit in memory until the process ended and the
+/// write it carries would never reach Origin.
+#[test]
+fn a_delta_queued_during_a_flush_is_not_retired_unwritten() {
+    let mut engine = CrdtEngine::new(1).unwrap();
+    engine
+        .upsert("users", "u1", &[("n", LoroValue::I64(1))])
+        .unwrap();
+
+    let planned: Vec<(u64, u64)> = engine
+        .pending_deltas_needing_write()
+        .map(|(delta, revision)| (delta.mutation_id, revision))
+        .collect();
+    assert_eq!(planned.len(), 1);
+
+    // The batch is committing. A second write lands before it is acknowledged.
+    let queued_during = engine
+        .upsert("users", "u2", &[("n", LoroValue::I64(2))])
+        .unwrap();
+
+    engine.mark_pending_deltas_persisted(planned);
+
+    let still_dirty: Vec<u64> = engine
+        .pending_deltas_needing_write()
+        .map(|(delta, _)| delta.mutation_id)
+        .collect();
+    assert_eq!(
+        still_dirty,
+        vec![queued_during],
+        "the entry queued while the batch was in flight was not in it, so it must still be \
+         waiting to be written"
+    );
+    assert_eq!(
+        engine.pending_delta_write_count(),
+        1,
+        "only the entry the batch actually carried counts as written"
+    );
+}
+
+/// The same window catches an *edit*, not only an insertion: assigning a stream
+/// seq rewrites an entry that is already on disk, so its stored form goes stale
+/// and it has to be written again.
+#[test]
+fn a_delta_resequenced_during_a_flush_stays_dirty() {
+    let mut engine = CrdtEngine::new(1).unwrap();
+    let mid = engine
+        .upsert("users", "u1", &[("n", LoroValue::I64(1))])
+        .unwrap();
+
+    let planned: Vec<(u64, u64)> = engine
+        .pending_deltas_needing_write()
+        .map(|(delta, revision)| (delta.mutation_id, revision))
+        .collect();
+
+    // The batch is committing. The delta is sent and is assigned its seq.
+    engine.set_pending_delta_seq(mid, 7);
+
+    engine.mark_pending_deltas_persisted(planned);
+
+    assert!(
+        engine.has_unpersisted_deltas(),
+        "the stored entry carries seq 0 while the queue carries seq 7 — a resend after a \
+         restart would use the wrong seq, so it must be rewritten"
+    );
+}
+
+/// An acknowledgement that arrives with nothing to report leaves the queue
+/// alone: replaying one must not resurrect an entry or double-count a write.
+#[test]
+fn acknowledging_the_same_batch_twice_changes_nothing() {
+    let mut engine = CrdtEngine::new(1).unwrap();
+    engine
+        .upsert("users", "u1", &[("n", LoroValue::I64(1))])
+        .unwrap();
+
+    let planned: Vec<(u64, u64)> = engine
+        .pending_deltas_needing_write()
+        .map(|(delta, revision)| (delta.mutation_id, revision))
+        .collect();
+
+    engine.mark_pending_deltas_persisted(planned.clone());
+    engine.mark_pending_deltas_persisted(planned);
+
+    assert!(!engine.has_unpersisted_deltas());
+    assert_eq!(
+        engine.pending_delta_write_count(),
+        1,
+        "the second report describes the same write, not another one"
+    );
+}
+
+/// Compaction replaces a document without moving its frontier, so a flush that
+/// planned against the previous form cannot be recognised as stale by frontier
+/// alone. A compaction landing while the batch commits must leave the
+/// collection needing a fresh checkpoint.
+#[test]
+fn a_compaction_during_a_flush_is_not_undone_by_its_acknowledgement() {
+    let mut engine = CrdtEngine::new(1).unwrap();
+    engine
+        .upsert("users", "u1", &[("n", LoroValue::I64(1))])
+        .unwrap();
+
+    let plan = engine.plan_persistence().unwrap();
+    assert_eq!(plan.len(), 1);
+    let persisted: Vec<_> = plan.iter().map(|write| write.persisted()).collect();
+
+    // The batch is committing. Compaction discards the history underneath it.
+    engine.compact_history().unwrap();
+
+    engine.mark_persisted(persisted);
+
+    let replan = engine.plan_persistence().unwrap();
+    assert_eq!(
+        replan.len(),
+        1,
+        "the compacted document is not what the in-flight batch wrote, so it must still be \
+         planned for"
+    );
+    assert!(
+        matches!(replan[0].kind, super::CrdtWriteKind::Checkpoint { .. }),
+        "and as a fresh checkpoint — an update exported from the discarded history does not \
+         apply to the base on disk"
+    );
+}