From e599d88f4cee999406ee03b9d65bad5c96d0ff1d Mon Sep 17 00:00:00 2001 From: Mohd Khairi Date: Sat, 1 Aug 2026 19:04:58 +0800 Subject: [PATCH 1/5] 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/5] 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/5] 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/5] 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 + ); +} From 79f387a487fdb8b8c15e4c6174c56161d779a083 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Mon, 3 Aug 2026 12:11:42 +0800 Subject: [PATCH 5/5] fix(crdt): guard flush acknowledgement against concurrent state changes A flush plans and exports CRDT writes under the engine lock, then releases it while the batch commits before re-taking it to record what is durable. Concurrent flushes, deltas queued or resequenced in that window, and compactions landing mid-commit could all be acknowledged incorrectly by the old membership/frontier-only checks, either stranding writes in memory forever or marking a compacted document as persisted in a form it no longer has. Serialize flushes with a dedicated lock, stamp each pending delta with a revision so acknowledgement can tell which batch it was actually written in, and track a per-collection compaction epoch so a write is only recorded once its epoch still matches. Restore also now discards CRDT updates whose base snapshot is missing instead of failing every subsequent open on them, and logs rather than silently swallows a failed corrupted-snapshot delete. --- .../src/engine/crdt/engine/checkpoint.rs | 34 +++++ .../src/engine/crdt/engine/flush_ack_tests.rs | 140 ++++++++++++++++++ .../src/engine/crdt/engine/lifecycle.rs | 15 +- nodedb-lite/src/engine/crdt/engine/mod.rs | 2 + nodedb-lite/src/engine/crdt/engine/mutate.rs | 4 +- nodedb-lite/src/engine/crdt/engine/pending.rs | 71 ++++++--- nodedb-lite/src/engine/crdt/engine/persist.rs | 5 +- nodedb-lite/src/engine/crdt/engine/types.rs | 33 ++++- nodedb-lite/src/nodedb/core/flush.rs | 18 ++- .../src/nodedb/core/open/constructors.rs | 1 + .../src/nodedb/core/open/restore/crdt.rs | 53 ++++++- nodedb-lite/src/nodedb/core/types.rs | 11 ++ .../tests/crdt_flush_dirty_tracking.rs | 81 +++++++++- 13 files changed, 428 insertions(+), 40 deletions(-) create mode 100644 nodedb-lite/src/engine/crdt/engine/flush_ack_tests.rs diff --git a/nodedb-lite/src/engine/crdt/engine/checkpoint.rs b/nodedb-lite/src/engine/crdt/engine/checkpoint.rs index 8544c50..8fe65bf 100644 --- a/nodedb-lite/src/engine/crdt/engine/checkpoint.rs +++ b/nodedb-lite/src/engine/crdt/engine/checkpoint.rs @@ -31,6 +31,11 @@ pub struct CrdtWrite { /// Frontier the payload was exported at. Report it back through /// [`CrdtEngine::mark_persisted`] once the write has committed. pub version: loro::VersionVector, + /// Compaction epoch the collection was at when this was planned. A + /// compaction landing before the write commits makes the payload stale in + /// a way the frontier cannot show, since compaction leaves the frontier + /// where it was; the epoch is what distinguishes the two. + pub epoch: u64, } /// Which of the two shapes a [`CrdtWrite`] carries. @@ -47,10 +52,16 @@ pub enum CrdtWriteKind { /// A committed [`CrdtWrite`], reported back so the engine can advance its /// bookkeeping. Carries the payload's length rather than the payload. pub struct CrdtPersisted { + /// Collection the committed write belongs to. pub collection: String, + /// Whether it was a fresh base snapshot or an update on top of one. pub kind: CrdtWriteKind, + /// Length of the payload that was written. pub bytes: usize, + /// Frontier the payload was exported at. pub version: loro::VersionVector, + /// Compaction epoch the write was planned at — see [`CrdtWrite::epoch`]. + pub epoch: u64, } impl CrdtWrite { @@ -61,6 +72,7 @@ impl CrdtWrite { kind: self.kind, bytes: self.bytes.len(), version: self.version.clone(), + epoch: self.epoch, } } } @@ -111,16 +123,38 @@ impl CrdtEngine { kind, bytes, version, + epoch: self.state_epoch(collection), }); } Ok(out) } + /// Compaction epoch a collection is currently at. Absent means never + /// compacted, which is epoch zero. + pub(in crate::engine::crdt) fn state_epoch(&self, collection: &str) -> u64 { + self.state_epochs.get(collection).copied().unwrap_or(0) + } + + /// Record that a collection's document has been structurally rewritten, so + /// any write planned against the previous form is no longer describable by + /// its frontier. See [`CrdtEngine::state_epochs`]. + pub(in crate::engine::crdt) fn advance_state_epoch(&mut self, collection: &str) { + *self.state_epochs.entry(collection.to_string()).or_insert(0) += 1; + } + /// Advance the bookkeeping for writes that are now durable. /// + /// A write whose collection was compacted between the plan and the commit + /// is discarded rather than recorded: the bytes on disk describe a document + /// that no longer exists, and the marks compaction dropped must stay + /// dropped so the next flush writes a fresh checkpoint. + /// /// Call only after the batch has committed — see [`Self::plan_persistence`]. pub fn mark_persisted(&mut self, persisted: impl IntoIterator) { for entry in persisted { + if self.state_epoch(&entry.collection) != entry.epoch { + continue; + } match entry.kind { CrdtWriteKind::Checkpoint { .. } => { self.checkpoint_bytes diff --git a/nodedb-lite/src/engine/crdt/engine/flush_ack_tests.rs b/nodedb-lite/src/engine/crdt/engine/flush_ack_tests.rs new file mode 100644 index 0000000..5c9086a --- /dev/null +++ b/nodedb-lite/src/engine/crdt/engine/flush_ack_tests.rs @@ -0,0 +1,140 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! What a flush may still claim once it has let go of the engine. +//! +//! A flush plans and exports under the engine lock, releases it while its batch +//! commits, then re-takes it to record what is now durable. Anything that +//! happens in that window changed state the batch does not carry, so the +//! acknowledgement has to be able to tell the two apart. These assert that it +//! does, for each way the window can be used. + +use loro::LoroValue; + +use super::types::CrdtEngine; + +/// A write queued while the batch was committing was not in it. Retiring its +/// dirty mark strands it: an append-only queue is only revisited when it +/// changes, so the entry would sit in memory until the process ended and the +/// write it carries would never reach Origin. +#[test] +fn a_delta_queued_during_a_flush_is_not_retired_unwritten() { + let mut engine = CrdtEngine::new(1).unwrap(); + engine + .upsert("users", "u1", &[("n", LoroValue::I64(1))]) + .unwrap(); + + let planned: Vec<(u64, u64)> = engine + .pending_deltas_needing_write() + .map(|(delta, revision)| (delta.mutation_id, revision)) + .collect(); + assert_eq!(planned.len(), 1); + + // The batch is committing. A second write lands before it is acknowledged. + let queued_during = engine + .upsert("users", "u2", &[("n", LoroValue::I64(2))]) + .unwrap(); + + engine.mark_pending_deltas_persisted(planned); + + let still_dirty: Vec = engine + .pending_deltas_needing_write() + .map(|(delta, _)| delta.mutation_id) + .collect(); + assert_eq!( + still_dirty, + vec![queued_during], + "the entry queued while the batch was in flight was not in it, so it must still be \ + waiting to be written" + ); + assert_eq!( + engine.pending_delta_write_count(), + 1, + "only the entry the batch actually carried counts as written" + ); +} + +/// The same window catches an *edit*, not only an insertion: assigning a stream +/// seq rewrites an entry that is already on disk, so its stored form goes stale +/// and it has to be written again. +#[test] +fn a_delta_resequenced_during_a_flush_stays_dirty() { + let mut engine = CrdtEngine::new(1).unwrap(); + let mid = engine + .upsert("users", "u1", &[("n", LoroValue::I64(1))]) + .unwrap(); + + let planned: Vec<(u64, u64)> = engine + .pending_deltas_needing_write() + .map(|(delta, revision)| (delta.mutation_id, revision)) + .collect(); + + // The batch is committing. The delta is sent and is assigned its seq. + engine.set_pending_delta_seq(mid, 7); + + engine.mark_pending_deltas_persisted(planned); + + assert!( + engine.has_unpersisted_deltas(), + "the stored entry carries seq 0 while the queue carries seq 7 — a resend after a \ + restart would use the wrong seq, so it must be rewritten" + ); +} + +/// An acknowledgement that arrives with nothing to report leaves the queue +/// alone: replaying one must not resurrect an entry or double-count a write. +#[test] +fn acknowledging_the_same_batch_twice_changes_nothing() { + let mut engine = CrdtEngine::new(1).unwrap(); + engine + .upsert("users", "u1", &[("n", LoroValue::I64(1))]) + .unwrap(); + + let planned: Vec<(u64, u64)> = engine + .pending_deltas_needing_write() + .map(|(delta, revision)| (delta.mutation_id, revision)) + .collect(); + + engine.mark_pending_deltas_persisted(planned.clone()); + engine.mark_pending_deltas_persisted(planned); + + assert!(!engine.has_unpersisted_deltas()); + assert_eq!( + engine.pending_delta_write_count(), + 1, + "the second report describes the same write, not another one" + ); +} + +/// Compaction replaces a document without moving its frontier, so a flush that +/// planned against the previous form cannot be recognised as stale by frontier +/// alone. A compaction landing while the batch commits must leave the +/// collection needing a fresh checkpoint. +#[test] +fn a_compaction_during_a_flush_is_not_undone_by_its_acknowledgement() { + let mut engine = CrdtEngine::new(1).unwrap(); + engine + .upsert("users", "u1", &[("n", LoroValue::I64(1))]) + .unwrap(); + + let plan = engine.plan_persistence().unwrap(); + assert_eq!(plan.len(), 1); + let persisted: Vec<_> = plan.iter().map(|write| write.persisted()).collect(); + + // The batch is committing. Compaction discards the history underneath it. + engine.compact_history().unwrap(); + + engine.mark_persisted(persisted); + + let replan = engine.plan_persistence().unwrap(); + assert_eq!( + replan.len(), + 1, + "the compacted document is not what the in-flight batch wrote, so it must still be \ + planned for" + ); + assert!( + matches!(replan[0].kind, super::CrdtWriteKind::Checkpoint { .. }), + "and as a fresh checkpoint — an update exported from the discarded history does not \ + apply to the base on disk" + ); +} diff --git a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs b/nodedb-lite/src/engine/crdt/engine/lifecycle.rs index 88d58ef..2f32e80 100644 --- a/nodedb-lite/src/engine/crdt/engine/lifecycle.rs +++ b/nodedb-lite/src/engine/crdt/engine/lifecycle.rs @@ -46,12 +46,14 @@ impl CrdtEngine { policies: nodedb_crdt::PolicyRegistry::new(), registered_collections: std::collections::HashSet::new(), deferred: Vec::new(), - unpersisted_deltas: std::collections::HashSet::new(), + unpersisted_deltas: HashMap::new(), + delta_revision: 0, flushed_versions: HashMap::new(), checkpoint_bytes: HashMap::new(), delta_bytes: HashMap::new(), next_delta_seq: HashMap::new(), - delta_writes: AtomicU64::new(0), + state_epochs: HashMap::new(), + delta_writes: 0, snapshot_exports: AtomicU64::new(0), }) } @@ -232,9 +234,17 @@ impl CrdtEngine { // 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. + // + // Advancing the epoch is what keeps a flush that is committing right + // now from putting the marks back: its writes were exported from the + // document this call just replaced. self.flushed_versions.clear(); self.checkpoint_bytes.clear(); self.delta_bytes.clear(); + let compacted: Vec = self.states.keys().cloned().collect(); + for collection in compacted { + self.advance_state_epoch(&collection); + } Ok(()) } @@ -306,6 +316,7 @@ impl CrdtEngine { self.flushed_versions.remove(collection); self.checkpoint_bytes.remove(collection); self.delta_bytes.remove(collection); + self.advance_state_epoch(collection); Ok(()) } } diff --git a/nodedb-lite/src/engine/crdt/engine/mod.rs b/nodedb-lite/src/engine/crdt/engine/mod.rs index ca793f2..e637455 100644 --- a/nodedb-lite/src/engine/crdt/engine/mod.rs +++ b/nodedb-lite/src/engine/crdt/engine/mod.rs @@ -23,6 +23,8 @@ mod read; mod rotate; pub mod types; +#[cfg(test)] +mod flush_ack_tests; #[cfg(test)] mod tests; diff --git a/nodedb-lite/src/engine/crdt/engine/mutate.rs b/nodedb-lite/src/engine/crdt/engine/mutate.rs index a2d7bbb..884d25e 100644 --- a/nodedb-lite/src/engine/crdt/engine/mutate.rs +++ b/nodedb-lite/src/engine/crdt/engine/mutate.rs @@ -183,7 +183,7 @@ impl CrdtEngine { delta_bytes, seq: 0, }); - self.unpersisted_deltas.insert(mutation_id); + self.mark_delta_unpersisted(mutation_id); } Ok(count) @@ -256,7 +256,7 @@ impl CrdtEngine { delta_bytes, seq: 0, }); - self.unpersisted_deltas.insert(mutation_id); + self.mark_delta_unpersisted(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 e530ca6..c73577c 100644 --- a/nodedb-lite/src/engine/crdt/engine/pending.rs +++ b/nodedb-lite/src/engine/crdt/engine/pending.rs @@ -27,26 +27,38 @@ impl CrdtEngine { self.unpersisted_deltas.clear(); } - /// The pending deltas whose stored form may not match the queue. + /// Mark a queue entry as not matching its stored form, stamping it with a + /// fresh revision. + /// + /// Every path that adds an entry or edits one in place goes through here, + /// so an edit that lands while a flush is committing is distinguishable + /// from the state that flush actually wrote. + pub(in crate::engine::crdt) fn mark_delta_unpersisted(&mut self, mutation_id: u64) { + self.delta_revision += 1; + self.unpersisted_deltas + .insert(mutation_id, self.delta_revision); + } + + /// The pending deltas whose stored form may not match the queue, each with + /// the revision to report back once it is durable. /// /// 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); - }) + /// committed, passing the revision handed out here — not the entry's + /// current one, which may have moved on since. + pub fn pending_deltas_needing_write(&self) -> impl Iterator { + self.pending_deltas.iter().filter_map(|d| { + self.unpersisted_deltas + .get(&d.mutation_id) + .map(|&revision| (d, revision)) + }) } - /// Number of queue entries handed out for writing since this engine was - /// created. + /// Number of queue entries written and acknowledged durable since this + /// engine was created. pub fn pending_delta_write_count(&self) -> u64 { - self.delta_writes.load(std::sync::atomic::Ordering::Relaxed) + self.delta_writes } /// Whether any pending delta needs writing. @@ -54,12 +66,22 @@ impl CrdtEngine { !self.unpersisted_deltas.is_empty() } - /// Record that every pending delta is now stored as written. + /// Retire the dirty marks for queue entries that are now durable. /// - /// 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(); + /// Each `(mutation_id, revision)` pair must be one handed out by + /// [`Self::pending_deltas_needing_write`] for the batch that has just + /// committed. An entry whose revision has moved on since was added or + /// edited while that batch was in flight and so was never in it; its mark + /// stays, and the next flush writes it. + /// + /// Call only after the batch has committed. + pub fn mark_pending_deltas_persisted(&mut self, written: impl IntoIterator) { + for (mutation_id, revision) in written { + if self.unpersisted_deltas.get(&mutation_id) == Some(&revision) { + self.unpersisted_deltas.remove(&mutation_id); + self.delta_writes += 1; + } + } } /// Drop a single pending delta by `mutation_id` without touching CRDT state. @@ -79,15 +101,20 @@ impl CrdtEngine { /// the call is a no-op — the existing seq is reused on reconnect re-sends /// so Origin can deduplicate rather than double-apply. pub fn set_pending_delta_seq(&mut self, mutation_id: u64, seq: u64) { - if let Some(d) = self + let assigned = match self .pending_deltas .iter_mut() .find(|d| d.mutation_id == mutation_id) - && d.seq == 0 { - d.seq = seq; + Some(d) if d.seq == 0 => { + d.seq = seq; + true + } + _ => false, + }; + if assigned { // The stored entry now carries a stale seq. - self.unpersisted_deltas.insert(mutation_id); + self.mark_delta_unpersisted(mutation_id); } } diff --git a/nodedb-lite/src/engine/crdt/engine/persist.rs b/nodedb-lite/src/engine/crdt/engine/persist.rs index ce51b3e..1aad0e7 100644 --- a/nodedb-lite/src/engine/crdt/engine/persist.rs +++ b/nodedb-lite/src/engine/crdt/engine/persist.rs @@ -29,7 +29,10 @@ impl CrdtEngine { 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.unpersisted_deltas.clear(); + for delta in &deltas { + self.mark_delta_unpersisted(delta.mutation_id); + } self.pending_deltas = deltas; } Err(e) => { diff --git a/nodedb-lite/src/engine/crdt/engine/types.rs b/nodedb-lite/src/engine/crdt/engine/types.rs index 11b6174..9f43819 100644 --- a/nodedb-lite/src/engine/crdt/engine/types.rs +++ b/nodedb-lite/src/engine/crdt/engine/types.rs @@ -103,7 +103,22 @@ 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. + /// How many times each collection's document has been structurally + /// rewritten underneath its persisted form — that is, compacted. + /// + /// A flush plans and exports under the engine lock, then releases it while + /// its batch commits. Compaction runs on its own timer and takes the same + /// lock, so it can land in that window. It leaves the frontier where it was + /// but discards the history behind it, which invalidates both the base on + /// disk and any update exported from the old frontier — precisely what + /// [`CrdtEngine::compact_history`] drops the marks for. Acknowledging the + /// in-flight write by frontier alone would put those marks straight back, + /// leaving the collection recorded as persisted in a form that no longer + /// describes it. Each write carries the epoch it was planned at and is + /// applied only while that still matches. + pub(in crate::engine::crdt) state_epochs: HashMap, + /// Pending deltas whose stored form is not known to match the queue, each + /// with the revision of the entry that made it dirty. /// /// 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 @@ -111,12 +126,22 @@ pub struct CrdtEngine { /// 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. + /// + /// The revision is what makes the acknowledgement safe across the flush's + /// own await: an entry queued — or re-dirtied by `set_pending_delta_seq` — + /// while the batch was committing was never in that batch, and clearing the + /// mark by membership alone would retire it unwritten. Nothing brings it + /// back, because an append-only queue is only ever revisited when it + /// changes, so the entry would stay in memory until the process ended and + /// the write it carries would never reach Origin. + pub(in crate::engine::crdt) unpersisted_deltas: HashMap, + /// Revision stamped on the next queue entry to be added or edited. + pub(in crate::engine::crdt) delta_revision: u64, + /// Number of queue entries written and acknowledged durable. /// /// 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, + pub(in crate::engine::crdt) delta_writes: u64, /// 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 e72a06d..708971f 100644 --- a/nodedb-lite/src/nodedb/core/flush.rs +++ b/nodedb-lite/src/nodedb/core/flush.rs @@ -37,6 +37,11 @@ impl NodeDbLite { /// Persist all in-memory state to storage (call before shutdown). pub async fn flush(&self) -> NodeDbResult<()> { + // One flush at a time: the CRDT update sequence is allocated under the + // `crdt` guard but committed after it is released, so concurrent + // flushes would hand out the same numbers. See `flush_lock`. + let _flush_guard = self.flush_lock.lock().await; + // Drain the buffered KV writes first — they have their own batch-commit // path. Without this, `flush()` (and the auto-flush timer) would not // persist KV `put`s, contradicting "persist all in-memory state". @@ -68,7 +73,7 @@ impl NodeDbLite { // 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 (persisted, written_deltas) = { let crdt = self.crdt.lock_or_recover(); let plan = crdt.plan_persistence().map_err(NodeDbError::storage)?; let mut persisted = Vec::with_capacity(plan.len()); @@ -128,9 +133,14 @@ impl NodeDbLite { } } - for delta in crdt.pending_deltas_needing_write() { + // The revision each entry was written at travels with it: the + // acknowledgement below happens after an await, and an entry queued + // or re-sequenced in that window was never in this batch. + let mut written_deltas: Vec<(u64, u64)> = Vec::new(); + for (delta, revision) 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)?; + written_deltas.push((delta.mutation_id, revision)); ops.push(WriteOp::Put { ns: Namespace::Crdt, key, @@ -160,7 +170,7 @@ impl NodeDbLite { value: max_mid.to_le_bytes().to_vec(), }); - persisted + (persisted, written_deltas) }; // ── Persist per-collection CSR indices ── @@ -346,7 +356,7 @@ impl NodeDbLite { { let mut crdt = self.crdt.lock_or_recover(); crdt.mark_persisted(persisted); - crdt.mark_pending_deltas_persisted(); + crdt.mark_pending_deltas_persisted(written_deltas); } // ── Write HNSW vector segments to pagedb (native PagedbStorage only) ── diff --git a/nodedb-lite/src/nodedb/core/open/constructors.rs b/nodedb-lite/src/nodedb/core/open/constructors.rs index 8feb5e9..49df529 100644 --- a/nodedb-lite/src/nodedb/core/open/constructors.rs +++ b/nodedb-lite/src/nodedb/core/open/constructors.rs @@ -245,6 +245,7 @@ impl NodeDbLite { timeseries_outbound: timeseries_outbound_init, identity: Mutex::new(lite_identity), identity_change: tokio::sync::Mutex::new(()), + flush_lock: tokio::sync::Mutex::new(()), sync_enabled, kv_cache: Mutex::new(lru::LruCache::new(kv_cache_capacity)), kv_write_buf: Mutex::new(KvWriteBuffer { diff --git a/nodedb-lite/src/nodedb/core/open/restore/crdt.rs b/nodedb-lite/src/nodedb/core/open/restore/crdt.rs index 55365b3..57738a1 100644 --- a/nodedb-lite/src/nodedb/core/open/restore/crdt.rs +++ b/nodedb-lite/src/nodedb/core/open/restore/crdt.rs @@ -80,8 +80,18 @@ impl NodeDbLite { "CRDT snapshot CRC32C mismatch — discarding corrupted snapshot for this \ collection. A full re-sync from Origin is needed for it." ); - // Delete the corrupted snapshot so we don't re-read it. - let _ = storage.delete(Namespace::LoroState, key).await; + // Delete the corrupted snapshot so we don't re-read it. A + // failed delete leaves it to be re-read and re-reported on + // the next open, which is recoverable — but it must not be + // silent, or the collection looks cleanly dropped. + if let Err(e) = storage.delete(Namespace::LoroState, key).await { + tracing::error!( + collection = %collection, + error = %e, + "failed to delete the corrupted CRDT snapshot; it will be \ + re-read and discarded again on the next open" + ); + } } } } @@ -90,8 +100,19 @@ impl NodeDbLite { // 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. + // Skipping an update whose base is present would silently roll that + // collection back to its last checkpoint, so a corrupt one is an error + // rather than a warning. + // + // An update whose base is *absent* is the opposite case. Every update is + // exported as the operations since a base that was written in an earlier + // batch, so without that base its causal predecessors are missing and + // Loro buffers it as pending — which `import_local` reports as an error. + // Replaying one would therefore fail the whole open, and since nothing + // removes these keys, it would fail every subsequent open too: a + // collection whose snapshot was discarded one line above would take the + // entire store down with it, permanently, instead of being the isolated + // re-sync the discard is there to make it. They are deleted instead. let mut delta_bytes: std::collections::HashMap = std::collections::HashMap::new(); let mut next_delta_seq: std::collections::HashMap = @@ -99,6 +120,7 @@ impl NodeDbLite { let delta_entries = storage .scan_prefix(Namespace::LoroState, CrdtEngine::state_delta_key_prefix()) .await?; + let mut orphaned_keys: Vec> = Vec::new(); for (key, envelope) in &delta_entries { let Some((collection, seq)) = CrdtEngine::state_delta_from_key(key) else { tracing::error!( @@ -107,6 +129,18 @@ impl NodeDbLite { ); continue; }; + if !base_bytes.contains_key(collection) { + tracing::error!( + collection = %collection, + seq, + "CRDT update has no base snapshot to apply to — discarding it. Its base \ + was corrupt or missing, so the operations it carries have no causal \ + predecessors and cannot be replayed. A full re-sync from Origin is \ + needed for this collection." + ); + orphaned_keys.push(key.clone()); + 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 \ @@ -125,6 +159,17 @@ impl NodeDbLite { )) })?; } + for key in orphaned_keys { + // A failed delete is recoverable — the entry is skipped by the same + // base-absent check on the next open — but never silent. + if let Err(e) = storage.delete(Namespace::LoroState, &key).await { + tracing::error!( + error = %e, + "failed to delete an orphaned CRDT update; it will be skipped and \ + reported again on the next open" + ); + } + } // Seed the checkpoint accounting from what was on disk, so the first // flush after open does not rewrite a base that is already current. diff --git a/nodedb-lite/src/nodedb/core/types.rs b/nodedb-lite/src/nodedb/core/types.rs index f0ebf98..7e34762 100644 --- a/nodedb-lite/src/nodedb/core/types.rs +++ b/nodedb-lite/src/nodedb/core/types.rs @@ -127,6 +127,17 @@ pub struct NodeDbLite { /// overwriting the first — leaving the documents re-authored under an id /// the persisted record no longer names. pub(crate) identity_change: tokio::sync::Mutex<()>, + /// Serializes `flush` calls against each other. + /// + /// A flush plans its CRDT writes under the `crdt` guard, releases it while + /// the batch commits, then re-takes it to record what is now durable — a + /// span no `std::sync` guard can cover. Two flushes in flight would each + /// plan from the same bookkeeping and hand out the same update sequence, + /// so one batch's checkpoint could retire entries the other had just + /// written under those numbers, leaving updates on disk that no later + /// checkpoint knows to delete. Serializing them keeps the sequence a + /// single writer's to allocate. + pub(crate) flush_lock: tokio::sync::Mutex<()>, /// When `false`, KV operations go directly to storage, bypassing Loro. pub(crate) sync_enabled: bool, /// Buffered KV writes awaiting batch commit to storage. diff --git a/nodedb-lite/tests/crdt_flush_dirty_tracking.rs b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs index a03f287..2fc9cb9 100644 --- a/nodedb-lite/tests/crdt_flush_dirty_tracking.rs +++ b/nodedb-lite/tests/crdt_flush_dirty_tracking.rs @@ -14,7 +14,9 @@ use std::sync::Arc; use nodedb_client::NodeDb; -use nodedb_lite::{Encryption, LiteConfig, NodeDbLite, PagedbStorageDefault}; +use nodedb_lite::engine::crdt::CrdtEngine; +use nodedb_lite::{Encryption, LiteConfig, NodeDbLite, PagedbStorageDefault, StorageEngine}; +use nodedb_types::Namespace; use nodedb_types::document::Document; use nodedb_types::value::Value; @@ -280,3 +282,80 @@ async fn queued_deltas_survive_reopen() { dump.storage_counts ); } + +// --------------------------------------------------------------------------- +// a_corrupt_base_does_not_strand_its_updates +// --------------------------------------------------------------------------- + +/// A collection whose base snapshot fails its CRC is dropped on open so the +/// rest of the store still comes up. Its updates must be dropped with it. +/// +/// Each update carries only the operations since that base, so without the base +/// its causal predecessors are missing, Loro buffers it as pending, and the +/// import reports an error. Replaying one would therefore fail the open — and +/// because nothing else removes these keys, it would fail every open after it +/// too, turning an isolated re-sync into a store that never opens again. +#[tokio::test] +async fn a_corrupt_base_does_not_strand_its_updates() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("corrupt_base.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"); + for i in 1..4 { + put_note(&db, &format!("note{i}"), &format!("update {i}")).await; + db.flush().await.expect("flush an update"); + } + } + + // Corrupt the base the updates were exported against. + { + let storage = PagedbStorageDefault::open(&path, Encryption::Plaintext) + .await + .expect("open storage"); + let updates = storage + .scan_prefix(Namespace::LoroState, CrdtEngine::state_delta_key_prefix()) + .await + .expect("scan updates"); + assert!( + !updates.is_empty(), + "this test only means something while there are updates sitting on top of the base" + ); + + let key = CrdtEngine::snapshot_key_for("bt_notes"); + let mut envelope = storage + .get(Namespace::LoroState, &key) + .await + .expect("get base") + .expect("the base snapshot must be on disk"); + let last = envelope.len() - 1; + envelope[last] ^= 0xff; + storage + .put(Namespace::LoroState, &key, &envelope) + .await + .expect("write the corrupted base"); + } + + // Opening must succeed rather than fail on updates it cannot apply. + drop(open_manual_flush(&path).await); + + let storage = PagedbStorageDefault::open(&path, Encryption::Plaintext) + .await + .expect("reopen storage"); + let leftover = storage + .scan_prefix(Namespace::LoroState, CrdtEngine::state_delta_key_prefix()) + .await + .expect("scan updates after recovery"); + assert!( + leftover.is_empty(), + "an update whose base is gone can never be applied; leaving it on disk makes every \ + future open fail on it — {} remain", + leftover.len() + ); +}