Skip to content

Commit 5ae4a50

Browse files
authored
Merge pull request #231 from mkhairi/fix/crdt-local-import-admission
feat(crdt): separate local reload from peer admission
2 parents 371fb5a + e7c2c68 commit 5ae4a50

10 files changed

Lines changed: 401 additions & 76 deletions

File tree

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

Lines changed: 36 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -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,7 +43,7 @@ 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.
@@ -51,16 +53,39 @@ pub struct CrdtState {
5153
impl CrdtState {
5254
/// Create a new empty state for the given peer.
5355
pub fn new(peer_id: u64) -> Result<Self> {
54-
let doc = LoroDoc::new();
55-
doc.set_peer_id(peer_id)
56-
.map_err(|e| CrdtError::Loro(format!("failed to set peer_id {peer_id}: {e}")))?;
5756
Ok(Self {
58-
doc,
57+
doc: DocumentCell::new(Self::new_doc(peer_id)?),
5958
peer_id,
6059
_single_owner: PhantomData,
6160
})
6261
}
6362

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+
6489
/// Fetch a row's existing `LoroMap` container, or create one if absent.
6590
/// Shared by `upsert` and `set_fields` — both need the same row handle
6691
/// before diverging on prune-vs-preserve semantics.
@@ -284,9 +309,13 @@ impl CrdtState {
284309
}
285310

286311
/// List all collection names (top-level map keys in the Loro doc).
312+
///
313+
/// Reads the shallow value: the keys are at the top level, and
314+
/// `get_deep_value` would materialise every row and field of every
315+
/// collection to reach them — O(document) for a list whose size is the
316+
/// number of collections. Name resolution calls this per query.
287317
pub fn collection_names(&self) -> Vec<String> {
288-
let root = self.doc.get_deep_value();
289-
match root {
318+
match self.doc.get_value() {
290319
LoroValue::Map(map) => map.keys().map(|k| k.to_string()).collect(),
291320
_ => Vec::new(),
292321
}
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 & 18 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::{CrdtImportLimits, admit_import};
1312
use super::restore_containers;
1413

1514
impl CrdtState {
@@ -95,23 +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-
admit_import(&snapshot, &new_doc.oplog_vv(), CrdtImportLimits::default())?;
109-
new_doc
110-
.import(&snapshot)
111-
.map_err(|e| CrdtError::Loro(format!("shallow snapshot import: {e}")))?;
112-
113-
self.doc = new_doc;
114-
Ok(())
97+
self.compact_to_frontiers(&self.doc.vv_to_frontiers(version))
11598
}
11699

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

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

Lines changed: 45 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -74,29 +74,62 @@ pub(crate) fn admit_import(
7474
});
7575
}
7676

77+
let admission = admit_structure(bytes, current)?;
78+
if admission.encoded_operations > limits.max_encoded_operations {
79+
return Err(CrdtError::ImportOperationLimitExceeded {
80+
limit: limits.max_encoded_operations,
81+
actual: admission.encoded_operations,
82+
});
83+
}
84+
if admission.new_operations > limits.max_new_operations {
85+
return Err(CrdtError::ImportOperationLimitExceeded {
86+
limit: limits.max_new_operations,
87+
actual: admission.new_operations,
88+
});
89+
}
90+
Ok(admission)
91+
}
92+
93+
/// Validate an encoded Loro import this process produced itself.
94+
///
95+
/// Structurally identical to [`admit_import`] — the same authenticated
96+
/// metadata decode, the same rejection of regressing per-peer ranges — but
97+
/// without the size ceilings.
98+
///
99+
/// Those ceilings answer a question that only arises for bytes off the wire:
100+
/// how much work an untrusted peer may make this document do. Applied to
101+
/// locally-generated bytes they instead cap how large a document this library
102+
/// may reload after writing it, which is a different and unintended promise —
103+
/// `export_snapshot` has no bound at all, so a document a healthy process
104+
/// wrote could exceed what the same binary would re-import. Restoring from
105+
/// disk and folding a shallow snapshot back in during compaction are both that
106+
/// case: past the ceiling the document can no longer be opened *and* no longer
107+
/// be compacted, so the one operation that would bring it back under the limit
108+
/// is gated by the limit.
109+
///
110+
/// Deliberately a separate function rather than an unbounded `CrdtImportLimits`
111+
/// constructor: bytes from a peer cannot reach this path by passing a value.
112+
pub(crate) fn admit_local_import(
113+
bytes: &[u8],
114+
current: &loro::VersionVector,
115+
) -> Result<ImportAdmission> {
116+
admit_structure(bytes, current)
117+
}
118+
119+
/// The checks that hold regardless of where the bytes came from: authenticated
120+
/// metadata, and per-peer ranges that do not regress.
121+
fn admit_structure(bytes: &[u8], current: &loro::VersionVector) -> Result<ImportAdmission> {
77122
let metadata = loro::LoroDoc::decode_import_blob_meta(bytes, true).map_err(|error| {
78123
CrdtError::ImportMalformed {
79124
detail: error.to_string(),
80125
}
81126
})?;
82127
let encoded = count_range_operations(&metadata.partial_start_vv, &metadata.partial_end_vv)?;
83-
if encoded > limits.max_encoded_operations {
84-
return Err(CrdtError::ImportOperationLimitExceeded {
85-
limit: limits.max_encoded_operations,
86-
actual: encoded,
87-
});
88-
}
89128
let imported = count_new_operations(
90129
&metadata.partial_start_vv,
91130
&metadata.partial_end_vv,
92131
current,
93132
)?;
94-
if imported > limits.max_new_operations {
95-
return Err(CrdtError::ImportOperationLimitExceeded {
96-
limit: limits.max_new_operations,
97-
actual: imported,
98-
});
99-
}
100133
Ok(ImportAdmission {
101134
encoded_operations: encoded,
102135
new_operations: imported,

‎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 & 1 deletion
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,7 +152,7 @@ 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,
157158
};

0 commit comments

Comments
 (0)