diff --git a/nodedb-lite/src/engine/crdt/engine/checkpoint.rs b/nodedb-lite/src/engine/crdt/engine/checkpoint.rs new file mode 100644 index 0000000..8544c50 --- /dev/null +++ b/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, + /// 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, 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) { + 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) + } +} diff --git a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs b/nodedb-lite/src/engine/crdt/engine/lifecycle.rs index 90f8eb4..88d58ef 100644 --- a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs +++ b/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,9 +183,7 @@ 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)`, @@ -186,14 +191,30 @@ impl CrdtEngine { pub fn export_all_snapshots(&self) -> Result)>, 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, 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(()) } } diff --git a/nodedb-lite/src/engine/crdt/engine/mod.rs b/nodedb-lite/src/engine/crdt/engine/mod.rs index 2b05b3d..ca793f2 100644 --- a/nodedb-lite/src/engine/crdt/engine/mod.rs +++ b/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}; diff --git a/nodedb-lite/src/engine/crdt/engine/mutate.rs b/nodedb-lite/src/engine/crdt/engine/mutate.rs index 3ef8344..a2d7bbb 100644 --- a/nodedb-lite/src/engine/crdt/engine/mutate.rs +++ b/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)) } } diff --git a/nodedb-lite/src/engine/crdt/engine/pending.rs b/nodedb-lite/src/engine/crdt/engine/pending.rs index c28c940..e530ca6 100644 --- a/nodedb-lite/src/engine/crdt/engine/pending.rs +++ b/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 { + 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. diff --git a/nodedb-lite/src/engine/crdt/engine/persist.rs b/nodedb-lite/src/engine/crdt/engine/persist.rs index 711fdf5..ce51b3e 100644 --- a/nodedb-lite/src/engine/crdt/engine/persist.rs +++ b/nodedb-lite/src/engine/crdt/engine/persist.rs @@ -4,7 +4,9 @@ use std::sync::atomic::Ordering; -use super::types::{CrdtEngine, DELTA_KEY_PREFIX, PendingDelta, SNAPSHOT_KEY, VCLOCK_KEY}; +use super::types::{ + CrdtEngine, DELTA_KEY_PREFIX, PendingDelta, SNAPSHOT_KEY, STATE_DELTA_KEY, VCLOCK_KEY, +}; impl CrdtEngine { // ─── Persistence Helpers ───────────────────────────────────────── @@ -25,6 +27,9 @@ impl CrdtEngine { // Advance mutation ID counter past any restored deltas. let max_id = deltas.iter().map(|d| d.mutation_id).max().unwrap_or(0); self.next_mutation_id.store(max_id + 1, Ordering::Relaxed); + // The bulk blob is the only copy these came from, so none of + // them is stored under its own key yet. + self.unpersisted_deltas = deltas.iter().map(|d| d.mutation_id).collect(); self.pending_deltas = deltas; } Err(e) => { @@ -66,6 +71,9 @@ impl CrdtEngine { if let Some(max_id) = deltas.iter().map(|d| d.mutation_id).max() { self.next_mutation_id.store(max_id + 1, Ordering::Relaxed); } + // Every one of these was just read from its own key, so the stored + // form matches by construction. + self.unpersisted_deltas.clear(); self.pending_deltas = deltas; } @@ -83,6 +91,39 @@ impl CrdtEngine { SNAPSHOT_KEY } + /// Key for one incremental update on top of a collection's snapshot: + /// `loro_delta::`. + /// + /// Zero-padded hex so lexicographic key order is replay order, which is + /// what a prefix scan returns them in. + pub fn state_delta_key_for(collection: &str, seq: u64) -> Vec { + let mut key = Vec::with_capacity(STATE_DELTA_KEY.len() + collection.len() + 17); + key.extend_from_slice(STATE_DELTA_KEY); + key.extend_from_slice(collection.as_bytes()); + key.extend_from_slice(format!(":{seq:016x}").as_bytes()); + key + } + + /// Prefix shared by every state-update key, for prefix scans. + pub fn state_delta_key_prefix() -> &'static [u8] { + STATE_DELTA_KEY + } + + /// Recover `(collection, seq)` from a key produced by + /// [`Self::state_delta_key_for`]. + /// + /// Returns `None` for keys that do not carry the prefix, whose collection + /// is not UTF-8, or whose sequence does not parse — such an entry cannot be + /// routed or ordered, so the caller must skip it rather than guess. + pub fn state_delta_from_key(key: &[u8]) -> Option<(&str, u64)> { + let suffix = std::str::from_utf8(key.strip_prefix(STATE_DELTA_KEY)?).ok()?; + let (collection, seq) = suffix.rsplit_once(':')?; + if collection.is_empty() { + return None; + } + Some((collection, u64::from_str_radix(seq, 16).ok()?)) + } + /// Recover the collection name from a snapshot key produced by /// [`Self::snapshot_key_for`]. /// diff --git a/nodedb-lite/src/engine/crdt/engine/types.rs b/nodedb-lite/src/engine/crdt/engine/types.rs index d7fa511..11b6174 100644 --- a/nodedb-lite/src/engine/crdt/engine/types.rs +++ b/nodedb-lite/src/engine/crdt/engine/types.rs @@ -22,9 +22,30 @@ pub(super) const DELTA_KEY_PREFIX: &[u8] = b"delta:"; /// Each collection owns its own Loro document, so each gets its own entry: /// `loro_snapshot:`. pub(super) const SNAPSHOT_KEY: &[u8] = b"loro_snapshot:"; +/// Key prefix for the incremental updates written on top of a collection's +/// snapshot in the `LoroState` namespace: `loro_delta::`. +/// +/// Distinct from [`DELTA_KEY_PREFIX`], which holds the unsent-to-Origin sync +/// queue. These entries are durability, not sync: they are replayed on open +/// and deleted when the base snapshot that contains them is rewritten, +/// whereas a sync delta is deleted when Origin acknowledges it. +pub(super) const STATE_DELTA_KEY: &[u8] = b"loro_delta:"; /// Key for the vector clock in the `Meta` namespace. pub(super) const VCLOCK_KEY: &[u8] = b"vector_clock"; +/// Rewrite a collection's base snapshot once its accumulated updates reach +/// this fraction of it — a ratio, so the bound holds at any collection size. +/// +/// Restore replays every update written since the base, so this also bounds +/// the replay: open costs the base plus at most this fraction again. +pub(super) const DELTA_CHECKPOINT_RATIO: usize = 4; +/// Never rewrite the base for less than this many accumulated update bytes. +/// +/// A fraction of a small document is a few hundred bytes, which would restore +/// the full-rewrite-per-flush behaviour for precisely the collections where +/// incremental writes cost least. +pub(super) const DELTA_CHECKPOINT_MIN_BYTES: usize = 64 * 1024; + /// CRDT engine for edge devices. /// /// Not `Send` — owned by a single task. The `NodeDbLite` wrapper handles @@ -64,6 +85,43 @@ pub struct CrdtEngine { /// Deferred writes awaiting `flush_deltas()`, in the order they were /// applied. pub(super) deferred: Vec, + /// Oplog frontier each collection had when its snapshot was last written to + /// storage. + /// + /// A snapshot export is O(document), not O(new operations), so exporting a + /// collection whose frontier has not moved rewrites the identical bytes. + /// Comparing against this map is what lets an idle store do no snapshot + /// work at all. + pub(in crate::engine::crdt) flushed_versions: HashMap, + /// Size of each collection's base snapshot as last written. + /// + /// The denominator of the checkpoint decision: updates are folded back + /// into the base once they reach a fraction of it. + pub(in crate::engine::crdt) checkpoint_bytes: HashMap, + /// Update bytes written on top of each collection's current base. + pub(in crate::engine::crdt) delta_bytes: HashMap, + /// Sequence the next update for each collection is stored under. Also the + /// count of updates a checkpoint must delete. + pub(in crate::engine::crdt) next_delta_seq: HashMap, + /// Pending deltas whose stored form is not known to match the queue. + /// + /// The queue is append-only and each entry is written under its own key, + /// so an entry already on disk does not need rewriting. Only the ones + /// added — or edited, when a send assigns a `seq` — since the last flush + /// do. Without this the whole outbox is rewritten every tick, which for a + /// replica with no Origin to acknowledge it means an unbounded queue + /// rewritten in full once per `auto_flush_ms`. + pub(in crate::engine::crdt) unpersisted_deltas: std::collections::HashSet, + /// Number of queue entries handed out for writing. + /// + /// Exposed through [`CrdtEngine::pending_delta_write_count`] so callers can + /// assert on write volume directly: an idle store must not advance it. + pub(in crate::engine::crdt) delta_writes: AtomicU64, + /// Number of full snapshot exports performed for persistence. + /// + /// Exposed through [`CrdtEngine::snapshot_export_count`] so callers can + /// assert on export volume directly instead of inferring it from timings. + pub(in crate::engine::crdt) snapshot_exports: AtomicU64, } /// One deferred write awaiting `flush_deltas`, with the exact counter range diff --git a/nodedb-lite/src/engine/crdt/mod.rs b/nodedb-lite/src/engine/crdt/mod.rs index fcb4b84..72c66af 100644 --- a/nodedb-lite/src/engine/crdt/mod.rs +++ b/nodedb-lite/src/engine/crdt/mod.rs @@ -3,4 +3,4 @@ pub mod engine; mod policy; -pub use engine::{CrdtBatchOp, CrdtEngine, CrdtField}; +pub use engine::{CrdtBatchOp, CrdtEngine, CrdtField, CrdtPersisted, CrdtWrite, CrdtWriteKind}; diff --git a/nodedb-lite/src/nodedb/core/auto_compact.rs b/nodedb-lite/src/nodedb/core/auto_compact.rs index cfec04f..3a50b62 100644 --- a/nodedb-lite/src/nodedb/core/auto_compact.rs +++ b/nodedb-lite/src/nodedb/core/auto_compact.rs @@ -64,6 +64,10 @@ impl NodeDbLite { crate::runtime::spawn(async move { let mut ticker = crate::runtime::interval(period); + // A compaction that outlasts its own period must not be followed by + // the ticks it missed, back to back — repacking is the heavier of + // the two periodic tasks, so a burst of it is worse. + ticker.delay_missed_ticks(); // Consume the first tick so the initial period elapses before the // first compaction (matches Tokio's immediate-first-tick semantics // on native; on WASM the first tick already waits one period). diff --git a/nodedb-lite/src/nodedb/core/auto_flush.rs b/nodedb-lite/src/nodedb/core/auto_flush.rs index 1745710..e974ab3 100644 --- a/nodedb-lite/src/nodedb/core/auto_flush.rs +++ b/nodedb-lite/src/nodedb/core/auto_flush.rs @@ -56,6 +56,12 @@ impl NodeDbLite { crate::runtime::spawn(async move { let mut ticker = crate::runtime::interval(period); + // A flush that outlasts its own period must not be followed by the + // ticks it missed, back to back. Each pass takes the `crdt` lock, + // so a burst of them is a stretch of wall time in which no CRDT + // read can be scheduled — the shape a large store degenerated into + // once a single flush took longer than `interval_ms`. + ticker.delay_missed_ticks(); // Consume the first tick so the initial period elapses before the // first flush (matches Tokio's immediate-first-tick semantics on // native; on WASM the first tick already waits one period). diff --git a/nodedb-lite/src/nodedb/core/flush.rs b/nodedb-lite/src/nodedb/core/flush.rs index e107786..e72a06d 100644 --- a/nodedb-lite/src/nodedb/core/flush.rs +++ b/nodedb-lite/src/nodedb/core/flush.rs @@ -6,7 +6,7 @@ use crate::storage::engine::{StorageEngine, WriteOp}; use nodedb_types::Namespace; use nodedb_types::error::{NodeDbError, NodeDbResult}; -use crate::engine::crdt::CrdtEngine; +use crate::engine::crdt::{CrdtEngine, CrdtWriteKind}; use crate::nodedb::lock_ext::LockExt; use super::types::{ @@ -15,6 +15,26 @@ use super::types::{ }; impl NodeDbLite { + /// Number of full CRDT snapshot exports performed since this handle was + /// opened. + /// + /// A snapshot export costs O(document), so this is the term that decides + /// both flush latency and how much the file grows per tick. It is a + /// counter, not a timing, so a caller can assert on it directly: an idle + /// store must not advance it. + pub fn crdt_snapshot_export_count(&self) -> u64 { + self.crdt.lock_or_recover().snapshot_export_count() + } + + /// Number of unsent-delta queue entries written since this handle was + /// opened. + /// + /// The queue is append-only, so this advances by what was added, not by + /// the queue's length. An idle store must not advance it at all. + pub fn crdt_delta_write_count(&self) -> u64 { + self.crdt.lock_or_recover().pending_delta_write_count() + } + /// Persist all in-memory state to storage (call before shutdown). pub async fn flush(&self) -> NodeDbResult<()> { // Drain the buffered KV writes first — they have their own batch-commit @@ -40,21 +60,56 @@ impl NodeDbLite { // ── Persist one CRDT snapshot per collection (CRC32C wrapped) ── // Each collection owns its own Loro document, so each gets its own // storage entry under `loro_snapshot:`. - { + // + // A collection whose frontier has not moved is not written at all; one + // that has moved is written as an update since its last persisted + // frontier, and only periodically as a fresh snapshot. Exporting a full + // snapshot per collection per tick cost O(document) regardless of the + // write rate — unbounded file growth on an otherwise idle store, and an + // export duty cycle that starved readers once the document outgrew the + // flush interval. + let persisted = { let crdt = self.crdt.lock_or_recover(); - for (collection, snapshot) in - crdt.export_all_snapshots().map_err(NodeDbError::storage)? - { - ops.push(WriteOp::Put { - ns: Namespace::LoroState, - key: CrdtEngine::snapshot_key_for(&collection), - value: crate::storage::checksum::wrap(&snapshot), - }); + let plan = crdt.plan_persistence().map_err(NodeDbError::storage)?; + let mut persisted = Vec::with_capacity(plan.len()); + for write in plan { + persisted.push(write.persisted()); + match write.kind { + CrdtWriteKind::Checkpoint { superseded_deltas } => { + // In the same batch as the new base, so no restore ever + // sees a base with updates in front of it that it + // already contains. + for seq in 0..superseded_deltas { + ops.push(WriteOp::Delete { + ns: Namespace::LoroState, + key: CrdtEngine::state_delta_key_for(&write.collection, seq), + }); + } + ops.push(WriteOp::Put { + ns: Namespace::LoroState, + key: CrdtEngine::snapshot_key_for(&write.collection), + value: crate::storage::checksum::wrap(&write.bytes), + }); + } + CrdtWriteKind::Delta { seq } => { + ops.push(WriteOp::Put { + ns: Namespace::LoroState, + key: CrdtEngine::state_delta_key_for(&write.collection, seq), + value: crate::storage::checksum::wrap(&write.bytes), + }); + } + } } // Write pending deltas individually (append-only persistence). // Each delta is stored under `crdt:delta:{mutation_id:016x}`. - // Also write the legacy bulk blob for backward compatibility. + // + // Only the entries added or edited since the last flush are + // written. The queue is append-only and each entry owns its key, + // so rewriting an unchanged one stores bytes identical to the ones + // already there — and a replica with no Origin to acknowledge its + // deltas accumulates them without bound, which made that rewrite + // the whole outbox, once per `auto_flush_ms`. let pending = crdt.pending_deltas(); let max_mid = pending.iter().map(|d| d.mutation_id).max().unwrap_or(0); @@ -62,8 +117,10 @@ impl NodeDbLite { .iter() .map(|d| CrdtEngine::delta_storage_key(d.mutation_id)) .collect(); + let mut retired_any = false; for key in persisted_delta_keys { if !live_keys.contains(&key) { + retired_any = true; ops.push(WriteOp::Delete { ns: Namespace::Crdt, key, @@ -71,7 +128,7 @@ impl NodeDbLite { } } - for delta in pending { + for delta in crdt.pending_deltas_needing_write() { let key = CrdtEngine::delta_storage_key(delta.mutation_id); let value = CrdtEngine::serialize_delta(delta).map_err(NodeDbError::storage)?; ops.push(WriteOp::Put { @@ -81,15 +138,20 @@ impl NodeDbLite { }); } - // Legacy bulk blob (for clients that haven't upgraded to incremental restore). - let deltas_bulk = crdt - .serialize_pending_deltas() - .map_err(NodeDbError::storage)?; - ops.push(WriteOp::Put { - ns: Namespace::Crdt, - key: META_CRDT_DELTAS.to_vec(), - value: deltas_bulk, - }); + // Legacy bulk blob (for clients that haven't upgraded to incremental + // restore). It duplicates every entry above, so it is rewritten only + // when the queue actually changed rather than on every tick. + let queue_changed = retired_any || crdt.has_unpersisted_deltas(); + if queue_changed { + let deltas_bulk = crdt + .serialize_pending_deltas() + .map_err(NodeDbError::storage)?; + ops.push(WriteOp::Put { + ns: Namespace::Crdt, + key: META_CRDT_DELTAS.to_vec(), + value: deltas_bulk, + }); + } // Write the last-flushed mutation_id for partial flush safety. ops.push(WriteOp::Put { @@ -97,7 +159,9 @@ impl NodeDbLite { key: META_LAST_FLUSHED_MID.to_vec(), value: max_mid.to_le_bytes().to_vec(), }); - } + + persisted + }; // ── Persist per-collection CSR indices ── // When the pagedb segment extension is available (native PagedbStorage): @@ -275,6 +339,16 @@ impl NodeDbLite { .await .map_err(NodeDbError::storage)?; + // The CRDT writes are durable now, so advance the frontiers and the + // checkpoint accounting. Doing this only after the write means a failed + // batch leaves every collection outstanding and the next flush retries + // it. + { + let mut crdt = self.crdt.lock_or_recover(); + crdt.mark_persisted(persisted); + crdt.mark_pending_deltas_persisted(); + } + // ── Write HNSW vector segments to pagedb (native PagedbStorage only) ── #[cfg(not(target_arch = "wasm32"))] if let Some(ext) = seg_ext { diff --git a/nodedb-lite/src/nodedb/core/open/restore/crdt.rs b/nodedb-lite/src/nodedb/core/open/restore/crdt.rs index 8f85465..55365b3 100644 --- a/nodedb-lite/src/nodedb/core/open/restore/crdt.rs +++ b/nodedb-lite/src/nodedb/core/open/restore/crdt.rs @@ -57,6 +57,8 @@ impl NodeDbLite { let snapshot_entries = storage .scan_prefix(Namespace::LoroState, CrdtEngine::snapshot_key_prefix()) .await?; + let mut base_bytes: std::collections::HashMap = + std::collections::HashMap::new(); for (key, envelope) in &snapshot_entries { let Some(collection) = CrdtEngine::collection_from_snapshot_key(key) else { tracing::error!( @@ -67,6 +69,7 @@ impl NodeDbLite { }; match crate::storage::checksum::unwrap(envelope) { Some(snapshot) => { + base_bytes.insert(collection.to_string(), snapshot.len()); crdt.import_snapshot(collection, &snapshot).map_err(|e| { NodeDbError::storage(format!("CRDT restore of '{collection}' failed: {e}")) })?; @@ -83,6 +86,61 @@ impl NodeDbLite { } } + // ── Replay the incremental updates written on top of each snapshot ── + // Flush writes a full snapshot only periodically; between checkpoints it + // appends `loro_delta::` entries. A prefix scan returns + // them in key order, which the zero-padded sequence makes replay order. + // Skipping this loop would silently roll a collection back to its last + // checkpoint, so a corrupt entry is an error rather than a warning. + let mut delta_bytes: std::collections::HashMap = + std::collections::HashMap::new(); + let mut next_delta_seq: std::collections::HashMap = + std::collections::HashMap::new(); + let delta_entries = storage + .scan_prefix(Namespace::LoroState, CrdtEngine::state_delta_key_prefix()) + .await?; + for (key, envelope) in &delta_entries { + let Some((collection, seq)) = CrdtEngine::state_delta_from_key(key) else { + tracing::error!( + "CRDT update key is not a valid `loro_delta::` entry — \ + skipping; its collection cannot be determined without guessing." + ); + continue; + }; + let Some(update) = crate::storage::checksum::unwrap(envelope) else { + return Err(NodeDbError::segment_corrupted(format!( + "CRDT update '{collection}' #{seq} failed its CRC32C check; the writes it \ + carries are not in the snapshot behind it, so opening without it would \ + silently roll the collection back" + ))); + }; + delta_bytes + .entry(collection.to_string()) + .and_modify(|n| *n += update.len()) + .or_insert(update.len()); + next_delta_seq.insert(collection.to_string(), seq + 1); + crdt.import_snapshot(collection, &update).map_err(|e| { + NodeDbError::storage(format!( + "CRDT update replay for '{collection}' #{seq} failed: {e}" + )) + })?; + } + + // Seed the checkpoint accounting from what was on disk, so the first + // flush after open does not rewrite a base that is already current. + for (collection, base) in &base_bytes { + let Some(version) = crdt.state(collection).map(|s| s.oplog_version_vector()) else { + continue; + }; + crdt.adopt_persisted_state( + collection, + version, + *base, + delta_bytes.get(collection).copied().unwrap_or(0), + next_delta_seq.get(collection).copied().unwrap_or(0), + ); + } + // Rebuild the CRDT's registered-collection set from persisted bitemporal // flags so that SELECT queries on bitemporal collections work immediately // after open, even for collections with no inserted documents yet. diff --git a/nodedb-lite/src/runtime.rs b/nodedb-lite/src/runtime.rs index 161db85..a5f2e05 100644 --- a/nodedb-lite/src/runtime.rs +++ b/nodedb-lite/src/runtime.rs @@ -91,6 +91,24 @@ pub struct Interval { } impl Interval { + /// Measure the next period from when this tick is consumed rather than + /// from the schedule, so a slow consumer does not come back to a burst of + /// catch-up ticks. + /// + /// Tokio's default replays every missed tick immediately. For a periodic + /// task whose work can outlast its own period — a flush, say — that turns + /// one slow pass into back-to-back passes with no gap between them, and the + /// task never yields long enough for anything else to make progress. On + /// WASM each tick already sleeps a full period after the previous one, so + /// this is the behaviour there either way. + pub fn delay_missed_ticks(&mut self) { + #[cfg(not(target_arch = "wasm32"))] + { + self.inner + .set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + } + } + /// Wait until the next tick. pub async fn tick(&mut self) { #[cfg(not(target_arch = "wasm32"))] diff --git a/nodedb-lite/tests/crdt_flush_dirty_tracking.rs b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs new file mode 100644 index 0000000..a03f287 --- /dev/null +++ b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs @@ -0,0 +1,282 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Flush must do snapshot work in proportion to what changed. +//! +//! A Loro snapshot export is O(document), and `flush()` runs on a timer, so an +//! export that ignores whether the document moved makes an idle store rewrite +//! its entire state once per `auto_flush_ms` — file growth with no writes +//! behind it, and once the document is large enough to outlast the interval, a +//! flush duty cycle that leaves no room for readers. +//! +//! These assert on the export counter rather than on elapsed time, so they fail +//! for the reason they name on any machine. + +use std::sync::Arc; + +use nodedb_client::NodeDb; +use nodedb_lite::{Encryption, LiteConfig, NodeDbLite, PagedbStorageDefault}; +use nodedb_types::document::Document; +use nodedb_types::value::Value; + +/// Open with auto-flush disabled so every flush in the test is one we asked for. +async fn open_manual_flush(path: &std::path::Path) -> Arc> { + let storage = PagedbStorageDefault::open(path, Encryption::Plaintext) + .await + .expect("open storage"); + let config = LiteConfig { + auto_flush_ms: 0, + ..LiteConfig::default() + }; + NodeDbLite::open_with_config(storage, config) + .await + .expect("open db") +} + +async fn put_note(db: &NodeDbLite, id: &str, body: &str) { + let mut doc = Document::new(id.to_string()); + doc.set("body", Value::String(body.to_string())); + db.document_put("bt_notes", doc) + .await + .expect("document_put"); +} + +// --------------------------------------------------------------------------- +// idle_flush_exports_no_snapshots +// --------------------------------------------------------------------------- + +/// An idle store performs zero snapshot exports, however many times it flushes. +#[tokio::test] +async fn idle_flush_exports_no_snapshots() { + let dir = tempfile::tempdir().expect("tempdir"); + let db = open_manual_flush(&dir.path().join("idle_flush.pagedb")).await; + + db.execute_sql("CREATE COLLECTION bt_notes WITH (bitemporal=true)", &[]) + .await + .expect("create bitemporal collection"); + put_note(&db, "note0", "first").await; + db.flush().await.expect("flush the write"); + + let after_write = db.crdt_snapshot_export_count(); + assert!( + after_write > 0, + "the flush that persists a new write must export its snapshot" + ); + + for _ in 0..8 { + db.flush().await.expect("idle flush"); + } + + assert_eq!( + db.crdt_snapshot_export_count(), + after_write, + "an idle store must export nothing: {} flushes with no writes in between exported {} \ + additional snapshots, each one a full rewrite of the document", + 8, + db.crdt_snapshot_export_count() - after_write + ); +} + +// --------------------------------------------------------------------------- +// write_after_flush_is_persisted +// --------------------------------------------------------------------------- + +/// Skipping unchanged collections must not turn into skipping changed ones: a +/// write made after a flush is persisted by the next one and reads back after a +/// reopen. +#[tokio::test] +async fn write_after_flush_is_persisted() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("dirty_after_write.pagedb"); + + { + let db = open_manual_flush(&path).await; + db.execute_sql("CREATE COLLECTION bt_notes WITH (bitemporal=true)", &[]) + .await + .expect("create bitemporal collection"); + + put_note(&db, "note0", "first").await; + db.flush().await.expect("flush first write"); + + put_note(&db, "note1", "second").await; + db.flush().await.expect("flush second write"); + } + + let db = open_manual_flush(&path).await; + let fetched = db + .document_get("bt_notes", "note1") + .await + .expect("document_get after reopen"); + assert!( + fetched.is_some(), + "the row written after the first flush must be durable" + ); + + let dump = db.diagnostic_dump().await; + assert!( + dump.storage_counts.loro_state > 0, + "the collection's CRDT snapshot must be on disk, not just its row: {:?}", + dump.storage_counts + ); +} + +// --------------------------------------------------------------------------- +// sustained_writes_do_not_export_a_snapshot_per_flush +// --------------------------------------------------------------------------- + +/// Under sustained writes — the workload dirty-tracking alone does not help — +/// flush writes updates, not a fresh snapshot per tick. +#[tokio::test] +async fn sustained_writes_do_not_export_a_snapshot_per_flush() { + let dir = tempfile::tempdir().expect("tempdir"); + let db = open_manual_flush(&dir.path().join("sustained.pagedb")).await; + + db.execute_sql("CREATE COLLECTION bt_notes WITH (bitemporal=true)", &[]) + .await + .expect("create bitemporal collection"); + put_note(&db, "note0", "first").await; + db.flush().await.expect("flush the first write"); + let after_first = db.crdt_snapshot_export_count(); + + for i in 1..64 { + put_note(&db, &format!("note{i}"), &format!("body {i}")).await; + db.flush().await.expect("flush"); + } + + assert_eq!( + db.crdt_snapshot_export_count(), + after_first, + "63 flushes, each with one small write behind it, exported {} full snapshots; a write of \ + a few hundred bytes must cost a few hundred bytes of update, not a rewrite of the \ + collection", + db.crdt_snapshot_export_count() - after_first + ); +} + +// --------------------------------------------------------------------------- +// updates_written_between_checkpoints_survive_reopen +// --------------------------------------------------------------------------- + +/// The writes that reach disk as updates rather than as part of a snapshot must +/// come back on open. Losing them is invisible until a reopen, so this asserts +/// the replay directly. +#[tokio::test] +async fn updates_written_between_checkpoints_survive_reopen() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("update_replay.pagedb"); + + { + let db = open_manual_flush(&path).await; + db.execute_sql("CREATE COLLECTION bt_notes WITH (bitemporal=true)", &[]) + .await + .expect("create bitemporal collection"); + + put_note(&db, "note0", "checkpointed").await; + db.flush().await.expect("flush the checkpoint"); + let exports = db.crdt_snapshot_export_count(); + + for i in 1..8 { + put_note(&db, &format!("note{i}"), &format!("update {i}")).await; + db.flush().await.expect("flush an update"); + } + assert_eq!( + db.crdt_snapshot_export_count(), + exports, + "these writes must have gone to disk as updates for this test to mean anything" + ); + } + + let db = open_manual_flush(&path).await; + for i in 1..8 { + assert!( + db.document_get("bt_notes", &format!("note{i}")) + .await + .expect("document_get after reopen") + .is_some(), + "note{i} was written as an update after the last checkpoint and must be replayed on \ + open, not rolled back to the checkpoint" + ); + } +} + +// --------------------------------------------------------------------------- +// idle_flush_does_not_rewrite_the_sync_queue +// --------------------------------------------------------------------------- + +/// The unsent-delta queue is append-only and each entry owns its key, so an +/// idle flush must write none of it. +/// +/// A replica with no Origin never has a delta acknowledged, so the queue only +/// grows — and rewriting it in full per tick is the same unbounded cost the +/// snapshot rewrite had, one layer over. +#[tokio::test] +async fn idle_flush_does_not_rewrite_the_sync_queue() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("idle_queue.pagedb"); + let db = open_manual_flush(&path).await; + + db.execute_sql("CREATE COLLECTION bt_notes WITH (bitemporal=true)", &[]) + .await + .expect("create bitemporal collection"); + for i in 0..32 { + put_note(&db, &format!("note{i}"), &format!("body {i}")).await; + } + db.flush().await.expect("flush the writes"); + + let after_writes = db.crdt_delta_write_count(); + assert!( + after_writes >= 32, + "the flush that persists the queue must write its entries" + ); + + for _ in 0..8 { + db.flush().await.expect("idle flush"); + } + + assert_eq!( + db.crdt_delta_write_count(), + after_writes, + "an idle store must write none of the queue: 8 flushes with nothing queued in between \ + rewrote {} entries, and a replica with no Origin never retires any of them", + db.crdt_delta_write_count() - after_writes + ); +} + +// --------------------------------------------------------------------------- +// queued_deltas_survive_reopen +// --------------------------------------------------------------------------- + +/// Writing only the changed entries must still leave the whole queue on disk: +/// the deltas are what a replica has yet to send, so losing them loses writes +/// that no Origin has seen. +#[tokio::test] +async fn queued_deltas_survive_reopen() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("queue_reopen.pagedb"); + + { + let db = open_manual_flush(&path).await; + db.execute_sql("CREATE COLLECTION bt_notes WITH (bitemporal=true)", &[]) + .await + .expect("create bitemporal collection"); + + // Two batches with a flush in between, so the second flush writes only + // the second batch and the first must already be durable. + for i in 0..4 { + put_note(&db, &format!("first{i}"), "a").await; + } + db.flush().await.expect("flush first batch"); + for i in 0..4 { + put_note(&db, &format!("second{i}"), "b").await; + } + db.flush().await.expect("flush second batch"); + } + + let db = open_manual_flush(&path).await; + let dump = db.diagnostic_dump().await; + assert!( + dump.storage_counts.crdt >= 8, + "every queued delta must be readable after reopen, from both batches; \ + storage_counts: {:?}", + dump.storage_counts + ); +}