From a31b23fae657b9d9dd35271ac3e5e5214b81016f Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Sun, 2 Aug 2026 04:44:48 +0800 Subject: [PATCH 1/2] perf(crdt): resolve UNIQUE probes through the row container in hand field_value_exists formatted a collection/row/field path string and re-resolved it from the document root for every row, turning one UNIQUE probe into a per-row allocation plus a path parse and walk. Reach the row through the already-fetched collection container instead. --- nodedb-crdt/src/state/core.rs | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/nodedb-crdt/src/state/core.rs b/nodedb-crdt/src/state/core.rs index 3b9a83abe..51fc0a4b9 100644 --- a/nodedb-crdt/src/state/core.rs +++ b/nodedb-crdt/src/state/core.rs @@ -345,17 +345,20 @@ impl CrdtState { if exclude_row_id == Some(key.as_ref()) { continue; } - let path = format!("{collection}/{key}/{field}"); - if let Some(voc) = self.doc.get_by_str_path(&path) { - let field_val = match voc { - ValueOrContainer::Value(v) => v, - ValueOrContainer::Container(_) => { - continue; - } - }; - if &field_val == value { - return true; - } + // Reached through the row container rather than a built path. + // `get_by_str_path` formats a `collection/row/field` string and + // re-resolves it from the document root for every row, which turns + // one UNIQUE probe into a per-row allocation plus a path parse and + // walk — the dominant cost of validating a single row against a + // large collection. The containers here are already in hand. + let Some(ValueOrContainer::Container(loro::Container::Map(row))) = coll.get(&key) + else { + continue; + }; + if let Some(ValueOrContainer::Value(field_val)) = row.get(field) + && &field_val == value + { + return true; } } false From 245f74f3852cf376cf74190c0150565f274e64d7 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Sun, 2 Aug 2026 04:45:01 +0800 Subject: [PATCH 2/2] perf(crdt): retain validation candidate across delta applies Validating a delta means importing it into a copy of the collection so a rejection never touches authoritative state, but building that copy costs a full encode and decode. Keep the copy alive across a run of deltas (WAL replay, a committed batch, a sync burst) instead of paying that cost per delta, and drop it whenever it no longer matches the authoritative state it was cloned from (a local write, rollback, snapshot import, or a refused delta that left partial operations behind) so a stale or poisoned candidate is always rebuilt rather than reused. --- .../src/data/executor/core_loop/response.rs | 12 + nodedb/src/data/executor/wal_replay/crdt.rs | 6 + .../crdt/tenant_state/apply_validated.rs | 277 ++++++++++++++++-- nodedb/src/engine/crdt/tenant_state/core.rs | 10 + 4 files changed, 285 insertions(+), 20 deletions(-) diff --git a/nodedb/src/data/executor/core_loop/response.rs b/nodedb/src/data/executor/core_loop/response.rs index d3cfc8350..75017321e 100644 --- a/nodedb/src/data/executor/core_loop/response.rs +++ b/nodedb/src/data/executor/core_loop/response.rs @@ -193,4 +193,16 @@ impl CoreLoop { } Ok(self.crdt_engines.get_mut(&key).expect("just inserted")) } + + /// Release the per-collection validation candidates every CRDT engine on + /// this core is holding. + /// + /// Called when a run of delta applies ends. The candidates make a run cost + /// one collection copy instead of one per delta; keeping them past the run + /// would just be a second document per collection sitting idle. + pub(in crate::data::executor) fn release_crdt_apply_candidates(&mut self) { + for engine in self.crdt_engines.values_mut() { + engine.clear_apply_candidates(); + } + } } diff --git a/nodedb/src/data/executor/wal_replay/crdt.rs b/nodedb/src/data/executor/wal_replay/crdt.rs index abb4f925e..5eb470f80 100644 --- a/nodedb/src/data/executor/wal_replay/crdt.rs +++ b/nodedb/src/data/executor/wal_replay/crdt.rs @@ -345,6 +345,12 @@ impl CoreLoop { replayed += 1; } + // Replay is the longest run of deltas the engine ever sees, and each + // apply validates into a copy of its collection. Holding that copy + // across the run is what keeps recovery linear in the number of + // deltas instead of quadratic; it has no reason to outlive the run. + self.release_crdt_apply_candidates(); + if replayed > 0 { tracing::info!(core = self.core_id, replayed, "WAL CRDT replay complete"); } diff --git a/nodedb/src/engine/crdt/tenant_state/apply_validated.rs b/nodedb/src/engine/crdt/tenant_state/apply_validated.rs index 66a7a6fc0..56c6fcb83 100644 --- a/nodedb/src/engine/crdt/tenant_state/apply_validated.rs +++ b/nodedb/src/engine/crdt/tenant_state/apply_validated.rs @@ -121,26 +121,9 @@ impl TenantCrdtEngine { { return ValidatedApplyOutcome::Malformed; } - // Seeding the candidate replays this collection's own snapshot, exported - // on the line above. It is admitted as local: judged against the peer - // ceilings, a collection that outgrew them would fail to seed and every - // subsequent delta would be reported as `Malformed` — the collection - // silently unwritable, and the sender blamed for it. - let candidate = match self.collections.get(collection) { - Some(current) => { - let snapshot = match current.export_snapshot() { - Ok(snapshot) => snapshot, - Err(_) => return ValidatedApplyOutcome::Malformed, - }; - match CrdtState::from_local_snapshot(self.peer_id, &snapshot) { - Ok(state) => state, - Err(_) => return ValidatedApplyOutcome::Malformed, - } - } - None => match CrdtState::new(self.peer_id) { - Ok(state) => state, - Err(_) => return ValidatedApplyOutcome::Malformed, - }, + let candidate = match self.take_apply_candidate(collection) { + Ok(candidate) => candidate, + Err(()) => return ValidatedApplyOutcome::Malformed, }; let before = candidate.frontier(); let admission = match candidate.import(delta) { @@ -208,12 +191,96 @@ impl TenantCrdtEngine { return ValidatedApplyOutcome::Rejected(violation); } + // The candidate is now authoritative. Bring the doc it displaced up to + // the same version and keep it as the next delta's candidate: it starts + // identical to authoritative, so the next apply pays the delta rather + // than another full copy of the collection. + // + // Both docs held the same state and take the same delta, so the CRDT + // merge leaves them identical. If that second import somehow does not + // land, no candidate is retained and the next apply rebuilds one — + // slower, never wrong. + if let Some(previous) = previous + && previous.import(delta).is_ok() + { + self.apply_candidates + .insert(collection.to_owned(), previous); + } + ValidatedApplyOutcome::Clean { write_set, imported_ops: admission.new_operations, } } + /// A document identical to `collection`'s authoritative state, to import a + /// delta into before deciding whether to keep it. + /// + /// Reuses the candidate left behind by the previous apply when there is + /// one. Building it costs a full encode and decode of the collection — + /// Loro's own `fork` is implemented the same way, so there is no cheaper + /// copy available — and a run of deltas (WAL replay, a committed batch, a + /// sync burst) would otherwise pay that per delta rather than once. + /// + /// The candidate is only ever returned to the pool after a clean apply. A + /// delta that was refused for any reason leaves its operations in the + /// candidate — Loro buffers even causally-pending ones — so a poisoned + /// candidate is dropped rather than reused. + fn take_apply_candidate(&mut self, collection: &str) -> Result { + if let Some(candidate) = self.apply_candidates.remove(collection) { + // A candidate is only usable while it still matches the state it + // was cloned from. Anything else that touches this collection — a + // local write through `state_mut`, a transaction rollback, a + // snapshot import, a purge — leaves it behind, and installing a + // behind candidate as authoritative would silently discard those + // writes. Checking the frontier here means no other mutation site + // has to remember this one exists; a stale candidate is simply + // rebuilt. + let current_frontier = self.collections.get(collection).map(|s| s.frontier()); + if current_frontier.is_some_and(|frontier| frontier == candidate.frontier()) { + return Ok(candidate); + } + } + // The candidate becomes authoritative on a clean apply, so it must be + // born with the same derived peer id `state_mut` would have given the + // collection. The node's base peer id would mint operation ids that + // collide across collections — the silent row-drop `collection_peer_id` + // exists to prevent. + let peer_id = Self::collection_peer_id(self.peer_id, collection); + match self.collections.get(collection) { + // Its own snapshot, exported on the line above: admitted as local, + // because under the peer ceilings a collection that outgrew them + // would fail to seed and every subsequent delta would report + // `Malformed` — silently unwritable, with the sender blamed for it. + Some(current) => { + let snapshot = current.export_snapshot().map_err(|_| ())?; + CrdtState::from_local_snapshot(peer_id, &snapshot).map_err(|_| ()) + } + None => CrdtState::new(peer_id).map_err(|_| ()), + } + } + + /// Drop every retained apply candidate. + /// + /// Callers that apply a run of deltas — WAL replay, a committed batch — + /// call this when the run ends, so the second document per collection does + /// not outlive the work it accelerates. + pub fn clear_apply_candidates(&mut self) { + self.apply_candidates.clear(); + } + + /// How many collections are currently holding a retained candidate. + #[cfg(test)] + pub(crate) fn apply_candidate_count(&self) -> usize { + self.apply_candidates.len() + } + + /// The Loro peer id of a collection's authoritative document. + #[cfg(test)] + pub(crate) fn collection_peer_id_for_test(&self, collection: &str) -> Option { + self.collections.get(collection).map(|s| s.peer_id()) + } + /// Enqueue a rejected delta to the DLQ and translate the internal violation /// into the deterministic wire [`ViolationType`]. /// @@ -699,6 +766,176 @@ mod tests { ); } + /// The load-bearing safety property of retaining a candidate across + /// applies: the candidate becomes authoritative on a clean apply, so a + /// candidate that missed a local write would silently erase it. + /// + /// Nothing tells the candidate a local write happened — `doc_upsert` goes + /// straight to `state_mut`. It is caught by the candidate no longer + /// matching the state it was cloned from. + #[test] + fn a_local_write_between_applies_is_not_erased_by_the_candidate() { + let mut engine = unique_engine(); + + // The first apply creates the collection, so there is no displaced + // document to keep; the second leaves a candidate behind. + let delta_a = row_delta(2, "a", "a@y.com", "A"); + engine.apply_committed_delta_validated( + "users", + &delta_a, + nodedb_types::Surrogate::ZERO, + "a", + 2, + ); + let delta_seed = row_delta(9, "seed", "seed@y.com", "S"); + engine.apply_committed_delta_validated( + "users", + &delta_seed, + nodedb_types::Surrogate::ZERO, + "seed", + 9, + ); + assert_eq!(engine.apply_candidate_count(), 1, "candidate is retained"); + + // A local write the candidate knows nothing about. + engine + .doc_upsert( + "users", + "local", + &[("email", LoroValue::String("local@y.com".into()))], + ) + .expect("local write"); + + // Second apply must not install a candidate that predates it. + let delta_b = row_delta(3, "b", "b@y.com", "B"); + let outcome = engine.apply_committed_delta_validated( + "users", + &delta_b, + nodedb_types::Surrogate::ZERO, + "b", + 3, + ); + + assert!(matches!(outcome, ValidatedApplyOutcome::Clean { .. })); + assert!( + engine.row_exists("users", "local"), + "the local write was erased by a stale validation candidate" + ); + assert!(engine.row_exists("users", "a")); + assert!(engine.row_exists("users", "b")); + } + + /// A refused delta leaves its operations inside the candidate — Loro + /// buffers even causally-pending ones — so the candidate must be discarded + /// rather than handed to the next delta. + #[test] + fn a_rejected_delta_does_not_poison_the_next_apply() { + let mut engine = unique_engine(); + let seed = row_delta(2, "a", "x@y.com", "A"); + engine.apply_committed_delta_validated( + "users", + &seed, + nodedb_types::Surrogate::ZERO, + "a", + 2, + ); + + // Rejected: reuses row a's email under a strict UNIQUE policy. + let dup = row_delta(3, "b", "x@y.com", "B"); + let rejected = engine.apply_committed_delta_validated( + "users", + &dup, + nodedb_types::Surrogate::ZERO, + "b", + 3, + ); + assert!(matches!(rejected, ValidatedApplyOutcome::Rejected(_))); + assert_eq!( + engine.apply_candidate_count(), + 0, + "a candidate holding refused operations must not be kept" + ); + + // The next delta must see a clean base: neither row b nor its email. + let next = row_delta(4, "c", "c@y.com", "C"); + let outcome = engine.apply_committed_delta_validated( + "users", + &next, + nodedb_types::Surrogate::ZERO, + "c", + 4, + ); + assert!(matches!(outcome, ValidatedApplyOutcome::Clean { .. })); + assert!(engine.row_exists("users", "c")); + assert!( + !engine.row_exists("users", "b"), + "the refused row resurfaced through a reused candidate" + ); + } + + /// A run of applies reuses one candidate rather than copying the + /// collection per delta, and releasing it is what stops the second + /// document outliving the run. + #[test] + fn a_run_of_applies_reuses_one_candidate() { + let mut engine = unique_engine(); + for i in 0..8 { + let delta = row_delta( + 10 + i, + &format!("r{i}"), + &format!("r{i}@y.com"), + &format!("R{i}"), + ); + let outcome = engine.apply_committed_delta_validated( + "users", + &delta, + nodedb_types::Surrogate::ZERO, + &format!("r{i}"), + 10 + i, + ); + assert!( + matches!(outcome, ValidatedApplyOutcome::Clean { .. }), + "delta {i} should apply cleanly, got {outcome:?}" + ); + } + for i in 0..8 { + assert!(engine.row_exists("users", &format!("r{i}"))); + } + assert_eq!( + engine.apply_candidate_count(), + 1, + "one candidate serves the whole run" + ); + + engine.clear_apply_candidates(); + assert_eq!(engine.apply_candidate_count(), 0); + } + + /// A collection first created by a delta apply must get the same derived + /// peer id `state_mut` would have given it. The node's base peer id mints + /// operation identities that collide across collections, which is the + /// silent row-drop `collection_peer_id` exists to prevent. + #[test] + fn a_collection_created_by_apply_gets_the_derived_peer_id() { + let mut engine = unique_engine(); + let delta = row_delta(5, "a", "a@y.com", "A"); + engine.apply_committed_delta_validated( + "users", + &delta, + nodedb_types::Surrogate::ZERO, + "a", + 5, + ); + + let expected = TenantCrdtEngine::collection_peer_id(engine.peer_id(), "users"); + assert_eq!( + engine.collection_peer_id_for_test("users"), + Some(expected), + "a collection born from a delta apply must carry the derived peer \ + id, not the node's base id" + ); + } + #[test] fn corrupt_delta_is_malformed() { let mut engine = unique_engine(); diff --git a/nodedb/src/engine/crdt/tenant_state/core.rs b/nodedb/src/engine/crdt/tenant_state/core.rs index 51a04730c..fbd1c2703 100644 --- a/nodedb/src/engine/crdt/tenant_state/core.rs +++ b/nodedb/src/engine/crdt/tenant_state/core.rs @@ -93,6 +93,15 @@ pub struct TenantCrdtEngine { /// stale set re-proposed at a higher data-log index can never clobber a /// newer one. Collections absent from the map are treated as version `0`. pub(super) constraint_versions: HashMap, + + /// Per-collection validation candidate retained between applies. + /// + /// A delta is validated by importing it into a copy of the collection, so + /// a rejection can be dropped without touching authoritative state. Copying + /// costs a full encode and decode of the collection, so the copy is kept + /// alive across a run of deltas and rebuilt only when a delta is refused. + /// Cleared by `clear_apply_candidates` when the run ends. + pub(super) apply_candidates: HashMap, } impl TenantCrdtEngine { @@ -109,6 +118,7 @@ impl TenantCrdtEngine { validator: Validator::new(constraints, 1000), collections: HashMap::new(), constraint_versions: HashMap::new(), + apply_candidates: HashMap::new(), }) }