Skip to content

Commit e7c2c68

Browse files
committed
fix(crdt): admit local-snapshot reloads without the peer ceilings
Rebuilding a CrdtState from a snapshot this process exported itself went through the plain constructor plus a raw import, which enforced the peer size ceilings meant for untrusted replicated deltas. Once a collection outgrew them, replicated-apply seeding and transaction rollback restore would both fail on data the process wrote itself: apply would misreport every subsequent delta as malformed, and rollback would leave the transaction driver with a failure it can't act on. Add CrdtState::from_local_snapshot, which imports through the same admit_local_import path already used by compaction, and route apply_validated's seeding and snapshot_restore's rollback through it. Share the compaction swap itself via a new compact_to_frontiers helper, and move the document plus its derived memory-estimate cache into a DocumentCell so replacing the document (compaction) can no longer leave a stale cache describing bytes that no longer exist.
1 parent 184114a commit e7c2c68

9 files changed

Lines changed: 259 additions & 85 deletions

File tree

‎nodedb-crdt/src/state/core.rs‎

Lines changed: 31 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
//! CrdtState core: document handle, row CRUD, uniqueness probes.
44
5-
use std::cell::{Cell, RefCell};
5+
use std::cell::Cell;
66
use std::collections::HashSet;
77
use std::marker::PhantomData;
88

@@ -12,6 +12,8 @@ use crate::error::{CrdtError, Result};
1212
use crate::row_lookup::RowLookup;
1313
use crate::validator::bitemporal::{VALID_UNTIL, VALID_UNTIL_OPEN};
1414

