fix: reject shallow imports that depend on trimmed history instead of aborting

.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.

.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.

context/internal-encoding.md

@@ -216,6 +216,20 @@ 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`, 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
 
 `json_schema.rs` is not wrapped in the binary `loro` envelope. Its

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,
     }
 }
 

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);
             }
 

crates/loro-internal/src/oplog.rs

@@ -559,7 +559,7 @@ impl OpLog {
         &mut self,
         remote_changes: Vec<Change>,
         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)
     }
 

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,23 +814,21 @@ 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;
+        // 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;
         }
 
-        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 +1198,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<VersionVector> {
         if frontiers == &self.shallow_root_frontiers_deps {
             let vv = VersionVector::from_im_vv(&self.shallow_since_vv);
@@ -1453,6 +1467,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();

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<Change>,
         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,19 +267,28 @@ impl OpLog {
             }
         }
 
-        still_pending
+        (still_pending, dropped_trimmed)
     }
 }
 
 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<ID>,
         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));
     }