fix: enforce shallow snapshot concurrency boundaries

.changeset/shallow-import-root-dependencies.md

@@ -0,0 +1,8 @@
+---
+"loro-crdt": patch
+---
+
+Reject updates concurrent with the shallow root frontier with
+ImportUpdatesThatDependsOnOutdatedVersion instead of queuing them and panicking
+while resolving a dependency that has already been trimmed. Rejected updates do
+not enter pending storage; subsequent valid post-root updates still apply.

context/internal-encoding.md

@@ -216,6 +216,11 @@ Pre-shallow frontier safety lives in `loro.rs`: `checkout`, `diff`, and
 `revert_to` must return `SwitchToVersionBeforeShallowRoot` instead of traversing
 history before the shallow root.
 
+Merge semantics of a shallow replica meeting concurrent full-history peers
+(which updates apply, pend, or are rejected, and why a never-synced peer can
+never merge): [docs/shallow-snapshot-concurrency.md](../docs/shallow-snapshot-concurrency.md)
+with tests in `crates/loro/tests/shallow_snapshot_concurrency.rs`.
+
 ## JSON Updates
 
 `json_schema.rs` is not wrapped in the binary `loro` envelope. Its

crates/loro-internal/src/oplog/loro_dag.rs

@@ -802,6 +802,17 @@ impl AppDag {
             return true;
         }
 
+        // Deps equal to the root's own deps describe a change CONCURRENT with
+        // the root frontier op: its causal past is covered by the root state,
+        // so the boundary shortcut in `frontiers_to_vv` would resolve it to
+        // the shallow vv below. But the dep ids themselves are trimmed from
+        // the dag, so no lamport can be computed for such a change — it would
+        // be parked as pending and then panic in `calc_unknown_lamport_change`.
+        // Reject it like any other pre-root update.
+        if deps == &self.shallow_root_frontiers_deps {
+            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);
@@ -1453,6 +1464,21 @@ mod ensure_vv_for_tests {
         assert!(dag.import_deps_before_shallow_root(&deps));
     }
 
+    /// Deps exactly equal to the root's own deps describe a change concurrent
+    /// with the root frontier op. The boundary shortcut in `frontiers_to_vv`
+    /// resolves them to the shallow vv, but the dep ids are trimmed from the
+    /// dag, so the change could never get a lamport — reject it.
+    #[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));
+
+        // The boundary shortcut resolves these deps to the shallow vv...
+        assert!(dag.frontiers_to_vv(&deps).is_some());
+        // ...but the import check must still reject them.
+        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/tests/import_atomicity.rs

@@ -666,3 +666,48 @@ fn snapshot_import_rejects_corrupt_inner_sstable_with_valid_envelope_checksum()
             .is_some_and(|value| value.is_empty()));
     }
 }
+
+/// A change whose deps are before the shallow root is rejected AND dropped:
+/// it must not linger in the pending store, where it could never be unlocked.
+/// Locks in the "dropped, not pending" guarantee documented by
+/// `loro/tests/shallow_snapshot_concurrency.rs`.
+#[test]
+fn outdated_update_on_shallow_doc_is_dropped_not_pending() {
+    let a = LoroDoc::new_auto_commit();
+    a.set_peer_id(1).unwrap();
+    a.get_map("m").insert("a", 1).unwrap();
+    a.commit_then_renew();
+    let v_vv = a.oplog_vv();
+    let v_frontiers = a.oplog_frontiers();
+    a.get_map("m").insert("b", 2).unwrap();
+    a.commit_then_renew();
+    let f = a.oplog_frontiers();
+    a.get_map("m").insert("c", 3).unwrap();
+    a.commit_then_renew();
+
+    // B is bootstrapped from the shallow snapshot at F.
+    let b = LoroDoc::new();
+    b.import(&a.export(ExportMode::shallow_snapshot(&f)).unwrap())
+        .unwrap();
+    assert_eq!(pending_len(&b), 0);
+
+    // C holds full history up to V and edits on top of it, concurrent with F.
+    let c = LoroDoc::new();
+    c.import(&a.export(ExportMode::snapshot_at(&v_frontiers)).unwrap())
+        .unwrap();
+    c.set_peer_id(2).unwrap();
+    c.get_map("m").insert("from_c", true).unwrap();
+    c.commit_then_renew();
+    let updates = c.export(ExportMode::updates(&v_vv)).unwrap();
+
+    let err = b.import(&updates).unwrap_err();
+    assert!(matches!(
+        err,
+        LoroError::ImportUpdatesThatDependsOnOutdatedVersion
+    ));
+    assert_eq!(
+        pending_len(&b),
+        0,
+        "outdated changes must be dropped, not parked as pending"
+    );
+}