feat(crdt): separate local reload from peer admission

nodedb-crdt/src/state/core.rs

@@ -12,6 +12,8 @@ use crate::error::{CrdtError, Result};
 use crate::row_lookup::RowLookup;
 use crate::validator::bitemporal::{VALID_UNTIL, VALID_UNTIL_OPEN};
 
+use super::document_cell::DocumentCell;
+
 /// A row is live when its `_ts_valid_until` field is absent, null, or the
 /// open sentinel (`i64::MAX`). Rows with any finite `_ts_valid_until` are
 /// treated as superseded, independent of wall-clock time — the write path
@@ -41,7 +43,7 @@ fn key_is_container(row: &LoroMap, key: &str) -> bool {
 pub struct CrdtState {
     /// Kept private to this state module so callers cannot clone or share a
     /// raw Loro handle around the preview's pending-check/fork critical section.
-    pub(in crate::state) doc: LoroDoc,
+    pub(in crate::state) doc: DocumentCell,
     pub(super) peer_id: u64,
     /// Loro auto-commit state must stay single-owner. This makes accidental
     /// cross-thread sharing of a `CrdtState` a compile error.
@@ -51,16 +53,39 @@ pub struct CrdtState {
 impl CrdtState {
     /// Create a new empty state for the given peer.
     pub fn new(peer_id: u64) -> Result<Self> {
-        let doc = LoroDoc::new();
-        doc.set_peer_id(peer_id)
-            .map_err(|e| CrdtError::Loro(format!("failed to set peer_id {peer_id}: {e}")))?;
         Ok(Self {
-            doc,
+            doc: DocumentCell::new(Self::new_doc(peer_id)?),
             peer_id,
             _single_owner: PhantomData,
         })
     }
 
+    /// Load a state from an encoded document this process produced itself — a
+    /// snapshot read back from durable storage, or a pre-image exported moments
+    /// earlier for transaction rollback.
+    ///
+    /// Pairs `new` with [`CrdtState::import_local`], because those two steps
+    /// belong together: a caller that assembles them by hand has to know that
+    /// its own bytes must not go through the peer ceilings, and a caller that
+    /// gets that wrong writes a document it can no longer open. See
+    /// `import_local` for why the ceilings do not apply here.
+    pub fn from_local_snapshot(peer_id: u64, snapshot: &[u8]) -> Result<Self> {
+        let state = Self::new(peer_id)?;
+        state.import_local(snapshot)?;
+        Ok(state)
+    }
+
+    /// A fresh Loro document bound to `peer_id`.
+    ///
+    /// Shared by `new` and by the compaction paths, which need the same
+    /// peer-bound document before loading a shallow snapshot into it.
+    pub(in crate::state) fn new_doc(peer_id: u64) -> Result<LoroDoc> {
+        let doc = LoroDoc::new();
+        doc.set_peer_id(peer_id)
+            .map_err(|e| CrdtError::Loro(format!("failed to set peer_id {peer_id}: {e}")))?;
+        Ok(doc)
+    }
+
     /// Fetch a row's existing `LoroMap` container, or create one if absent.
     /// Shared by `upsert` and `set_fields` — both need the same row handle
     /// before diverging on prune-vs-preserve semantics.
@@ -284,9 +309,13 @@ impl CrdtState {
     }
 
     /// List all collection names (top-level map keys in the Loro doc).
+    ///
+    /// Reads the shallow value: the keys are at the top level, and
+    /// `get_deep_value` would materialise every row and field of every
+    /// collection to reach them — O(document) for a list whose size is the
+    /// number of collections. Name resolution calls this per query.
     pub fn collection_names(&self) -> Vec<String> {
-        let root = self.doc.get_deep_value();
-        match root {
+        match self.doc.get_value() {
             LoroValue::Map(map) => map.keys().map(|k| k.to_string()).collect(),
             _ => Vec::new(),
         }

nodedb-crdt/src/state/document_cell.rs

@@ -0,0 +1,84 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! The document handle and every value derived from it.
+
+use std::cell::RefCell;
+use std::ops::Deref;
+
+use loro::LoroDoc;
+
+/// A `LoroDoc` together with the values cached from it.
+///
+/// Compaction does not mutate a document, it replaces one: `compact_history`
+/// and `compact_at_version` build a fresh doc from a shallow snapshot and swap
+/// it in. Anything derived from the old doc and stored beside it survives that
+/// swap and goes on describing a document that no longer exists.
+///
+/// The size estimate is the case that bites. A shallow snapshot preserves the
+/// version vector — peers have to keep delta-syncing across a compaction — so
+/// a cache keyed on the version alone still looks current after the bytes it
+/// measured are gone. A caller polling the estimate to decide when to compact
+/// would then never observe its own compaction landing, and would compact
+/// again, and again.
+///
+/// Keeping the cache *inside* the cell removes the possibility: `replace` is
+/// the only way to swap the document, and it drops the derived state with it.
+/// Reads go through `Deref`, so every `self.doc.…` call site is untouched.
+pub(in crate::state) struct DocumentCell {
+    doc: LoroDoc,
+    /// Snapshot size and the oplog version it was measured at. `None` until
+    /// the estimate is first asked for.
+    memory_estimate: RefCell<Option<(loro::VersionVector, usize)>>,
+}
+
+impl DocumentCell {
+    /// Wrap a document with an empty derived-value cache.
+    pub(in crate::state) fn new(doc: LoroDoc) -> Self {
+        Self {
+            doc,
+            memory_estimate: RefCell::new(None),
+        }
+    }
+
+    /// Swap in a different document, discarding everything cached from the
+    /// previous one. The only way to reassign the document.
+    pub(in crate::state) fn replace(&mut self, doc: LoroDoc) {
+        *self = Self::new(doc);
+    }
+
+    /// Snapshot size in bytes, as a proxy for memory footprint.
+    ///
+    /// Loro exposes no direct memory metric. A snapshot export is proportional
+    /// to state size, which is good enough for pressure monitoring but costs
+    /// O(document) — and the callers that want it are polling loops. It is
+    /// therefore measured once per version rather than once per call.
+    /// The version is a sound cache key because `oplog_vv` counts operations
+    /// that are still in an open transaction — a write is visible to the key
+    /// the moment it happens, not when it commits. `state::tests` pins that
+    /// property, since the cache is wrong the day it stops holding.
+    pub(in crate::state) fn estimated_bytes(&self) -> usize {
+        let version = self.doc.oplog_vv();
+        if let Some((measured_at, bytes)) = self.memory_estimate.borrow().as_ref()
+            && *measured_at == version
+        {
+            return *bytes;
+        }
+
+        let Ok(snapshot) = self.doc.export(loro::ExportMode::Snapshot) else {
+            // A failed export is not a measurement. Caching the zero would pin
+            // the document at "empty" until its next write moved the version.
+            return 0;
+        };
+        let bytes = snapshot.len();
+        *self.memory_estimate.borrow_mut() = Some((version, bytes));
+        bytes
+    }
+}
+
+impl Deref for DocumentCell {
+    type Target = LoroDoc;
+
+    fn deref(&self) -> &Self::Target {
+        &self.doc
+    }
+}

nodedb-crdt/src/state/history.rs

@@ -9,7 +9,6 @@ use loro::{LoroDoc, LoroMap, LoroValue, ValueOrContainer};
 use crate::error::{CrdtError, Result};
 
 use super::core::CrdtState;
-use super::import_admission::{CrdtImportLimits, admit_import};
 use super::restore_containers;
 
 impl CrdtState {
@@ -95,23 +94,7 @@ impl CrdtState {
     /// Discards oplog entries before the target version. Current state and
     /// all versions after the target are preserved.
     pub fn compact_at_version(&mut self, version: &loro::VersionVector) -> Result<()> {
-        let frontiers = self.doc.vv_to_frontiers(version);
-        let snapshot = self
-            .doc
-            .export(loro::ExportMode::shallow_snapshot(&frontiers))
-            .map_err(|e| CrdtError::Loro(format!("shallow snapshot export: {e}")))?;
-
-        let new_doc = LoroDoc::new();
-        new_doc
-            .set_peer_id(self.peer_id)
-            .map_err(|e| CrdtError::Loro(format!("set peer_id on compacted doc: {e}")))?;
-        admit_import(&snapshot, &new_doc.oplog_vv(), CrdtImportLimits::default())?;
-        new_doc
-            .import(&snapshot)
-            .map_err(|e| CrdtError::Loro(format!("shallow snapshot import: {e}")))?;
-
-        self.doc = new_doc;
-        Ok(())
+        self.compact_to_frontiers(&self.doc.vv_to_frontiers(version))
     }
 
     /// Generate a forward restore delta without changing authoritative state.

nodedb-crdt/src/state/import_admission.rs

@@ -74,29 +74,62 @@ pub(crate) fn admit_import(
         });
     }
 
+    let admission = admit_structure(bytes, current)?;
+    if admission.encoded_operations > limits.max_encoded_operations {
+        return Err(CrdtError::ImportOperationLimitExceeded {
+            limit: limits.max_encoded_operations,
+            actual: admission.encoded_operations,
+        });
+    }
+    if admission.new_operations > limits.max_new_operations {
+        return Err(CrdtError::ImportOperationLimitExceeded {
+            limit: limits.max_new_operations,
+            actual: admission.new_operations,
+        });
+    }
+    Ok(admission)
+}
+
+/// Validate an encoded Loro import this process produced itself.
+///
+/// Structurally identical to [`admit_import`] — the same authenticated
+/// metadata decode, the same rejection of regressing per-peer ranges — but
+/// without the size ceilings.
+///
+/// Those ceilings answer a question that only arises for bytes off the wire:
+/// how much work an untrusted peer may make this document do. Applied to
+/// locally-generated bytes they instead cap how large a document this library
+/// may reload after writing it, which is a different and unintended promise —
+/// `export_snapshot` has no bound at all, so a document a healthy process
+/// wrote could exceed what the same binary would re-import. Restoring from
+/// disk and folding a shallow snapshot back in during compaction are both that
+/// case: past the ceiling the document can no longer be opened *and* no longer
+/// be compacted, so the one operation that would bring it back under the limit
+/// is gated by the limit.
+///
+/// Deliberately a separate function rather than an unbounded `CrdtImportLimits`
+/// constructor: bytes from a peer cannot reach this path by passing a value.
+pub(crate) fn admit_local_import(
+    bytes: &[u8],
+    current: &loro::VersionVector,
+) -> Result<ImportAdmission> {
+    admit_structure(bytes, current)
+}
+
+/// The checks that hold regardless of where the bytes came from: authenticated
+/// metadata, and per-peer ranges that do not regress.
+fn admit_structure(bytes: &[u8], current: &loro::VersionVector) -> Result<ImportAdmission> {
     let metadata = loro::LoroDoc::decode_import_blob_meta(bytes, true).map_err(|error| {
         CrdtError::ImportMalformed {
             detail: error.to_string(),
         }
     })?;
     let encoded = count_range_operations(&metadata.partial_start_vv, &metadata.partial_end_vv)?;
-    if encoded > limits.max_encoded_operations {
-        return Err(CrdtError::ImportOperationLimitExceeded {
-            limit: limits.max_encoded_operations,
-            actual: encoded,
-        });
-    }
     let imported = count_new_operations(
         &metadata.partial_start_vv,
         &metadata.partial_end_vv,
         current,
     )?;
-    if imported > limits.max_new_operations {
-        return Err(CrdtError::ImportOperationLimitExceeded {
-            limit: limits.max_new_operations,
-            actual: imported,
-        });
-    }
     Ok(ImportAdmission {
         encoded_operations: encoded,
         new_operations: imported,

nodedb-crdt/src/state/mod.rs

@@ -8,6 +8,7 @@
 
 pub mod bitemporal_archive;
 pub mod core;
+pub(crate) mod document_cell;
 pub mod frontier_digest;
 pub mod history;
 pub(crate) mod import_admission;

nodedb-crdt/src/state/preview.rs

@@ -16,6 +16,7 @@ use crate::error::{CrdtError, Result};
 use crate::loro_value::loro_to_value;
 
 use super::core::CrdtState;
+use super::document_cell::DocumentCell;
 use super::import_admission::{CrdtImportLimits, admit_import};
 
 /// Maximum raw CRDT delta bytes accepted by the default authoritative preview.
@@ -151,7 +152,7 @@ impl CrdtState {
 
         let resulting_frontier = fork.state_frontiers();
         let fork_state = CrdtState {
-            doc: fork,
+            doc: DocumentCell::new(fork),
             peer_id: self.peer_id,
             _single_owner: std::marker::PhantomData,
         };