From 351242a9b730ed8a8a91d8f64b297e0903b7a3ad Mon Sep 17 00:00:00 2001 From: Magnus Rex Date: Wed, 2 Sep 2026 13:39:13 +0200 Subject: [PATCH 1/2] fix: reject shallow imports whose deps are the shallow root's own deps Importing into a shallow doc an update whose deps equal the shallow root's deps aborted the process. `import_deps_before_shallow_root` resolved the deps through `frontiers_to_vv`, which special-cases exactly that set to `shallow_since_vv` so a shallow doc can re-export at its own cut; the vv inclusion check then passed. With no lamport computable the change was parked as pending, and the pending replay (`remote_change_apply_state`) checks deps against the oplog vv, which covers trimmed history, so it applied the change and hit `calc_unknown_lamport_change(..).unwrap()`. The panic poisoned the doc mutex and the process aborted in a destructor. Such an update is concurrent with the shallow root: the doc's earliest state is the root's and the dag has no trimmed nodes, so it can never be merged, any more than an update that depends further below the cut. The preflight now checks for trimmed deps before resolving the frontier and rejects with `ImportUpdatesThatDependsOnOutdatedVersion`. The old boundary arm, which let any frontier touching the root through, was a second route to the same abort for hand-written JSON deps that mix the root with a trimmed id of another peer; it is gone. Rationale in context/internal-encoding.md. --- .changeset/import-at-shallow-root-deps.md | 10 ++ context/internal-encoding.md | 11 ++ crates/loro-internal/src/oplog/loro_dag.rs | 60 ++++++--- .../integration_test/shallow_snapshot_test.rs | 115 +++++++++++++++++- 4 files changed, 178 insertions(+), 18 deletions(-) create mode 100644 .changeset/import-at-shallow-root-deps.md diff --git a/.changeset/import-at-shallow-root-deps.md b/.changeset/import-at-shallow-root-deps.md new file mode 100644 index 000000000..462808142 --- /dev/null +++ b/.changeset/import-at-shallow-root-deps.md @@ -0,0 +1,10 @@ +--- +"loro-crdt": patch +--- + +Importing into a shallow doc an update whose deps are the shallow root's own +deps, or that mix the root with a trimmed id of another peer, now returns +`ImportUpdatesThatDependsOnOutdatedVersion` instead of panicking (and aborting +the process through a poisoned doc mutex). Such an update is concurrent with +the shallow root, so it is rejected like any other update that branches off +trimmed history. diff --git a/context/internal-encoding.md b/context/internal-encoding.md index 01270e5a2..b3ab78699 100644 --- a/context/internal-encoding.md +++ b/context/internal-encoding.md @@ -216,6 +216,17 @@ Pre-shallow frontier safety lives in `loro.rs`: `checkout`, `diff`, and `revert_to` must return `SwitchToVersionBeforeShallowRoot` instead of traversing history before the shallow root. +On the import side the gate is `AppDag::import_deps_before_shallow_root`, +reached from `preflight_import_changes` and from +`import_changes_to_oplog`. A change with any dep inside `shallow_since_vv` +must be rejected with `ImportUpdatesThatDependsOnOutdatedVersion`: trimmed ids +have no dag node, so no lamport can ever be computed for the change, and the +pending-change replay (`remote_change_apply_state`) checks deps against the +oplog vv, which covers trimmed history, so it would apply the change and hit +the `unwrap` in `apply_change_from_remote`. The check must run before +`frontiers_to_vv`, which resolves the root's own deps to `shallow_since_vv` +even though they are trimmed. + ## JSON Updates `json_schema.rs` is not wrapped in the binary `loro` envelope. Its diff --git a/crates/loro-internal/src/oplog/loro_dag.rs b/crates/loro-internal/src/oplog/loro_dag.rs index bafc28d3e..6cf58330b 100644 --- a/crates/loro-internal/src/oplog/loro_dag.rs +++ b/crates/loro-internal/src/oplog/loro_dag.rs @@ -802,23 +802,23 @@ impl AppDag { return true; } - let shallow_vv = VersionVector::from_im_vv(&self.shallow_since_vv); - if let Some(vv) = self.frontiers_to_vv(deps) { - return !vv.includes_vv(&shallow_vv); - } - - // Import only needs to reject updates whose causal source is older than - // the shallow root. A dependency set that touches the retained boundary - // can still be a valid post-root update, even when the rest of the deps - // are imported later in the same batch. - if deps - .iter() - .any(|id| self.shallow_since_frontiers.contains(&id)) - { - return false; + // A trimmed dep has no dag node, so the change can never get a lamport + // and could only abort or park forever. This has to be decided before + // `frontiers_to_vv`: the root's own deps are all trimmed, yet + // `frontiers_to_vv` resolves exactly that set to `shallow_since_vv`, + // which would pass the inclusion check below even though such a change + // is concurrent with the root. + if deps.iter().any(|id| self.shallow_since_vv.includes_id(id)) { + return true; } - deps.iter().any(|id| self.shallow_since_vv.includes_id(id)) + // Resolvable deps whose past does not cover the trimmed history sit on + // a retained branch concurrent with the root, and the doc has no state + // before the root to replay them against. Deps that are not imported + // yet are not older than the root, so the change can wait as pending, + // even when the rest of the deps are imported later in the same batch. + self.frontiers_to_vv(deps) + .is_some_and(|vv| !vv.includes_vv(&self.shallow_since_vv.to_vv())) } /// Travel the ancestors of the given id, and call the callback for each node @@ -1188,7 +1188,11 @@ impl AppDag { /// Convert a frontiers to a version vector /// - /// If the frontiers version is not found in the dag, return None + /// If the frontiers version is not found in the dag, return None. The one + /// exception is the shallow root's own deps: they have no dag nodes but + /// resolve to `shallow_since_vv`, so a shallow doc can re-export at its + /// own cut. Code deciding whether something can be imported must check + /// `shallow_since_vv` first (see `import_deps_before_shallow_root`). pub fn frontiers_to_vv(&self, frontiers: &Frontiers) -> Option { if frontiers == &self.shallow_root_frontiers_deps { let vv = VersionVector::from_im_vv(&self.shallow_since_vv); @@ -1453,6 +1457,30 @@ mod ensure_vv_for_tests { assert!(dag.import_deps_before_shallow_root(&deps)); } + /// A change whose deps are exactly the shallow root's own deps is + /// concurrent with the root. `frontiers_to_vv` resolves that set (it must, + /// so a shallow doc can re-export at its own cut), so the trimmed-dep + /// check has to win. + #[test] + fn import_deps_before_shallow_root_rejects_deps_equal_to_root_deps() { + let dag = make_shallow_dag_for_import_deps(); + let deps = Frontiers::from_id(ID::new(1, 1)); + + assert!(dag.frontiers_to_vv(&deps).is_some()); + assert!(dag.get_lamport(&ID::new(1, 1)).is_none()); + assert!(dag.import_deps_before_shallow_root(&deps)); + } + + /// The root is retained and resolvable, so a change built on it passes + /// without the boundary special case the trimmed-dep check replaced. + #[test] + fn import_deps_before_shallow_root_allows_deps_on_root() { + let dag = make_shallow_dag_for_import_deps(); + let deps = Frontiers::from_id(ID::new(1, 2)); + + assert!(!dag.import_deps_before_shallow_root(&deps)); + } + #[test] fn import_deps_before_shallow_root_allows_boundary_with_missing_peer() { let dag = make_shallow_dag_for_import_deps(); diff --git a/crates/loro/tests/integration_test/shallow_snapshot_test.rs b/crates/loro/tests/integration_test/shallow_snapshot_test.rs index abb819336..ec294af61 100644 --- a/crates/loro/tests/integration_test/shallow_snapshot_test.rs +++ b/crates/loro/tests/integration_test/shallow_snapshot_test.rs @@ -5,8 +5,8 @@ use std::{ use super::gen_action; use loro::{ - cursor::CannotFindRelativePosition, ExpandType, ExportMode, Frontiers, LoroDoc, LoroValue, - StyleConfig, StyleConfigMap, ID, + cursor::CannotFindRelativePosition, Counter, ExpandType, ExportMode, Frontiers, LoroDoc, + LoroError, LoroValue, StyleConfig, StyleConfigMap, VersionVector, ID, }; /// Byte-level scan of an exported blob. Only used for *absence* checks, and @@ -528,6 +528,117 @@ fn shallow_doc_accepts_cross_peer_op_whose_deps_include_boundary() -> anyhow::Re Ok(()) } +/// Counter of the shallow root in `import_fork_into_shallow_doc`. +const SHALLOW_ROOT: Counter = 2; + +/// Peer 1 writes three single-char ops. A shallow snapshot at `2@1` keeps +/// `2@1` as the root and trims `0@1` and `1@1`. Peer 2 forks at `fork@1` +/// and writes one op, whose deps are therefore `[fork@1]`. Returns the +/// shallow doc's import result for that op together with the doc. +fn import_fork_into_shallow_doc(fork: Counter) -> (LoroDoc, loro::LoroResult<()>) { + let doc = LoroDoc::new(); + doc.set_peer_id(1).unwrap(); + for i in 0..=SHALLOW_ROOT { + doc.get_text("t").insert(i as usize, "a").unwrap(); + doc.commit(); + } + + let root = Frontiers::from(ID::new(1, SHALLOW_ROOT)); + let snap = doc.export(ExportMode::shallow_snapshot(&root)).unwrap(); + let shallow_doc = LoroDoc::new(); + shallow_doc.import(&snap).unwrap(); + assert_eq!(shallow_doc.shallow_since_frontiers(), root); + + let forked = doc.fork_at(&Frontiers::from(ID::new(1, fork))).unwrap(); + forked.set_peer_id(2).unwrap(); + forked.get_text("t").insert(0, "b").unwrap(); + forked.commit(); + let update = forked.export(ExportMode::updates(&doc.oplog_vv())).unwrap(); + let result = shallow_doc.import(&update).map(|_| ()); + (shallow_doc, result) +} + +/// Deps strictly below the shallow root's deps are rejected. +#[test] +fn shallow_doc_rejects_op_depending_below_root() { + let (_, result) = import_fork_into_shallow_doc(SHALLOW_ROOT - 2); + assert_eq!( + result.unwrap_err(), + LoroError::ImportUpdatesThatDependsOnOutdatedVersion + ); +} + +/// Deps equal to the shallow root's own deps (`[1@1]`) make the op concurrent +/// with the root: the doc holds no state without the root op, and the dag has +/// no node, hence no lamport, for a trimmed id. This used to pass the +/// preflight, get parked as pending, and abort on +/// `calc_unknown_lamport_change(..).unwrap()` when the pending replay applied +/// it, poisoning the doc mutex. +#[test] +fn shallow_doc_rejects_op_depending_on_root_deps() { + let (shallow_doc, result) = import_fork_into_shallow_doc(SHALLOW_ROOT - 1); + assert_eq!( + result.unwrap_err(), + LoroError::ImportUpdatesThatDependsOnOutdatedVersion + ); + // The rejected import left no trace and the doc stays usable. + assert_eq!( + shallow_doc.oplog_frontiers(), + Frontiers::from(ID::new(1, SHALLOW_ROOT)) + ); + assert_eq!(shallow_doc.get_text("t").to_string(), "aaa"); + shallow_doc.get_text("t").insert(0, "c").unwrap(); + shallow_doc.commit(); + assert_eq!(shallow_doc.get_text("t").to_string(), "caaa"); +} + +/// Deps on the shallow root itself are importable. +#[test] +fn shallow_doc_accepts_op_depending_on_root() { + let (shallow_doc, result) = import_fork_into_shallow_doc(SHALLOW_ROOT); + result.unwrap(); + assert_eq!(shallow_doc.get_text("t").to_string(), "baaa"); +} + +/// Deps mixing the shallow root with a trimmed id of another peer cannot come +/// from loro itself (a frontier holds one id per peer and is minimal), but +/// they can come from hand-written JSON. They used to slip past the preflight +/// through its boundary special case and abort like the case above. +#[test] +fn shallow_doc_rejects_json_op_mixing_root_with_trimmed_dep() { + // Trimmed history spanning two peers: on top of the shallow doc holding + // `2@1` and peer 2's `0@2`, peer 3 writes one op and the doc is + // re-exported shallow at that op. + let (shallow_doc, result) = import_fork_into_shallow_doc(SHALLOW_ROOT); + result.unwrap(); + shallow_doc.set_peer_id(3).unwrap(); + shallow_doc.get_text("t").insert(0, "c").unwrap(); + shallow_doc.commit(); + let root = Frontiers::from(ID::new(3, 0)); + let snap = shallow_doc + .export(ExportMode::shallow_snapshot(&root)) + .unwrap(); + let nested = LoroDoc::new(); + nested.import(&snap).unwrap(); + let trimmed = nested.shallow_since_vv().to_vv(); + assert!(trimmed.includes_id(ID::new(1, SHALLOW_ROOT))); + assert!(trimmed.includes_id(ID::new(2, 0))); + + let other = LoroDoc::new(); + other.set_peer_id(7).unwrap(); + other.get_text("t").insert(0, "x").unwrap(); + other.commit(); + let mut json = other + .export_json_updates_without_peer_compression(&VersionVector::default(), &other.oplog_vv()); + json.changes[0].deps = vec![ID::new(3, 0), ID::new(1, SHALLOW_ROOT)]; + + assert_eq!( + nested.import_json_updates(json).unwrap_err(), + LoroError::ImportUpdatesThatDependsOnOutdatedVersion + ); + assert_eq!(nested.get_text("t").to_string(), "cbaaa"); +} + /// Shallow snapshots are documented as a content-redaction mechanism: exporting at the /// current frontiers is supposed to drop the trimmed history, leaving only the live state. /// The value of a rich-text style op whose whole range has been deleted must be dropped From b9776cfd565a12dd2f4cde1e5f6be2e66e0c431e Mon Sep 17 00:00:00 2001 From: Magnus Rex Date: Wed, 2 Sep 2026 14:33:06 +0200 Subject: [PATCH 2/2] fix: reject a parked change whose deps a later shallow snapshot trimmed A change parked as pending (its deps not imported yet) stays parked when the doc then imports a shallow snapshot: the snapshot path only requires the dag and the arena to be empty, and a parked change with no arena allocations (a map op, say) passes that. If the shallow cut trims the parked change's deps, the change is concurrent with the shallow root and can never be merged. The pending replay did not see that: `remote_change_apply_state` ignored its `shallow_vv` parameter and checked deps against the oplog vv, which covers trimmed history, so the next import that touched the parked slot applied the change and hit `calc_unknown_lamport_change(..).unwrap()` (no dag node, no lamport). The panic poisoned the doc mutex and the process aborted in a destructor. Reachable through `import`, `import_json_updates` and `import_batch`. The replay now runs the same trimmed-dep test as the import preflight (`import_deps_before_shallow_root`), shared as `AppDag::deps_reach_trimmed_history`, before looking at the oplog vv. Such a change is dropped and the unlocking import returns `ImportUpdatesThatDependsOnOutdatedVersion`, the outcome it would have had in the other import order; the valid part of the import is applied and kept, as for an import that carries a rejected change of its own. The check runs after the applied check: the shallow root's own change has trimmed deps too, and a doc can hold it parked when the snapshot brings it in. That makes the error arm of the `!applies_to_dag` branch in `import_changes_and_apply_delta_to_state_if_needed` reachable for the first time: the preflight rejects every import the apply could flag on its own, so there the flag can only come from a dropped parked change. Its arena rollback is gone, since the import's own changes are parked and reference what they allocated in the arena; with the rollback kept, a parked text op read freed arena bytes once unlocked. The detached and applying paths never rolled back on this error either. --- .../pending-change-across-shallow-cut.md | 9 + context/internal-encoding.md | 21 +- crates/loro-internal/src/encoding.rs | 13 +- crates/loro-internal/src/loro.rs | 7 +- crates/loro-internal/src/oplog.rs | 2 +- crates/loro-internal/src/oplog/loro_dag.rs | 24 ++- .../src/oplog/pending_changes.rs | 44 ++++- .../integration_test/shallow_snapshot_test.rs | 186 +++++++++++++++++- 8 files changed, 277 insertions(+), 29 deletions(-) create mode 100644 .changeset/pending-change-across-shallow-cut.md diff --git a/.changeset/pending-change-across-shallow-cut.md b/.changeset/pending-change-across-shallow-cut.md new file mode 100644 index 000000000..e70c8257e --- /dev/null +++ b/.changeset/pending-change-across-shallow-cut.md @@ -0,0 +1,9 @@ +--- +"loro-crdt": patch +--- + +A change parked as pending before the doc imported a shallow snapshot, whose deps +the snapshot then trimmed, no longer aborts the process when a later import +unlocks it. It is dropped and that import returns +`ImportUpdatesThatDependsOnOutdatedVersion`, the same outcome as importing it +after the doc became shallow. diff --git a/context/internal-encoding.md b/context/internal-encoding.md index b3ab78699..59336d029 100644 --- a/context/internal-encoding.md +++ b/context/internal-encoding.md @@ -217,15 +217,18 @@ Pre-shallow frontier safety lives in `loro.rs`: `checkout`, `diff`, and history before the shallow root. On the import side the gate is `AppDag::import_deps_before_shallow_root`, -reached from `preflight_import_changes` and from -`import_changes_to_oplog`. A change with any dep inside `shallow_since_vv` -must be rejected with `ImportUpdatesThatDependsOnOutdatedVersion`: trimmed ids -have no dag node, so no lamport can ever be computed for the change, and the -pending-change replay (`remote_change_apply_state`) checks deps against the -oplog vv, which covers trimmed history, so it would apply the change and hit -the `unwrap` in `apply_change_from_remote`. The check must run before -`frontiers_to_vv`, which resolves the root's own deps to `shallow_since_vv` -even though they are trimmed. +reached from `preflight_import_changes` and from `import_changes_to_oplog`, and +the pending replay (`pending_changes.rs:remote_change_apply_state`) applies the +same test through `AppDag::deps_reach_trimmed_history`. A change with any dep +inside `shallow_since_vv` must be rejected with +`ImportUpdatesThatDependsOnOutdatedVersion`: trimmed ids have no dag node, so no +lamport can ever be computed for the change. The preflight must run the test +before `frontiers_to_vv`, which resolves the root's own deps to +`shallow_since_vv` even though they are trimmed. The replay needs it because a +snapshot import into an empty doc leaves already parked changes in place, so a +parked change can come to depend on trimmed history after the fact; such a +change is dropped and the import that unlocked it reports the same error, as if +the change had arrived after the cut. ## JSON Updates diff --git a/crates/loro-internal/src/encoding.rs b/crates/loro-internal/src/encoding.rs index 997a44ac6..af276fc96 100644 --- a/crates/loro-internal/src/encoding.rs +++ b/crates/loro-internal/src/encoding.rs @@ -260,6 +260,8 @@ pub(crate) fn decode_oplog_changes( pub(crate) struct ApplyDecodedChangesResult { pub status: ImportStatus, + /// A change of this import, or a parked change this import unlocked, depends + /// on trimmed history and was dropped. pub has_deps_before_shallow_root: bool, } @@ -275,18 +277,23 @@ pub(crate) fn apply_decoded_changes_to_oplog( } = import_changes_to_oplog(changes, oplog); // TODO: PERF: should we use hashmap to filter latest_ids with the same peer first? - oplog.try_apply_pending(latest_ids, Some(&mut imported)); + // A parked change unlocked here can turn out to depend on trimmed history + // (parked before the doc became shallow); it is dropped and reported like a + // rejected change of this import. + let mut dropped_trimmed = oplog.try_apply_pending(latest_ids, Some(&mut imported)); // Applying previously parked pending ops can unlock deps of `pending_changes`. // Those are applied here (and counted in `imported`); only still-blocked ones // remain in the returned pending range. - let pending = + let (pending, dropped) = oplog.import_unknown_lamport_pending_changes(pending_changes, Some(&mut imported)); + dropped_trimmed |= dropped; ApplyDecodedChangesResult { status: ImportStatus { success: imported, pending: (!pending.is_empty()).then_some(pending), }, - has_deps_before_shallow_root: !changes_that_have_deps_before_shallow_root.is_empty(), + has_deps_before_shallow_root: !changes_that_have_deps_before_shallow_root.is_empty() + || dropped_trimmed, } } diff --git a/crates/loro-internal/src/loro.rs b/crates/loro-internal/src/loro.rs index 6c4c532d4..451b22222 100644 --- a/crates/loro-internal/src/loro.rs +++ b/crates/loro-internal/src/loro.rs @@ -781,8 +781,13 @@ impl LoroDoc { if !preflight.applies_to_dag { let pending_root_containers = pending_root_containers_to_materialize(&oplog, &changes); let result = encoding::apply_decoded_changes_to_oplog(&mut oplog, changes); + // The preflight above already rejected this import if any of its own + // changes depends on trimmed history, so here the flag can only come + // from a previously parked change that the replay dropped. This + // import's changes are parked and keep referencing what they + // allocated in the arena, so the arena is not rolled back, as in + // the detached and applying paths. if result.has_deps_before_shallow_root { - oplog.arena.rollback(arena_checkpoint); return Err(LoroError::ImportUpdatesThatDependsOnOutdatedVersion); } diff --git a/crates/loro-internal/src/oplog.rs b/crates/loro-internal/src/oplog.rs index 6351c9456..1b06bf0af 100644 --- a/crates/loro-internal/src/oplog.rs +++ b/crates/loro-internal/src/oplog.rs @@ -559,7 +559,7 @@ impl OpLog { &mut self, remote_changes: Vec, would_affect: Option<&mut crate::version::VersionRange>, - ) -> crate::version::VersionRange { + ) -> (crate::version::VersionRange, bool) { self.extend_pending_changes_with_unknown_lamport(remote_changes, would_affect) } diff --git a/crates/loro-internal/src/oplog/loro_dag.rs b/crates/loro-internal/src/oplog/loro_dag.rs index 6cf58330b..81dc38996 100644 --- a/crates/loro-internal/src/oplog/loro_dag.rs +++ b/crates/loro-internal/src/oplog/loro_dag.rs @@ -793,6 +793,18 @@ impl AppDag { false } + /// Whether any dep is trimmed history. Such a dep has no dag node, so the + /// change can never get a lamport and could only abort or park forever: it + /// is concurrent with the shallow root and must be rejected with + /// `ImportUpdatesThatDependsOnOutdatedVersion`. Associated rather than a + /// method so the pending replay can call it with the vv it already holds. + pub(crate) fn deps_reach_trimmed_history( + shallow_since_vv: &ImVersionVector, + deps: &Frontiers, + ) -> bool { + deps.iter().any(|id| shallow_since_vv.includes_id(id)) + } + pub(crate) fn import_deps_before_shallow_root(&self, deps: &Frontiers) -> bool { if self.shallow_since_vv.is_empty() { return false; @@ -802,13 +814,11 @@ impl AppDag { return true; } - // A trimmed dep has no dag node, so the change can never get a lamport - // and could only abort or park forever. This has to be decided before - // `frontiers_to_vv`: the root's own deps are all trimmed, yet - // `frontiers_to_vv` resolves exactly that set to `shallow_since_vv`, - // which would pass the inclusion check below even though such a change - // is concurrent with the root. - if deps.iter().any(|id| self.shallow_since_vv.includes_id(id)) { + // This has to be decided before `frontiers_to_vv`: the root's own deps + // are all trimmed, yet `frontiers_to_vv` resolves exactly that set to + // `shallow_since_vv`, which would pass the inclusion check below even + // though such a change is concurrent with the root. + if Self::deps_reach_trimmed_history(&self.shallow_since_vv, deps) { return true; } diff --git a/crates/loro-internal/src/oplog/pending_changes.rs b/crates/loro-internal/src/oplog/pending_changes.rs index a7d251f72..7563d4bb6 100644 --- a/crates/loro-internal/src/oplog/pending_changes.rs +++ b/crates/loro-internal/src/oplog/pending_changes.rs @@ -5,6 +5,7 @@ use std::{ use crate::{ change::Change, + oplog::AppDag, version::{ImVersionVector, VersionRange}, OpLog, VersionVector, }; @@ -212,14 +213,17 @@ impl OpLog { /// later B change depended on). Treat that as normal: apply when possible, skip if /// already present, and only park changes that are still waiting on a missing dep. /// - /// Returns the version range of changes from `remote_changes` that remain pending. + /// Returns the version range of changes from `remote_changes` that remain pending, + /// and whether a change was dropped for depending on trimmed history (see + /// [`Self::try_apply_pending`]). pub(super) fn extend_pending_changes_with_unknown_lamport( &mut self, remote_changes: Vec, mut would_affect: Option<&mut VersionRange>, - ) -> VersionRange { + ) -> (VersionRange, bool) { let mut parked = Vec::new(); let mut newly_applied_ids = Vec::new(); + let mut dropped_trimmed = false; for change in remote_changes { let local_change = PendingChange::Unknown(change); @@ -233,11 +237,12 @@ impl OpLog { newly_applied_ids.push(local_change.id_last()); self.apply_change_from_remote(local_change, would_affect.as_deref_mut()); } + ChangeState::DependsOnTrimmedHistory => dropped_trimmed = true, } } if !newly_applied_ids.is_empty() { - self.try_apply_pending(newly_applied_ids, would_affect); + dropped_trimmed |= self.try_apply_pending(newly_applied_ids, would_affect); } // A parked change can already be partially covered by the oplog VV: a change whose @@ -262,7 +267,7 @@ impl OpLog { } } - still_pending + (still_pending, dropped_trimmed) } } @@ -270,11 +275,20 @@ impl OpLog { /// Try to apply pending changes. /// /// `new_ids` are the ID of the op that is just applied. + /// + /// Returns whether a parked change was dropped because it depends on trimmed + /// history. The caller reports that as `ImportUpdatesThatDependsOnOutdatedVersion`, + /// the same outcome the change would have had if it had arrived after the doc + /// became shallow, even though the import being reported may be sound itself. + /// Changes parked on the dropped one stay parked, as they do behind any dep + /// that never arrives. + #[must_use] pub(crate) fn try_apply_pending( &mut self, mut new_ids: Vec, mut would_affect: Option<&mut VersionRange>, - ) { + ) -> bool { + let mut dropped_trimmed = false; while let Some(id) = new_ids.pop() { let Some(tree) = self.pending_changes.changes.get_mut(&id.peer) else { continue; @@ -318,10 +332,13 @@ impl OpLog { ChangeState::AwaitingMissingDependency(miss_dep) => { self.push_pending_change(miss_dep, pending_change) } + ChangeState::DependsOnTrimmedHistory => dropped_trimmed = true, } } } } + + dropped_trimmed } pub(super) fn apply_change_from_remote( @@ -356,11 +373,18 @@ enum ChangeState { CanApplyDirectly, // The id of first missing dep AwaitingMissingDependency(ID), + /// A dep is trimmed history: it has no dag node, so the change can never get + /// a lamport. It is concurrent with the shallow root and is dropped with the + /// error the import preflight (`AppDag::import_deps_before_shallow_root`) + /// gives a change that arrives after the cut. The whole change is dropped: + /// a change straddling the cut with an applicable tail cannot reach here, + /// since a shallow snapshot never retains ops concurrent with its root. + DependsOnTrimmedHistory, } fn remote_change_apply_state( vv: &VersionVector, - _shallow_vv: &ImVersionVector, + shallow_vv: &ImVersionVector, change: &Change, ) -> ChangeState { let peer = change.id.peer; @@ -370,6 +394,14 @@ fn remote_change_apply_state( return ChangeState::Applied; } + // The oplog vv covers trimmed history, so the dep loop below would take a + // trimmed dep for satisfied. A change parked before the doc became shallow + // can hold one: a snapshot import into an empty doc leaves parked changes + // where they are. + if AppDag::deps_reach_trimmed_history(shallow_vv, &change.deps) { + return ChangeState::DependsOnTrimmedHistory; + } + if vv_latest_ctr < start { return ChangeState::AwaitingMissingDependency(change.id.inc(-1)); } diff --git a/crates/loro/tests/integration_test/shallow_snapshot_test.rs b/crates/loro/tests/integration_test/shallow_snapshot_test.rs index ec294af61..ac848e253 100644 --- a/crates/loro/tests/integration_test/shallow_snapshot_test.rs +++ b/crates/loro/tests/integration_test/shallow_snapshot_test.rs @@ -5,8 +5,8 @@ use std::{ use super::gen_action; use loro::{ - cursor::CannotFindRelativePosition, Counter, ExpandType, ExportMode, Frontiers, LoroDoc, - LoroError, LoroValue, StyleConfig, StyleConfigMap, VersionVector, ID, + cursor::CannotFindRelativePosition, Counter, ExpandType, ExportMode, Frontiers, IdSpan, + LoroDoc, LoroError, LoroValue, StyleConfig, StyleConfigMap, VersionVector, ID, }; /// Byte-level scan of an exported blob. Only used for *absence* checks, and @@ -882,3 +882,185 @@ fn shallow_snapshot_redacts_dead_styles_for_all_non_both_expands() -> anyhow::Re } Ok(()) } + +/// Peer 1 writes `0@1` and `1@1`; peer 2 forks at `0@1` and writes one op with +/// deps `[0@1]`. A fresh doc parks peer 2's op (its dep is unknown), then +/// imports a shallow snapshot rooted at `1@1`, which trims `0@1`. The parked op +/// now depends on trimmed history and can never be merged. +struct ParkedBelowCut { + /// The shallow doc holding the parked op. + doc: LoroDoc, + /// Peer 1's doc, one op (`2@1`) ahead of the shallow doc. + p1: LoroDoc, + /// The shallow doc's version: `p1.export(updates(&cut_vv))` is `2@1`. + cut_vv: VersionVector, + /// The update that parked peer 2's op. + parked: Vec, +} + +fn shallow_doc_with_parked_change_below_the_cut() -> ParkedBelowCut { + let p1 = LoroDoc::new(); + p1.set_peer_id(1).unwrap(); + p1.get_text("t").insert(0, "a").unwrap(); + p1.commit(); + p1.get_text("t").insert(1, "b").unwrap(); + p1.commit(); + + let p2 = p1.fork_at(&Frontiers::from(ID::new(1, 0))).unwrap(); + p2.set_peer_id(2).unwrap(); + // A map op allocates nothing in the arena, so the doc still counts as + // empty when the snapshot arrives and imports as a snapshot. + p2.get_map("m").insert("k", 1).unwrap(); + p2.commit(); + let parked = p2.export(ExportMode::updates(&p1.oplog_vv())).unwrap(); + + let root = Frontiers::from(ID::new(1, 1)); + let snap = p1.export(ExportMode::shallow_snapshot(&root)).unwrap(); + let cut_vv = p1.oplog_vv(); + p1.get_text("t").insert(2, "c").unwrap(); + p1.commit(); + + let doc = LoroDoc::new(); + let status = doc.import(&parked).unwrap(); + assert!(status.pending.is_some(), "{status:?}"); + doc.import(&snap).unwrap(); + assert_eq!(doc.shallow_since_frontiers(), root); + assert!(doc.shallow_since_vv().to_vv().includes_id(ID::new(1, 0))); + assert_eq!(doc.oplog_vv(), cut_vv); + ParkedBelowCut { + doc, + p1, + cut_vv, + parked, + } +} + +/// The parked op is gone for good: importing it again is rejected rather than +/// parked, and a later peer-1 op revisiting its pending slot imports cleanly. +/// The unlocking op itself was applied, as for any import that arrives +/// together with a rejected change, and the doc stays usable. +fn assert_parked_change_dropped(fx: &ParkedBelowCut) { + let ParkedBelowCut { + doc, p1, parked, .. + } = fx; + assert_eq!(doc.oplog_vv(), p1.oplog_vv()); + assert_eq!(doc.state_frontiers(), doc.oplog_frontiers()); + assert_eq!(doc.get_text("t").to_string(), p1.get_text("t").to_string()); + assert_eq!( + doc.import(parked).unwrap_err(), + LoroError::ImportUpdatesThatDependsOnOutdatedVersion + ); + assert_eq!(doc.oplog_vv(), p1.oplog_vv()); + + let t1 = p1.get_text("t"); + t1.insert(t1.len_unicode(), "z").unwrap(); + p1.commit(); + let status = doc + .import(&p1.export(ExportMode::updates(&doc.oplog_vv())).unwrap()) + .unwrap(); + assert!(status.pending.is_none(), "{status:?}"); + assert_eq!(doc.get_text("t").to_string(), t1.to_string()); +} + +/// A change parked before the doc became shallow, whose deps the shallow cut +/// then trimmed, is concurrent with the shallow root. The update that unlocks +/// it used to apply it anyway and abort on +/// `calc_unknown_lamport_change(..).unwrap()`, poisoning the doc mutex. +#[test] +fn parked_change_below_shallow_cut_is_rejected_when_unlocked() { + let fx = shallow_doc_with_parked_change_below_the_cut(); + let unlock = fx.p1.export(ExportMode::updates(&fx.cut_vv)).unwrap(); + assert_eq!( + fx.doc.import(&unlock).unwrap_err(), + LoroError::ImportUpdatesThatDependsOnOutdatedVersion + ); + assert_eq!(fx.doc.get_text("t").to_string(), "abc"); + assert_parked_change_dropped(&fx); +} + +#[test] +fn parked_change_below_shallow_cut_is_rejected_when_unlocked_by_json() { + let fx = shallow_doc_with_parked_change_below_the_cut(); + let unlock = fx.p1.export_json_updates(&fx.cut_vv, &fx.p1.oplog_vv()); + assert_eq!( + fx.doc.import_json_updates(unlock).unwrap_err(), + LoroError::ImportUpdatesThatDependsOnOutdatedVersion + ); + assert_parked_change_dropped(&fx); +} + +/// `import_batch` runs its blobs detached and hands a single blob to `import`; +/// the empty second blob keeps it on the batch path. The reattach at the end +/// must still happen and report the error. +#[test] +fn parked_change_below_shallow_cut_is_rejected_when_unlocked_by_batch() { + let fx = shallow_doc_with_parked_change_below_the_cut(); + let unlock = fx.p1.export(ExportMode::updates(&fx.cut_vv)).unwrap(); + let empty = LoroDoc::new().export(ExportMode::all_updates()).unwrap(); + assert_eq!( + fx.doc.import_batch(&[unlock, empty]).unwrap_err(), + LoroError::ImportUpdatesThatDependsOnOutdatedVersion + ); + assert!(!fx.doc.is_detached()); + assert_parked_change_dropped(&fx); +} + +/// An import that parks all of its own changes still revisits the pending +/// slot of the parked op and drops it. That import applies nothing, so it +/// takes the path that applies no state diff; rolling the arena back on its +/// error, as that path used to, would leave the import's own parked text op +/// pointing at freed arena bytes ("abcc" instead of "abcd" once unlocked). +#[test] +fn parked_change_below_shallow_cut_is_rejected_when_dropped_by_an_import_that_only_parks() { + let fx = shallow_doc_with_parked_change_below_the_cut(); + fx.p1.get_text("t").insert(3, "d").unwrap(); + fx.p1.commit(); + let only_3 = fx + .p1 + .export(ExportMode::updates_in_range(vec![IdSpan::new(1, 3, 4)])) + .unwrap(); + let only_2 = fx + .p1 + .export(ExportMode::updates_in_range(vec![IdSpan::new(1, 2, 3)])) + .unwrap(); + + assert_eq!( + fx.doc.import(&only_3).unwrap_err(), + LoroError::ImportUpdatesThatDependsOnOutdatedVersion + ); + assert_eq!(fx.doc.oplog_vv(), fx.cut_vv); + fx.doc.import(&only_2).unwrap(); + assert_eq!(fx.doc.get_text("t").to_string(), "abcd"); + assert_parked_change_dropped(&fx); +} + +/// The shallow root's own change depends on trimmed history too, and a doc can +/// hold it parked (imported before the snapshot) while the snapshot then brings +/// it in. Revisiting it must find it applied, not reject it: the applied check +/// comes before the trimmed-dep check. +#[test] +fn parked_change_that_the_shallow_snapshot_then_applied_is_not_rejected() { + let p1 = LoroDoc::new(); + p1.set_peer_id(1).unwrap(); + p1.get_map("a").insert("k", 0).unwrap(); + p1.commit(); + p1.get_map("b").insert("k", 1).unwrap(); + p1.commit(); + let root = Frontiers::from(ID::new(1, 1)); + let snap = p1.export(ExportMode::shallow_snapshot(&root)).unwrap(); + let cut_vv = p1.oplog_vv(); + p1.get_map("c").insert("k", 2).unwrap(); + p1.commit(); + let only_1 = p1 + .export(ExportMode::updates_in_range(vec![IdSpan::new(1, 1, 2)])) + .unwrap(); + + let doc = LoroDoc::new(); + assert!(doc.import(&only_1).unwrap().pending.is_some()); + doc.import(&snap).unwrap(); + assert_eq!(doc.shallow_since_frontiers(), root); + doc.import(&p1.export(ExportMode::updates(&cut_vv)).unwrap()) + .unwrap(); + assert_eq!(doc.oplog_vv(), p1.oplog_vv()); + assert_eq!(doc.get_deep_value(), p1.get_deep_value()); +}