diff --git a/nodedb-crdt/src/state/core.rs b/nodedb-crdt/src/state/core.rs index f354b37c9..3b9a83abe 100644 --- a/nodedb-crdt/src/state/core.rs +++ b/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 { - 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 { + 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 { + 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 { - 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(), } diff --git a/nodedb-crdt/src/state/document_cell.rs b/nodedb-crdt/src/state/document_cell.rs new file mode 100644 index 000000000..919e589f5 --- /dev/null +++ b/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>, +} + +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 + } +} diff --git a/nodedb-crdt/src/state/history.rs b/nodedb-crdt/src/state/history.rs index 6d9ad7ac9..9365bb9d0 100644 --- a/nodedb-crdt/src/state/history.rs +++ b/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. diff --git a/nodedb-crdt/src/state/import_admission.rs b/nodedb-crdt/src/state/import_admission.rs index 5007132bd..b7b261e95 100644 --- a/nodedb-crdt/src/state/import_admission.rs +++ b/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 { + 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 { 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, diff --git a/nodedb-crdt/src/state/mod.rs b/nodedb-crdt/src/state/mod.rs index e3a34048e..6121277b9 100644 --- a/nodedb-crdt/src/state/mod.rs +++ b/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; diff --git a/nodedb-crdt/src/state/preview.rs b/nodedb-crdt/src/state/preview.rs index cd35f6238..3b880574b 100644 --- a/nodedb-crdt/src/state/preview.rs +++ b/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, }; diff --git a/nodedb-crdt/src/state/snapshot.rs b/nodedb-crdt/src/state/snapshot.rs index 3adee2c26..237e5858a 100644 --- a/nodedb-crdt/src/state/snapshot.rs +++ b/nodedb-crdt/src/state/snapshot.rs @@ -2,12 +2,12 @@ //! Snapshot export/import, history compaction, memory estimation. -use loro::LoroDoc; - use crate::error::{CrdtError, Result}; use super::core::CrdtState; -use super::import_admission::{CrdtImportLimits, ImportAdmission, admit_import}; +use super::import_admission::{ + CrdtImportLimits, ImportAdmission, admit_import, admit_local_import, +}; impl CrdtState { /// Export the current state as bytes for sync. @@ -24,6 +24,35 @@ impl CrdtState { self.import_with_limits(data, CrdtImportLimits::default()) } + /// Import bytes this process produced itself — a snapshot read back from + /// durable storage, or a shallow snapshot taken during compaction. + /// + /// Keeps every structural check [`Self::import`] performs: metadata is + /// decoded with Loro's authenticated decoder, and per-peer ranges that + /// regress are rejected before any state mutation. What it drops is the + /// size ceilings, which bound how much work an untrusted peer may cause + /// and have no meaning for bytes this library just wrote. + /// + /// Without this split the ceilings apply to reloading as well as to + /// receiving, and since `export_snapshot` has no bound at all, a document a + /// healthy process wrote can exceed what the same binary will re-import. + /// Past that point the document neither opens nor compacts — compaction + /// being the operation that would bring it back under the limit. + /// + /// This is not a way to raise the peer limits: `import_with_limits` is the + /// knob for that, and bytes off the wire must go through it. + pub fn import_local(&self, data: &[u8]) -> Result { + let admission = admit_local_import(data, &self.doc.oplog_vv())?; + let status = self + .doc + .import(data) + .map_err(|e| CrdtError::DeltaApplyFailed(e.to_string()))?; + if status.pending.is_some() { + return Err(CrdtError::ImportPendingDependencies); + } + Ok(admission) + } + /// Import remote updates after bounded authenticated metadata admission. /// /// The limits are checked before Loro can import the blob, so rejected @@ -75,26 +104,41 @@ impl CrdtState { /// Call this periodically (e.g., every 30 minutes or when memory /// pressure exceeds threshold) to prevent unbounded history growth. pub fn compact_history(&mut self) -> Result<()> { - // Export a shallow snapshot at the current frontiers. - let frontiers = self.doc.oplog_frontiers(); + self.compact_to_frontiers(&self.doc.oplog_frontiers()) + } + + /// Replace this document with a shallow snapshot taken at `frontiers`. + /// + /// The whole of compaction, shared by `compact_history` (current + /// frontiers) and `compact_at_version` (a chosen version), so the two + /// cannot drift apart on how the snapshot is admitted or how the document + /// is swapped in. + /// + /// The snapshot is admitted through [`admit_local_import`]: this process + /// exported it one line earlier. Under the peer ceilings, compaction would + /// be refused for precisely the documents large enough to need it — the + /// operation that would bring a document back under the limit gated by + /// that limit. + pub(in crate::state) fn compact_to_frontiers( + &mut self, + frontiers: &loro::Frontiers, + ) -> Result<()> { let snapshot = self .doc - .export(loro::ExportMode::shallow_snapshot(&frontiers)) + .export(loro::ExportMode::shallow_snapshot(frontiers)) .map_err(|e| CrdtError::Loro(format!("shallow snapshot export: {e}")))?; - // Replace the doc with a fresh one loaded from the snapshot. - let new_doc = LoroDoc::new(); - new_doc - .set_peer_id(self.peer_id) - .map_err(|e| CrdtError::Loro(format!("failed to set peer_id on compacted doc: {e}")))?; - // This snapshot is generated locally, but it remains finite and goes - // through the same metadata/range admission before import. - admit_import(&snapshot, &new_doc.oplog_vv(), CrdtImportLimits::default())?; - new_doc + let compacted = Self::new_doc(self.peer_id)?; + admit_local_import(&snapshot, &compacted.oplog_vv())?; + compacted .import(&snapshot) .map_err(|e| CrdtError::Loro(format!("shallow snapshot import: {e}")))?; - self.doc = new_doc; + // `replace` also drops everything cached from the outgoing document. A + // shallow snapshot keeps the version vector, so a derived value keyed + // on the version alone would look current while describing bytes that + // no longer exist. + self.doc.replace(compacted); Ok(()) } @@ -102,14 +146,13 @@ impl CrdtState { /// /// Includes operation history, current state, and internal caches. /// Use this to decide when to trigger `compact_history()`. + /// + /// The proxy is a full snapshot export, so it is measured once per version + /// rather than once per call: a document nobody is writing cannot have + /// changed size. Compaction discards the measurement along with the + /// document it described. pub fn estimated_memory_bytes(&self) -> usize { - // Loro doesn't expose a direct memory metric. - // Use snapshot size as a proxy — it's proportional to state size. - // This is not precise but good enough for pressure monitoring. - self.doc - .export(loro::ExportMode::Snapshot) - .map(|s| s.len()) - .unwrap_or(0) + self.doc.estimated_bytes() } } @@ -148,6 +191,72 @@ mod tests { assert_eq!(state.frontier(), before); } + #[test] + fn local_import_admits_what_the_peer_ceilings_would_reject() { + let delta = source_delta(); + let capped = CrdtState::new(1).expect("capped state"); + let tight = CrdtImportLimits { + max_bytes: delta.len(), + max_encoded_operations: 0, + max_new_operations: 0, + }; + assert!( + matches!( + capped.import_with_limits(&delta, tight), + Err(CrdtError::ImportOperationLimitExceeded { .. }) + ), + "the peer path must stay capped" + ); + + let local = CrdtState::new(1).expect("local state"); + local.import_local(&delta).expect("local import"); + assert!( + local.row_exists("docs", "row"), + "bytes this process wrote must be reloadable whatever the peer ceilings are; a \ + document large enough to trip them can otherwise neither be opened nor compacted" + ); + } + + #[test] + fn from_local_snapshot_loads_without_the_peer_ceilings() { + let source = CrdtState::new(3).expect("source state"); + source + .upsert("docs", "row", &[("title", LoroValue::String("v".into()))]) + .expect("source write"); + let snapshot = source.export_snapshot().expect("snapshot"); + + let reloaded = CrdtState::from_local_snapshot(3, &snapshot).expect("reload"); + assert!( + reloaded.row_exists("docs", "row"), + "a state loaded from our own snapshot must carry its rows" + ); + } + + #[test] + fn from_local_snapshot_rejects_malformed_bytes() { + assert!( + matches!( + CrdtState::from_local_snapshot(1, &[0xff]), + Err(CrdtError::ImportMalformed { .. }) + ), + "the convenience constructor keeps every structural check" + ); + } + + #[test] + fn local_import_still_rejects_malformed_metadata() { + let state = CrdtState::new(1).expect("state"); + let before = state.frontier(); + assert!( + matches!( + state.import_local(&[0xff]), + Err(CrdtError::ImportMalformed { .. }) + ), + "dropping the size ceilings must not drop the structural checks" + ); + assert_eq!(state.frontier(), before); + } + #[test] fn import_rejects_malformed_metadata_before_state_mutation() { let state = CrdtState::new(1).expect("state"); diff --git a/nodedb-crdt/src/state/tests.rs b/nodedb-crdt/src/state/tests.rs index 95520dadc..3bccf4e24 100644 --- a/nodedb-crdt/src/state/tests.rs +++ b/nodedb-crdt/src/state/tests.rs @@ -190,6 +190,55 @@ fn estimated_memory_grows_with_data() { ); } +#[test] +fn estimated_memory_follows_compaction() { + let mut state = CrdtState::new(1).unwrap(); + // History, not breadth: compaction discards the operations, so the same + // row rewritten many times is what shrinks when it runs. + for i in 0..512 { + state + .upsert("items", "hot", &[("value", LoroValue::I64(i))]) + .unwrap(); + } + let before = state.estimated_memory_bytes(); + + state.compact_history().unwrap(); + let after = state.estimated_memory_bytes(); + + assert_eq!( + after, + state.export_snapshot().unwrap().len(), + "the estimate must describe the document that exists now, not the one \ + compaction replaced; a shallow snapshot keeps the version vector, so a \ + measurement keyed on the version alone still looks current after the \ + bytes it measured are gone" + ); + assert!( + after < before, + "compaction dropped 512 operations and the estimate did not move: \ + before={before}, after={after} — a caller polling this to decide when \ + to compact would compact forever" + ); +} + +#[test] +fn oplog_version_counts_uncommitted_operations() { + let state = CrdtState::new(1).unwrap(); + state + .upsert("items", "a", &[("value", LoroValue::I64(1))]) + .unwrap(); + + // The memory estimate is cached against this version vector. That is only + // sound because Loro advances it when the operation is written rather than + // when its transaction commits — otherwise a write could land while the + // key stood still, and the estimate would answer from before it. If this + // assertion ever fails, the cache in `document_cell` is what breaks. + assert!( + state.oplog_version_vector().get(&1).copied().unwrap_or(0) > 0, + "an operation that has not committed yet must still move the version" + ); +} + #[test] fn restore_to_version_produces_forward_delta() { let state = CrdtState::new(1).unwrap(); @@ -401,3 +450,23 @@ fn snapshot_roundtrip() { assert!(state2.row_exists("users", "u1")); } + +#[test] +fn collection_names_lists_top_level_keys_only() { + let state = CrdtState::new(1).unwrap(); + state + .upsert("users", "u1", &[("name", LoroValue::String("Bob".into()))]) + .unwrap(); + state + .upsert("orders", "o1", &[("total", LoroValue::I64(42))]) + .unwrap(); + + let mut names = state.collection_names(); + names.sort(); + assert_eq!( + names, + vec!["orders".to_string(), "users".to_string()], + "the collections are the top-level keys; row ids and fields belong to \ + the collections, not to this list" + ); +} diff --git a/nodedb/src/engine/crdt/tenant_state/apply_validated.rs b/nodedb/src/engine/crdt/tenant_state/apply_validated.rs index 900a30fc8..66a7a6fc0 100644 --- a/nodedb/src/engine/crdt/tenant_state/apply_validated.rs +++ b/nodedb/src/engine/crdt/tenant_state/apply_validated.rs @@ -121,19 +121,27 @@ impl TenantCrdtEngine { { return ValidatedApplyOutcome::Malformed; } - let candidate = match CrdtState::new(self.peer_id) { - Ok(state) => state, - Err(_) => return ValidatedApplyOutcome::Malformed, - }; - if let Some(current) = self.collections.get(collection) { - let snapshot = match current.export_snapshot() { - Ok(snapshot) => snapshot, - Err(_) => return ValidatedApplyOutcome::Malformed, - }; - if candidate.import(&snapshot).is_err() { - 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 before = candidate.frontier(); let admission = match candidate.import(delta) { Ok(admission) => admission, diff --git a/nodedb/src/engine/crdt/tenant_state/snapshot_restore.rs b/nodedb/src/engine/crdt/tenant_state/snapshot_restore.rs index a3647fbfe..0ca657711 100644 --- a/nodedb/src/engine/crdt/tenant_state/snapshot_restore.rs +++ b/nodedb/src/engine/crdt/tenant_state/snapshot_restore.rs @@ -30,9 +30,17 @@ impl TenantCrdtEngine { // Same per-collection derivation as `state_mut`: a rollback must not // hand the collection back a document whose operation identities // collide with a sibling collection's. - let replacement = CrdtState::new(Self::collection_peer_id(self.peer_id, collection)) - .map_err(crate::Error::Crdt)?; - replacement.import(snapshot).map_err(crate::Error::Crdt)?; + // + // The pre-image is a snapshot this process exported when the + // transaction opened, so it is admitted as local: under the peer + // ceilings a collection that grew past them could be written but never + // rolled back, and the transaction driver would be handed a failure it + // has no way to act on. + let replacement = CrdtState::from_local_snapshot( + Self::collection_peer_id(self.peer_id, collection), + snapshot, + ) + .map_err(crate::Error::Crdt)?; self.collections.insert(collection.to_owned(), replacement); Ok(()) }