15+
use super::document_cell::DocumentCell;
16+
1517
/// A row is live when its `_ts_valid_until` field is absent, null, or the
1618
/// open sentinel (`i64::MAX`). Rows with any finite `_ts_valid_until` are
1719
/// treated as superseded, independent of wall-clock time — the write path
@@ -41,34 +43,49 @@ fn key_is_container(row: &LoroMap, key: &str) -> bool {
4143
pub struct CrdtState {
4244
/// Kept private to this state module so callers cannot clone or share a
4345
/// raw Loro handle around the preview's pending-check/fork critical section.
44-
pub(in crate::state) doc: LoroDoc,
46+
pub(in crate::state) doc: DocumentCell,
4547
pub(super) peer_id: u64,
4648
/// Loro auto-commit state must stay single-owner. This makes accidental
4749
/// cross-thread sharing of a `CrdtState` a compile error.
4850
pub(super) _single_owner: PhantomData<Cell<()>>,
49-
/// Last memory estimate and the frontier it was taken at.
50-
///
51-
/// The estimate is a snapshot export, which costs O(document). Callers
52-
/// poll it to decide when to compact, and a document that is not being
53-
/// written has the same estimate every time, so it is computed once per
54-
/// frontier rather than once per call.
55-
pub(super) memory_estimate: RefCell<Option<(loro::VersionVector, usize)>>,
5651
}
5752

5853
impl CrdtState {
5954
/// Create a new empty state for the given peer.
6055
pub fn new(peer_id: u64) -> Result<Self> {
61-
let doc = LoroDoc::new();
62-
doc.set_peer_id(peer_id)
63-
.map_err(|e| CrdtError::Loro(format!("failed to set peer_id {peer_id}: {e}")))?;
6456
Ok(Self {
65-
doc,
57+
doc: DocumentCell::new(Self::new_doc(peer_id)?),
6658
peer_id,
6759
_single_owner: PhantomData,
68-
memory_estimate: RefCell::new(None),
6960
})
7061
}
7162

63+
/// Load a state from an encoded document this process produced itself — a
64+
/// snapshot read back from durable storage, or a pre-image exported moments
65+
/// earlier for transaction rollback.
66+
///
67+
/// Pairs `new` with [`CrdtState::import_local`], because those two steps
68+
/// belong together: a caller that assembles them by hand has to know that
69+
/// its own bytes must not go through the peer ceilings, and a caller that
70+
/// gets that wrong writes a document it can no longer open. See
71+
/// `import_local` for why the ceilings do not apply here.
72+
pub fn from_local_snapshot(peer_id: u64, snapshot: &[u8]) -> Result<Self> {
73+
let state = Self::new(peer_id)?;
74+
state.import_local(snapshot)?;
75+
Ok(state)
76+
}
77+
78+
/// A fresh Loro document bound to `peer_id`.
79+
///
80+
/// Shared by `new` and by the compaction paths, which need the same
81+
/// peer-bound document before loading a shallow snapshot into it.
82+
pub(in crate::state) fn new_doc(peer_id: u64) -> Result<LoroDoc> {
83+
let doc = LoroDoc::new();
84+
doc.set_peer_id(peer_id)
85+
.map_err(|e| CrdtError::Loro(format!("failed to set peer_id {peer_id}: {e}")))?;
86+
Ok(doc)
87+
}
88+
7289
/// Fetch a row's existing `LoroMap` container, or create one if absent.
7390
/// Shared by `upsert` and `set_fields` — both need the same row handle
7491
/// before diverging on prune-vs-preserve semantics.
Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
3+
//! The document handle and every value derived from it.
4+
5+
use std::cell::RefCell;
6+
use std::ops::Deref;
7+
8+
use loro::LoroDoc;
9+
10+
/// A `LoroDoc` together with the values cached from it.
11+
///
12+
/// Compaction does not mutate a document, it replaces one: `compact_history`
13+
/// and `compact_at_version` build a fresh doc from a shallow snapshot and swap
14+
/// it in. Anything derived from the old doc and stored beside it survives that
15+
/// swap and goes on describing a document that no longer exists.
16+
///
17+
/// The size estimate is the case that bites. A shallow snapshot preserves the
18+
/// version vector — peers have to keep delta-syncing across a compaction — so
19+
/// a cache keyed on the version alone still looks current after the bytes it
20+
/// measured are gone. A caller polling the estimate to decide when to compact
21+
/// would then never observe its own compaction landing, and would compact
22+
/// again, and again.
23+
///
24+
/// Keeping the cache *inside* the cell removes the possibility: `replace` is
25+
/// the only way to swap the document, and it drops the derived state with it.
26+
/// Reads go through `Deref`, so every `self.doc.…` call site is untouched.
27+
pub(in crate::state) struct DocumentCell {
28+
doc: LoroDoc,
29+
/// Snapshot size and the oplog version it was measured at. `None` until
30+
/// the estimate is first asked for.
31+
memory_estimate: RefCell<Option<(loro::VersionVector, usize)>>,
32+
}
33+
34+
impl DocumentCell {
35+
/// Wrap a document with an empty derived-value cache.
36+
pub(in crate::state) fn new(doc: LoroDoc) -> Self {
37+
Self {
38+
doc,
39+
memory_estimate: RefCell::new(None),
40+
}
41+
}
42+
43+
/// Swap in a different document, discarding everything cached from the
44+
/// previous one. The only way to reassign the document.
45+
pub(in crate::state) fn replace(&mut self, doc: LoroDoc) {
46+
*self = Self::new(doc);
47+
}
48+
49+
/// Snapshot size in bytes, as a proxy for memory footprint.
50+
///
51+
/// Loro exposes no direct memory metric. A snapshot export is proportional
52+
/// to state size, which is good enough for pressure monitoring but costs
53+
/// O(document) — and the callers that want it are polling loops. It is
54+
/// therefore measured once per version rather than once per call.
55+
/// The version is a sound cache key because `oplog_vv` counts operations
56+
/// that are still in an open transaction — a write is visible to the key
57+
/// the moment it happens, not when it commits. `state::tests` pins that
58+
/// property, since the cache is wrong the day it stops holding.
59+
pub(in crate::state) fn estimated_bytes(&self) -> usize {
60+
let version = self.doc.oplog_vv();
61+
if let Some((measured_at, bytes)) = self.memory_estimate.borrow().as_ref()
62+
&& *measured_at == version
63+
{
64+
return *bytes;
65+
}
66+
67+
let Ok(snapshot) = self.doc.export(loro::ExportMode::Snapshot) else {
68+
// A failed export is not a measurement. Caching the zero would pin
69+
// the document at "empty" until its next write moved the version.
70+
return 0;
71+
};
72+
let bytes = snapshot.len();
73+
*self.memory_estimate.borrow_mut() = Some((version, bytes));
74+
bytes
75+
}
76+
}
77+
78+
impl Deref for DocumentCell {
79+
type Target = LoroDoc;
80+
81+
fn deref(&self) -> &Self::Target {
82+
&self.doc
83+
}
84+
}

‎nodedb-crdt/src/state/history.rs‎

Lines changed: 1 addition & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@ use loro::{LoroDoc, LoroMap, LoroValue, ValueOrContainer};
99
use crate::error::{CrdtError, Result};
1010

1111
use super::core::CrdtState;
12-
use super::import_admission::admit_local_import;
1312
use super::restore_containers;
1413

1514
impl CrdtState {
@@ -95,24 +94,7 @@ impl CrdtState {
9594
/// Discards oplog entries before the target version. Current state and
9695
/// all versions after the target are preserved.
9796
pub fn compact_at_version(&mut self, version: &loro::VersionVector) -> Result<()> {
98-
let frontiers = self.doc.vv_to_frontiers(version);
99-
let snapshot = self
100-
.doc
101-
.export(loro::ExportMode::shallow_snapshot(&frontiers))
102-
.map_err(|e| CrdtError::Loro(format!("shallow snapshot export: {e}")))?;
103-
104-
let new_doc = LoroDoc::new();
105-
new_doc
106-
.set_peer_id(self.peer_id)
107-
.map_err(|e| CrdtError::Loro(format!("set peer_id on compacted doc: {e}")))?;
108-
// Locally generated — see `compact_history`.
109-
admit_local_import(&snapshot, &new_doc.oplog_vv())?;
110-
new_doc
111-
.import(&snapshot)
112-
.map_err(|e| CrdtError::Loro(format!("shallow snapshot import: {e}")))?;
113-
114-
self.doc = new_doc;
115-
Ok(())
97+
self.compact_to_frontiers(&self.doc.vv_to_frontiers(version))
11698
}
11799

118100
/// Generate a forward restore delta without changing authoritative state.

‎nodedb-crdt/src/state/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
99
pub mod bitemporal_archive;
1010
pub mod core;
11+
pub(crate) mod document_cell;
1112
pub mod frontier_digest;
1213
pub mod history;
1314
pub(crate) mod import_admission;

‎nodedb-crdt/src/state/preview.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ use crate::error::{CrdtError, Result};
1616
use crate::loro_value::loro_to_value;
1717

1818
use super::core::CrdtState;
19+
use super::document_cell::DocumentCell;
1920
use super::import_admission::{CrdtImportLimits, admit_import};
2021

2122
/// Maximum raw CRDT delta bytes accepted by the default authoritative preview.
@@ -151,10 +152,9 @@ impl CrdtState {
151152

152153
let resulting_frontier = fork.state_frontiers();
153154
let fork_state = CrdtState {
154-
doc: fork,
155+
doc: DocumentCell::new(fork),
155156
peer_id: self.peer_id,
156157
_single_owner: std::marker::PhantomData,
157-
memory_estimate: std::cell::RefCell::new(None),
158158
};
159159
// Loro exposes diffs only as a complete DiffBatch. This runs after
160160
// byte and imported-operation caps have bounded the fork/import work;

‎nodedb-crdt/src/state/snapshot.rs‎

Lines changed: 60 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@
22

33
//! Snapshot export/import, history compaction, memory estimation.
44
5-
use loro::LoroDoc;
6-
75
use crate::error::{CrdtError, Result};
86

97
use super::core::CrdtState;
@@ -106,54 +104,55 @@ impl CrdtState {
106104
/// Call this periodically (e.g., every 30 minutes or when memory
107105
/// pressure exceeds threshold) to prevent unbounded history growth.
108106
pub fn compact_history(&mut self) -> Result<()> {
109-
// Export a shallow snapshot at the current frontiers.
110-
let frontiers = self.doc.oplog_frontiers();
107+
self.compact_to_frontiers(&self.doc.oplog_frontiers())
108+
}
109+
110+
/// Replace this document with a shallow snapshot taken at `frontiers`.
111+
///
112+
/// The whole of compaction, shared by `compact_history` (current
113+
/// frontiers) and `compact_at_version` (a chosen version), so the two
114+
/// cannot drift apart on how the snapshot is admitted or how the document
115+
/// is swapped in.
116+
///
117+
/// The snapshot is admitted through [`admit_local_import`]: this process
118+
/// exported it one line earlier. Under the peer ceilings, compaction would
119+
/// be refused for precisely the documents large enough to need it — the
120+
/// operation that would bring a document back under the limit gated by
121+
/// that limit.
122+
pub(in crate::state) fn compact_to_frontiers(
123+
&mut self,
124+
frontiers: &loro::Frontiers,
125+
) -> Result<()> {
111126
let snapshot = self
112127
.doc
113-
.export(loro::ExportMode::shallow_snapshot(&frontiers))
128+
.export(loro::ExportMode::shallow_snapshot(frontiers))
114129
.map_err(|e| CrdtError::Loro(format!("shallow snapshot export: {e}")))?;
115130

116-
// Replace the doc with a fresh one loaded from the snapshot.
117-
let new_doc = LoroDoc::new();
118-
new_doc
119-
.set_peer_id(self.peer_id)
120-
.map_err(|e| CrdtError::Loro(format!("failed to set peer_id on compacted doc: {e}")))?;
121-
// Locally generated: it keeps the metadata and range checks but not
122-
// the peer size ceilings, which would otherwise make compaction
123-
// impossible for exactly the documents that need it most.
124-
admit_local_import(&snapshot, &new_doc.oplog_vv())?;
125-
new_doc
131+
let compacted = Self::new_doc(self.peer_id)?;
132+
admit_local_import(&snapshot, &compacted.oplog_vv())?;
133+
compacted
126134
.import(&snapshot)
127135
.map_err(|e| CrdtError::Loro(format!("shallow snapshot import: {e}")))?;
128136

129-
self.doc = new_doc;
137+
// `replace` also drops everything cached from the outgoing document. A
138+
// shallow snapshot keeps the version vector, so a derived value keyed
139+
// on the version alone would look current while describing bytes that
140+
// no longer exist.
141+
self.doc.replace(compacted);
130142
Ok(())
131143
}
132144

133145
/// Estimated memory usage of the CRDT state (bytes).
134146
///
135147
/// Includes operation history, current state, and internal caches.
136148
/// Use this to decide when to trigger `compact_history()`.
137-
/// Cached per frontier: the proxy is a full snapshot export, so polling it
138-
/// on an unchanged document would re-serialize the whole document to learn
139-
/// a number that cannot have moved.
149+
///
150+
/// The proxy is a full snapshot export, so it is measured once per version
151+
/// rather than once per call: a document nobody is writing cannot have
152+
/// changed size. Compaction discards the measurement along with the
153+
/// document it described.
140154
pub fn estimated_memory_bytes(&self) -> usize {
141-
let version = self.doc.oplog_vv();
142-
if let Some((cached_at, bytes)) = self.memory_estimate.borrow().as_ref()
143-
&& *cached_at == version
144-
{
145-
return *bytes;
146-
}
147-
// Loro doesn't expose a direct memory metric.
148-
// Use snapshot size as a proxy — it's proportional to state size.
149-
// This is not precise but good enough for pressure monitoring.
150-
let bytes = self
151-
.doc
152-
.export(loro::ExportMode::Snapshot)
153-
.map(|s| s.len())
154-
.unwrap_or(0);
155-
*self.memory_estimate.borrow_mut() = Some((version, bytes));
156-
bytes
155+
self.doc.estimated_bytes()
157156
}
158157
}
159158

@@ -218,6 +217,32 @@ mod tests {
218217
);
219218
}
220219

220+
#[test]
221+
fn from_local_snapshot_loads_without_the_peer_ceilings() {
222+
let source = CrdtState::new(3).expect("source state");
223+
source
224+
.upsert("docs", "row", &[("title", LoroValue::String("v".into()))])
225+
.expect("source write");
226+
let snapshot = source.export_snapshot().expect("snapshot");
227+
228+
let reloaded = CrdtState::from_local_snapshot(3, &snapshot).expect("reload");
229+
assert!(
230+
reloaded.row_exists("docs", "row"),
231+
"a state loaded from our own snapshot must carry its rows"
232+
);
233+
}
234+
235+
#[test]
236+
fn from_local_snapshot_rejects_malformed_bytes() {
237+
assert!(
238+
matches!(
239+
CrdtState::from_local_snapshot(1, &[0xff]),
240+
Err(CrdtError::ImportMalformed { .. })
241+
),
242+
"the convenience constructor keeps every structural check"
243+
);
244+
}
245+
221246
#[test]
222247
fn local_import_still_rejects_malformed_metadata() {
223248
let state = CrdtState::new(1).expect("state");

0 commit comments

Comments
 (0)