From e599d88f4cee999406ee03b9d65bad5c96d0ff1d Mon Sep 17 00:00:00 2001 From: Mohd Khairi Date: Sat, 1 Aug 2026 19:04:58 +0800 Subject: [PATCH 1/4] fix(crdt): export a collection's snapshot only when it has changed Every flush re-exported and rewrote the full Loro snapshot of every collection, with no check for whether the document had moved. A snapshot export costs O(document), so the work an idle store did per tick was the size of its whole state. Two consequences followed from that one assumption. The file grew by a full snapshot copy per `auto_flush_ms` with no writes behind it, and the superseded pages are only reclaimed by auto-compact, which is off by default. And once a document grew large enough that its export outlasted the flush interval, the flush task held the reader-visible `crdt` lock for essentially all wall time, so CRDT reads never got scheduled. Track the frontier each collection had when its snapshot was last written, and export only those whose `oplog_version_vector()` has moved since. The mark is recorded after the batch commits, so a failed write leaves the collection dirty and is retried rather than skipped. History compaction rewrites the document without advancing its frontier, so it drops the marks it invalidates. `crdt_snapshot_export_count()` exposes the export count, so the property can be asserted directly instead of inferred from a timing. --- .../src/engine/crdt/engine/lifecycle.rs | 71 +++++++++- nodedb-lite/src/engine/crdt/engine/types.rs | 13 ++ nodedb-lite/src/nodedb/core/flush.rs | 40 +++++- .../tests/crdt_flush_dirty_tracking.rs | 124 ++++++++++++++++++ 4 files changed, 236 insertions(+), 12 deletions(-) create mode 100644 nodedb-lite/tests/crdt_flush_dirty_tracking.rs diff --git a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs b/nodedb-lite/src/engine/crdt/engine/lifecycle.rs index 90f8eb4..b4f970a 100644 --- a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs +++ b/nodedb-lite/src/engine/crdt/engine/lifecycle.rs @@ -46,6 +46,8 @@ impl CrdtEngine { policies: nodedb_crdt::PolicyRegistry::new(), registered_collections: std::collections::HashSet::new(), deferred: Vec::new(), + flushed_versions: HashMap::new(), + snapshot_exports: AtomicU64::new(0), }) } @@ -176,9 +178,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 +186,63 @@ 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) } + /// Export only the collections whose oplog frontier has moved since their + /// snapshot was last persisted. + /// + /// Each entry carries the frontier the bytes were taken at; pass it back to + /// [`Self::mark_snapshots_flushed`] once the write has committed, so a + /// failed write is retried rather than silently skipped. Writes landing + /// between the export and the mark are not lost either — they advance the + /// frontier past the recorded one, leaving the collection dirty. + /// + /// An unmodified collection is absent from the result entirely, so an idle + /// store performs no export work. + pub fn export_dirty_snapshots( + &self, + ) -> Result, loro::VersionVector)>, LiteError> { + let mut out = Vec::new(); + for (collection, state) in &self.states { + let version = state.oplog_version_vector(); + if self.flushed_versions.get(collection) == Some(&version) { + continue; + } + let bytes = self.export_one(collection, state)?; + out.push((collection.clone(), bytes, version)); + } + Ok(out) + } + + /// Record the frontiers whose snapshots are now durable. + /// + /// Call only after the write has committed — see + /// [`Self::export_dirty_snapshots`]. + pub fn mark_snapshots_flushed( + &mut self, + flushed: impl IntoIterator, + ) { + self.flushed_versions.extend(flushed); + } + + /// 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) + } + + 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 +253,10 @@ impl CrdtEngine { detail: format!("history compaction for '{collection}' failed: {e}"), })?; } + // Compaction rewrites the document without advancing its frontier, so + // the persisted snapshot no longer matches what the document would + // export. Dropping the marks forces the next flush to rewrite it. + self.flushed_versions.clear(); Ok(()) } @@ -267,6 +320,10 @@ 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. + self.flushed_versions.remove(collection); + Ok(()) } } diff --git a/nodedb-lite/src/engine/crdt/engine/types.rs b/nodedb-lite/src/engine/crdt/engine/types.rs index d7fa511..adecaf2 100644 --- a/nodedb-lite/src/engine/crdt/engine/types.rs +++ b/nodedb-lite/src/engine/crdt/engine/types.rs @@ -64,6 +64,19 @@ 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, + /// 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/nodedb/core/flush.rs b/nodedb-lite/src/nodedb/core/flush.rs index e107786..4104a83 100644 --- a/nodedb-lite/src/nodedb/core/flush.rs +++ b/nodedb-lite/src/nodedb/core/flush.rs @@ -15,6 +15,17 @@ 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() + } + /// 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,16 +51,26 @@ 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:`. - { + // + // Only collections whose oplog frontier moved since their last durable + // snapshot are exported. A snapshot export costs O(document), so + // exporting on every tick regardless of writes rewrote the whole + // document once per `auto_flush_ms` — unbounded file growth on an + // otherwise idle store, and an export duty cycle that starved readers + // once the document outgrew the interval. + let flushed_versions = { let crdt = self.crdt.lock_or_recover(); - for (collection, snapshot) in - crdt.export_all_snapshots().map_err(NodeDbError::storage)? - { + let dirty = crdt + .export_dirty_snapshots() + .map_err(NodeDbError::storage)?; + let mut flushed = Vec::with_capacity(dirty.len()); + for (collection, snapshot, version) in dirty { ops.push(WriteOp::Put { ns: Namespace::LoroState, key: CrdtEngine::snapshot_key_for(&collection), value: crate::storage::checksum::wrap(&snapshot), }); + flushed.push((collection, version)); } // Write pending deltas individually (append-only persistence). @@ -97,7 +118,9 @@ impl NodeDbLite { key: META_LAST_FLUSHED_MID.to_vec(), value: max_mid.to_le_bytes().to_vec(), }); - } + + flushed + }; // ── Persist per-collection CSR indices ── // When the pagedb segment extension is available (native PagedbStorage): @@ -275,6 +298,13 @@ impl NodeDbLite { .await .map_err(NodeDbError::storage)?; + // The snapshots are durable now, so record the frontiers they were + // taken at. Doing this only after the write means a failed batch leaves + // every collection dirty and the next flush retries it. + self.crdt + .lock_or_recover() + .mark_snapshots_flushed(flushed_versions); + // ── 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/tests/crdt_flush_dirty_tracking.rs b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs new file mode 100644 index 0000000..68a68fb --- /dev/null +++ b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs @@ -0,0 +1,124 @@ +// 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 + ); +} + +// --------------------------------------------------------------------------- +// flush_after_write_exports_again +// --------------------------------------------------------------------------- + +/// Dirty tracking must not turn into never-export: a write after a flush is +/// exported by the next one, and reads back after a reopen. +#[tokio::test] +async fn flush_after_write_exports_again() { + 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"); + let after_first = db.crdt_snapshot_export_count(); + + put_note(&db, "note1", "second").await; + db.flush().await.expect("flush second write"); + assert!( + db.crdt_snapshot_export_count() > after_first, + "a collection written since its last flush must be exported again" + ); + } + + 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 + ); +} From 36dadb55b58fabf7b1f2516efbdcde42ac23c699 Mon Sep 17 00:00:00 2001 From: Mohd Khairi Date: Sat, 1 Aug 2026 19:22:31 +0800 Subject: [PATCH 2/4] feat(crdt): write incremental updates between snapshot checkpoints Exporting only changed collections removes the idle cost, but the store's actual workload is sustained writes, and under those every flush still rewrote each dirty collection in full. One row changing in a 77 MB collection cost 77 MB of export and 77 MB of superseded pages, once per `auto_flush_ms`. Between checkpoints a flush now exports the operations since the last persisted frontier and stores them under `loro_delta::`, so per-flush cost is O(new operations). The base snapshot is rewritten once the accumulated updates reach a quarter of it, which bounds what open has to replay and amortises the O(document) export over the writes that made it necessary. A checkpoint deletes the updates it supersedes in the same batch, so no restore ever replays operations the base already contains. These updates are durability, not sync: they are deleted when the base that contains them is rewritten, whereas `crdt:delta:` entries are the queue to Origin and are deleted on acknowledgement. Reusing that queue as the durability log would drop the state of every row Origin had already acknowledged. Restore replays the updates in key order after importing the base and seeds the checkpoint accounting from what it found, so the first flush after an open does not rewrite a base that is already current. An update that fails its CRC32C check is an error rather than a warning: opening without it would silently roll the collection back to the checkpoint. --- .../src/engine/crdt/engine/checkpoint.rs | 177 ++++++++++++++++++ .../src/engine/crdt/engine/lifecycle.rs | 62 +++--- nodedb-lite/src/engine/crdt/engine/mod.rs | 2 + nodedb-lite/src/engine/crdt/engine/persist.rs | 37 +++- nodedb-lite/src/engine/crdt/engine/types.rs | 31 +++ nodedb-lite/src/engine/crdt/mod.rs | 2 +- nodedb-lite/src/nodedb/core/flush.rs | 70 ++++--- .../src/nodedb/core/open/restore/crdt.rs | 58 ++++++ .../tests/crdt_flush_dirty_tracking.rs | 93 ++++++++- 9 files changed, 454 insertions(+), 78 deletions(-) create mode 100644 nodedb-lite/src/engine/crdt/engine/checkpoint.rs 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 b4f970a..8bec31d 100644 --- a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs +++ b/nodedb-lite/src/engine/crdt/engine/lifecycle.rs @@ -47,6 +47,9 @@ impl CrdtEngine { registered_collections: std::collections::HashSet::new(), deferred: Vec::new(), flushed_versions: HashMap::new(), + checkpoint_bytes: HashMap::new(), + delta_bytes: HashMap::new(), + next_delta_seq: HashMap::new(), snapshot_exports: AtomicU64::new(0), }) } @@ -192,50 +195,17 @@ impl CrdtEngine { Ok(out) } - /// Export only the collections whose oplog frontier has moved since their - /// snapshot was last persisted. - /// - /// Each entry carries the frontier the bytes were taken at; pass it back to - /// [`Self::mark_snapshots_flushed`] once the write has committed, so a - /// failed write is retried rather than silently skipped. Writes landing - /// between the export and the mark are not lost either — they advance the - /// frontier past the recorded one, leaving the collection dirty. - /// - /// An unmodified collection is absent from the result entirely, so an idle - /// store performs no export work. - pub fn export_dirty_snapshots( - &self, - ) -> Result, loro::VersionVector)>, LiteError> { - let mut out = Vec::new(); - for (collection, state) in &self.states { - let version = state.oplog_version_vector(); - if self.flushed_versions.get(collection) == Some(&version) { - continue; - } - let bytes = self.export_one(collection, state)?; - out.push((collection.clone(), bytes, version)); - } - Ok(out) - } - - /// Record the frontiers whose snapshots are now durable. - /// - /// Call only after the write has committed — see - /// [`Self::export_dirty_snapshots`]. - pub fn mark_snapshots_flushed( - &mut self, - flushed: impl IntoIterator, - ) { - self.flushed_versions.extend(flushed); - } - /// 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) } - fn export_one(&self, collection: &str, state: &CrdtState) -> Result, LiteError> { + 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 { @@ -254,9 +224,15 @@ impl CrdtEngine { })?; } // Compaction rewrites the document without advancing its frontier, so - // the persisted snapshot no longer matches what the document would - // export. Dropping the marks forces the next flush to rewrite it. + // 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(()) } @@ -322,8 +298,12 @@ impl CrdtEngine { 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. + // 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/persist.rs b/nodedb-lite/src/engine/crdt/engine/persist.rs index 711fdf5..66e2111 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 ───────────────────────────────────────── @@ -83,6 +85,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 adecaf2..7eb6039 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 @@ -72,6 +93,16 @@ pub struct CrdtEngine { /// 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, /// Number of full snapshot exports performed for persistence. /// /// Exposed through [`CrdtEngine::snapshot_export_count`] so callers can 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/flush.rs b/nodedb-lite/src/nodedb/core/flush.rs index 4104a83..ed54298 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::{ @@ -52,25 +52,44 @@ impl NodeDbLite { // Each collection owns its own Loro document, so each gets its own // storage entry under `loro_snapshot:`. // - // Only collections whose oplog frontier moved since their last durable - // snapshot are exported. A snapshot export costs O(document), so - // exporting on every tick regardless of writes rewrote the whole - // document once per `auto_flush_ms` — unbounded file growth on an - // otherwise idle store, and an export duty cycle that starved readers - // once the document outgrew the interval. - let flushed_versions = { + // 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(); - let dirty = crdt - .export_dirty_snapshots() - .map_err(NodeDbError::storage)?; - let mut flushed = Vec::with_capacity(dirty.len()); - for (collection, snapshot, version) in dirty { - ops.push(WriteOp::Put { - ns: Namespace::LoroState, - key: CrdtEngine::snapshot_key_for(&collection), - value: crate::storage::checksum::wrap(&snapshot), - }); - flushed.push((collection, version)); + 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). @@ -119,7 +138,7 @@ impl NodeDbLite { value: max_mid.to_le_bytes().to_vec(), }); - flushed + persisted }; // ── Persist per-collection CSR indices ── @@ -298,12 +317,11 @@ impl NodeDbLite { .await .map_err(NodeDbError::storage)?; - // The snapshots are durable now, so record the frontiers they were - // taken at. Doing this only after the write means a failed batch leaves - // every collection dirty and the next flush retries it. - self.crdt - .lock_or_recover() - .mark_snapshots_flushed(flushed_versions); + // 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. + self.crdt.lock_or_recover().mark_persisted(persisted); // ── Write HNSW vector segments to pagedb (native PagedbStorage only) ── #[cfg(not(target_arch = "wasm32"))] 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/tests/crdt_flush_dirty_tracking.rs b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs index 68a68fb..74ec636 100644 --- a/nodedb-lite/tests/crdt_flush_dirty_tracking.rs +++ b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs @@ -77,13 +77,14 @@ async fn idle_flush_exports_no_snapshots() { } // --------------------------------------------------------------------------- -// flush_after_write_exports_again +// write_after_flush_is_persisted // --------------------------------------------------------------------------- -/// Dirty tracking must not turn into never-export: a write after a flush is -/// exported by the next one, and reads back after a reopen. +/// 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 flush_after_write_exports_again() { +async fn write_after_flush_is_persisted() { let dir = tempfile::tempdir().expect("tempdir"); let path = dir.path().join("dirty_after_write.pagedb"); @@ -95,14 +96,9 @@ async fn flush_after_write_exports_again() { put_note(&db, "note0", "first").await; db.flush().await.expect("flush first write"); - let after_first = db.crdt_snapshot_export_count(); put_note(&db, "note1", "second").await; db.flush().await.expect("flush second write"); - assert!( - db.crdt_snapshot_export_count() > after_first, - "a collection written since its last flush must be exported again" - ); } let db = open_manual_flush(&path).await; @@ -122,3 +118,82 @@ async fn flush_after_write_exports_again() { 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" + ); + } +} From 108cbd2b3c0b5991986bc2a30ca32b34f9596e66 Mon Sep 17 00:00:00 2001 From: Mohd Khairi Date: Sat, 1 Aug 2026 19:24:44 +0800 Subject: [PATCH 3/4] fix(core): let a slow flush delay its next tick instead of bursting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tokio's default interval behaviour replays every missed tick as soon as the consumer comes back. For a task whose work can outlast its own period that turns one slow pass into back-to-back passes with no gap: each flush takes the reader-visible `crdt` lock, so a burst of them is a stretch of wall time in which no CRDT read is scheduled at all. Both periodic tasks now measure the next period from the end of the previous pass. This is already the WASM behaviour, so it only changes native. The remaining half of the lock problem is that the snapshot export itself runs under `self.crdt`. Moving it out needs a `CrdtState` handle that can be exported while the engine lock is released, which `nodedb-crdt` deliberately prevents today — its `_single_owner` marker makes sharing one a compile error. That is a change to its contract, not to this crate, so it is not folded in here. --- nodedb-lite/src/nodedb/core/auto_compact.rs | 4 ++++ nodedb-lite/src/nodedb/core/auto_flush.rs | 6 ++++++ nodedb-lite/src/runtime.rs | 18 ++++++++++++++++++ 3 files changed, 28 insertions(+) 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/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"))] From 501c17904a740d738945562e66949b2453160e3e Mon Sep 17 00:00:00 2001 From: Mohd Khairi Date: Sun, 2 Aug 2026 14:38:34 +0800 Subject: [PATCH 4/4] fix(crdt): write only the queue entries a flush actually changed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unsent-delta queue is append-only and each entry is stored under its own `crdt:delta:` key, but every flush re-wrote every entry and then wrote the same content again as the legacy bulk blob. The cost is the length of the queue, not what changed — and a replica with no Origin never has a delta acknowledged, so the queue only grows. Measured on a 26,409-object store: an idle flush, with no collection dirty and zero snapshot exports, still spent 1282 of its 1286 seconds inside one `batch_write` of 67,931 puts totalling 236 MB — 67,885 queue entries rewritten byte-identical, plus a 115 MB bulk blob duplicating them. That is 100% CPU with no reads and nothing changed to show for it, and it leaves 287 MB of superseded pages behind per pass. Track which entries are not known to match their stored form — the ones appended since the last flush, and any whose `seq` a send has since assigned — and write only those. The bulk blob duplicates them all, so it is rewritten only when the queue actually changed rather than every tick. Restoring from the bulk blob marks every entry unpersisted, since that path is the only copy they came from; restoring from the individual entries marks none, since each was just read from its own key. `crdt_delta_write_count()` exposes the count, so the property is a counter assertion rather than a timing. The bulk blob remains a full rewrite whenever the queue does change, which under sustained writes is every flush. It duplicates data that the per-entry keys already hold and that restore prefers over it, so it looks retireable — but that is an on-disk format decision, so it is left alone here. --- .../src/engine/crdt/engine/lifecycle.rs | 2 + nodedb-lite/src/engine/crdt/engine/mutate.rs | 2 + nodedb-lite/src/engine/crdt/engine/pending.rs | 41 +++++++++ nodedb-lite/src/engine/crdt/engine/persist.rs | 6 ++ nodedb-lite/src/engine/crdt/engine/types.rs | 14 ++++ nodedb-lite/src/nodedb/core/flush.rs | 50 ++++++++--- .../tests/crdt_flush_dirty_tracking.rs | 83 +++++++++++++++++++ 7 files changed, 186 insertions(+), 12 deletions(-) diff --git a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs b/nodedb-lite/src/engine/crdt/engine/lifecycle.rs index 8bec31d..88d58ef 100644 --- a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs +++ b/nodedb-lite/src/engine/crdt/engine/lifecycle.rs @@ -46,10 +46,12 @@ 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), }) } 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 66e2111..ce51b3e 100644 --- a/nodedb-lite/src/engine/crdt/engine/persist.rs +++ b/nodedb-lite/src/engine/crdt/engine/persist.rs @@ -27,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) => { @@ -68,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; } diff --git a/nodedb-lite/src/engine/crdt/engine/types.rs b/nodedb-lite/src/engine/crdt/engine/types.rs index 7eb6039..11b6174 100644 --- a/nodedb-lite/src/engine/crdt/engine/types.rs +++ b/nodedb-lite/src/engine/crdt/engine/types.rs @@ -103,6 +103,20 @@ pub struct CrdtEngine { /// 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 diff --git a/nodedb-lite/src/nodedb/core/flush.rs b/nodedb-lite/src/nodedb/core/flush.rs index ed54298..e72a06d 100644 --- a/nodedb-lite/src/nodedb/core/flush.rs +++ b/nodedb-lite/src/nodedb/core/flush.rs @@ -26,6 +26,15 @@ impl NodeDbLite { 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 @@ -94,7 +103,13 @@ impl NodeDbLite { // 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); @@ -102,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, @@ -111,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 { @@ -121,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 { @@ -321,7 +343,11 @@ impl NodeDbLite { // checkpoint accounting. Doing this only after the write means a failed // batch leaves every collection outstanding and the next flush retries // it. - self.crdt.lock_or_recover().mark_persisted(persisted); + { + 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"))] diff --git a/nodedb-lite/tests/crdt_flush_dirty_tracking.rs b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs index 74ec636..a03f287 100644 --- a/nodedb-lite/tests/crdt_flush_dirty_tracking.rs +++ b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs @@ -197,3 +197,86 @@ async fn updates_written_between_checkpoints_survive_reopen() { ); } } + +// --------------------------------------------------------------------------- +// 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 + ); +}