fix: reject imports that reuse local op ids with different content (#1118)

.changeset/import-reused-peer-id.md

@@ -0,0 +1,5 @@
+---
+"loro-crdt": patch
+---
+
+Importing updates or a snapshot that reuse op ids the document already has for different content (two clients that shared a peer id) now throws a "ID ... has been used" error and leaves the document unchanged and usable. Previously the import trapped the WASM instance (`RuntimeError: unreachable`) or silently scrambled text. Imports with a change that skips counters of its peer also throw instead of trapping.

AGENTS.md

@@ -24,6 +24,9 @@ Loro is a Rust CRDT workspace with JS/WASM packaging and a MoonBit codec.
   [context/mergeable-containers.md](context/mergeable-containers.md).
 - `import_batch` force-detach, batch-wide rollback scope, and the never-exit-detached
   invariant: [context/import-batch-atomicity.md](context/import-batch-atomicity.md).
+- Imports that reuse local op ids (a shared peer id): the known-history check,
+  what counts as equal, and what it cannot catch:
+  [context/import-peer-id-reuse.md](context/import-peer-id-reuse.md).
 - Movable-list `Move`/`Set` import validation, and why a move of a deleted element
   is applied rather than rejected:
   [context/movable-list-op-validation.md](context/movable-list-op-validation.md).

context/import-peer-id-reuse.md

@@ -0,0 +1,90 @@
+# Imports That Reuse Local Op Ids
+
+Verified against code 2026-09-30.
+
+An import skips the part of each change the doc already has by version vector and
+applies the rest on top of the local history. When two clients shared a peer id
+(loro-dev/loro#1118), the "known" part is different history. The rest was then
+applied to a state it was not written against. It either panicked inside
+`RichtextState::apply_diff` while the doc locks were held, poisoning them (a later
+`Drop` then aborted the process), or it returned `Ok` with scrambled text. Since
+loro-dev/loro#1160, `RichtextState::validate_diff` rejects the out-of-range
+case after the fact. The check below rejects both cases before anything changes,
+and names the reused id.
+
+## The check
+
+`OpLog::check_and_trim_known_part_of_changes` (`crates/loro-internal/src/oplog/known_history.rs`)
+runs in `LoroDoc::import_changes_and_apply_delta_to_state_if_needed` right after
+decode, before the preflight and before anything is mutated. That function is the
+single entry for binary updates, snapshots imported into a non-empty doc, JSON
+updates, and each blob of `import_batch`. An error takes the existing early-exit
+path, which rolls back only the arena (decode may have registered containers).
+
+1. **Content.** If the import brings anything new, every imported change that
+   overlaps the local version vector is compared with the local changes over the
+   overlap. A mismatch returns `LoroError::UsedOpID { id }` with the first
+   differing id.
+2. **Trim.** The known parts are dropped, as the decoders used to do.
+3. **Skipped counters.** A change whose deps are present but whose peer's
+   previous counter is not is rejected with `DecodeError`. The main pass of
+   `import_changes_to_oplog` would apply it and trip the counter assertion in
+   `AppDag::update_version_on_new_change` under the op log lock. The check replays
+   that pass on counter ends only. Parked pending changes already wait for their
+   own predecessor (`remote_change_apply_state`).
+
+The decoders (`ChangeStore::decode_block_bytes`, `decode_snapshot_for_updates`)
+therefore no longer trim. A snapshot with nothing new still returns no changes,
+without cloning them.
+
+## What "equal" means
+
+Both sides can split and merge the same history differently, so the comparison
+is per atom range, not per change:
+
+- **Deps** are compared at every counter where either side starts a change. Inside
+  a change the deps are implicitly `p@c-1`, and changes only merge when that holds,
+  so `deps_at` is canonical. Lamport and timestamp are not compared: the import
+  recomputes lamports from deps, and a merged change keeps only its first timestamp.
+- **Ops** are cut at every op boundary of either side and compared by meaning
+  (`op_eq`): list insert values through the arena (offsets differ), text by bytes
+  (`unicode_start` is an arena offset), and a one-atom delete ignores the sign of
+  `signed_len`, because slicing a reversed delete down to one atom yields `-1`
+  where a forward one yields `1`.
+- **JSON imports** use `ImportedValues::Lossy`: JSON text does not round-trip every
+  value (`NaN` becomes `null`, binary may come back as a list), so value payloads
+  are not compared there. Containers, positions, text, keys, ids and deps still are.
+- History below the shallow root, or any local change `get_change` cannot find,
+  is not compared. The import then behaves as it did before this check.
+
+## What it does not catch
+
+- **Delta sync.** If the conflicting prefix is not in the import (the sender
+  exported from the receiver's version vector), nothing can be compared: the
+  receiver's vv already claims those ids. A tail position past the end of a
+  sequence is then rejected by `validate_diff` and rolled back (loro-dev/loro#1160).
+  A tail that happens to land on valid positions is applied and scrambles the
+  state. Detecting that needs content hashes in the version vector.
+- **Imports with nothing new.** A conflicting duplicate that brings nothing new
+  is a no-op and is not reported. This keeps a no-op re-import as cheap as before.
+- **Same-peer changes that are concurrent but contiguous** (for example `p@2`
+  with deps `[]` on top of a local `p@0..=1`). They do not crash, and are still
+  accepted.
+- A peer id reused by a local commit after importing is outside import.
+
+## Cost
+
+Imports that don't overlap local history, or overlap only a little, cost the same
+as before. An import that re-sends all known history *and* brings new changes
+pays for comparing the overlap. On a 10k-change doc that was about +15% (55 → 64 ms)
+for both "all updates + 1 change" and "snapshot + 1 change". Probe:
+`crates/examples/examples/import_known_history_perf.rs`.
+
+## Tests
+
+- `crates/loro/tests/import_reused_peer_id.rs`: the #1118 repro through
+  updates, snapshot, `import_batch` and JSON; a same-length conflict that used to
+  scramble; a conflicting known change followed by a new one; skipped counters;
+  and re-imports that must still succeed (piecewise vs merged change stores, every
+  op kind including `NaN` and one-atom reversed deletes, shallow docs).
+- `crates/loro-wasm/tests/import_reused_peer_id.test.ts`.

crates/examples/examples/import_known_history_perf.rs

@@ -0,0 +1,150 @@
+//! Ad-hoc perf probe for imports that overlap history the doc already has
+//! (`OpLog::check_and_trim_known_part_of_changes`, loro-dev/loro#1118).
+//!
+//! Run on two revisions and compare. Not part of CI.
+use loro::{ExportMode, LoroDoc, VersionVector};
+use std::time::{Duration, Instant};
+
+/// Two peers editing text, a list and a map, syncing every few commits so the
+/// history has many changes with cross-peer deps.
+fn make_history(commits: usize) -> (LoroDoc, LoroDoc) {
+    let a = LoroDoc::new();
+    a.set_peer_id(1).unwrap();
+    let b = LoroDoc::new();
+    b.set_peer_id(2).unwrap();
+    for i in 0..commits {
+        let doc = if i % 2 == 0 { &a } else { &b };
+        let text = doc.get_text("text");
+        let len = text.len_unicode();
+        text.insert((i * 7) % (len + 1), "hello ").unwrap();
+        if len > 20 && i % 3 == 0 {
+            text.delete((i * 5) % (len - 10), 3).unwrap();
+        }
+        doc.get_list("list").push(i as i64).unwrap();
+        doc.get_map("map")
+            .insert(&format!("k{}", i % 100), i as i64)
+            .unwrap();
+        doc.commit();
+        if i % 4 == 3 {
+            a.import(&b.export(ExportMode::updates(&a.oplog_vv())).unwrap())
+                .unwrap();
+            b.import(&a.export(ExportMode::updates(&b.oplog_vv())).unwrap())
+                .unwrap();
+        }
+    }
+    a.import(&b.export(ExportMode::updates(&a.oplog_vv())).unwrap())
+        .unwrap();
+    (a, b)
+}
+
+fn time<T>(rounds: usize, mut f: impl FnMut() -> T) -> Duration {
+    let mut best = Duration::MAX;
+    for _ in 0..rounds {
+        let start = Instant::now();
+        std::hint::black_box(f());
+        best = best.min(start.elapsed());
+    }
+    best
+}
+
+fn main() {
+    let commits: usize = std::env::var("COMMITS")
+        .ok()
+        .and_then(|x| x.parse().ok())
+        .unwrap_or(20_000);
+    let rounds = 7;
+    let (source, _) = make_history(commits);
+    let base_snapshot = source.export(ExportMode::Snapshot).unwrap();
+    let base_vv = source.oplog_vv();
+    let fresh_base = || {
+        let doc = LoroDoc::new();
+        doc.import(&base_snapshot).unwrap();
+        doc
+    };
+
+    // One more change on top, from a third peer.
+    let next = LoroDoc::new();
+    next.import(&base_snapshot).unwrap();
+    next.set_peer_id(3).unwrap();
+    next.get_text("text").insert(0, "new").unwrap();
+    next.commit();
+    let all_plus_one = next.export(ExportMode::all_updates()).unwrap();
+    let snapshot_plus_one = next.export(ExportMode::Snapshot).unwrap();
+    let only_new = next.export(ExportMode::updates(&base_vv)).unwrap();
+    let all_known = source.export(ExportMode::all_updates()).unwrap();
+
+    // The last 20 commits' worth of history re-sent with the new change.
+    let mut resend_from = base_vv.clone();
+    for (_, counter) in resend_from.iter_mut() {
+        *counter = (*counter - 30).max(0);
+    }
+    let resend_plus_one = next.export(ExportMode::updates(&resend_from)).unwrap();
+
+    // Many small non-overlapping updates, imported one by one.
+    let (inc_src, _) = make_history(2_000);
+    let mut blobs = Vec::new();
+    let mut vv = VersionVector::default();
+    {
+        let replay = LoroDoc::new();
+        let json = inc_src.export_json_updates(&Default::default(), &inc_src.oplog_vv());
+        for change in json.changes {
+            replay
+                .import_json_updates(json_schema_with(&json.peers, change))
+                .unwrap();
+            blobs.push(replay.export(ExportMode::updates(&vv)).unwrap());
+            vv = replay.oplog_vv();
+        }
+    }
+
+    let load = std::fs::read_to_string("/proc/loadavg").ok().or_else(|| {
+        std::process::Command::new("sysctl")
+            .args(["-n", "vm.loadavg"])
+            .output()
+            .ok()
+            .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
+    });
+    println!(
+        "commits={commits} changes={} loadavg={}",
+        source.len_changes(),
+        load.unwrap_or_default()
+    );
+
+    let report =
+        |name: &str, d: Duration| println!("{name:<40} {:>10.3} ms", d.as_secs_f64() * 1e3);
+    report(
+        "incremental: 2k small blobs one by one",
+        time(rounds, || {
+            let doc = LoroDoc::new();
+            for b in &blobs {
+                doc.import(b).unwrap();
+            }
+            doc
+        }),
+    );
+    // Times the import alone, into a doc freshly loaded from the base snapshot.
+    let run = |name: &str, bytes: &[u8]| {
+        let mut best = Duration::MAX;
+        for _ in 0..rounds {
+            let doc = fresh_base();
+            let start = Instant::now();
+            doc.import(bytes).unwrap();
+            best = best.min(start.elapsed());
+        }
+        report(name, best);
+    };
+    run("only new change (no overlap)", &only_new);
+    run("re-sent tail + new change", &resend_plus_one);
+    run("all updates, all known (no-op)", &all_known);
+    run("all updates + new change", &all_plus_one);
+    run("snapshot, all known (no-op)", &base_snapshot);
+    run("snapshot + new change", &snapshot_plus_one);
+}
+
+fn json_schema_with(peers: &Option<Vec<u64>>, change: loro::JsonChange) -> loro::JsonSchema {
+    loro::JsonSchema {
+        schema_version: 1,
+        start_version: Default::default(),
+        peers: peers.clone(),
+        changes: vec![change],
+    }
+}

crates/loro-internal/AGENTS.md

@@ -52,6 +52,9 @@ over graceful degradation.
 - Movable-list `Move`/`Set` element validation on import, plus change-store
   rollback records and KV block-range decoding:
   [../../context/movable-list-op-validation.md](../../context/movable-list-op-validation.md).
+- Imported changes that reuse local op ids are checked against local history
+  before anything is applied (`src/oplog/known_history.rs`):
+  [../../context/import-peer-id-reuse.md](../../context/import-peer-id-reuse.md).
 - `import_batch` force-detach, its batch-wide rollback scope, and the
   never-exit-detached invariant:
   [../../context/import-batch-atomicity.md](../../context/import-batch-atomicity.md).

crates/loro-internal/src/arena.rs

@@ -630,6 +630,15 @@ impl SharedArena {
         (self.inner.values.lock()[range]).to_vec()
     }
 
+    /// Whether the values in two ranges are equal, without cloning them.
+    pub(crate) fn value_slices_eq(&self, a: Range<usize>, b: Range<usize>) -> bool {
+        let values = self.inner.values.lock();
+        match (values.get(a), values.get(b)) {
+            (Some(a), Some(b)) => a == b,
+            _ => false,
+        }
+    }
+
     /// Borrow the values in `range` without cloning them (unlike
     /// [`Self::get_values`], which clones into a fresh `Vec`).
     #[inline]

crates/loro-internal/src/encoding/fast_snapshot.rs

@@ -383,7 +383,6 @@ pub(crate) fn encode_updates_in_range<W: std::io::Write>(
 pub(crate) fn decode_updates(oplog: &mut OpLog, body: Bytes) -> Result<Vec<Change>, LoroError> {
     let mut reader: &[u8] = body.as_ref();
     let mut index = 0;
-    let self_vv = oplog.vv();
     let mut changes = Vec::new();
     while !reader.is_empty() {
         let old_reader_len = reader.len();
@@ -400,7 +399,7 @@ pub(crate) fn decode_updates(oplog: &mut OpLog, body: Bytes) -> Result<Vec<Chang
             ));
         }
         let block_bytes = body.slice(index..end);
-        let new_changes = ChangeStore::decode_block_bytes(block_bytes, &oplog.arena, self_vv)?;
+        let new_changes = ChangeStore::decode_block_bytes(block_bytes, &oplog.arena)?;
         changes.extend(new_changes);
         index = end;
         reader = &reader[len..];

crates/loro-internal/src/loro.rs

@@ -27,7 +27,7 @@ use crate::{
     id::PeerID,
     json::JsonChange,
     op::InnerContent,
-    oplog::{loro_dag::FrontiersNotIncluded, OpLog},
+    oplog::{loro_dag::FrontiersNotIncluded, ImportedValues, OpLog},
     state::DocState,
     subscription::{LocalUpdateCallback, Observer, Subscriber},
     undo::DiffBatch,
@@ -660,6 +660,7 @@ impl LoroDoc {
                 } else {
                     self.import_changes_and_apply_delta_to_state_if_needed(
                         |oplog| encoding::decode_oplog_changes(oplog, parsed),
+                        ImportedValues::Exact,
                         origin,
                     )
 
@@ -671,6 +672,7 @@ impl LoroDoc {
             }
             EncodeMode::FastUpdates => self.import_changes_and_apply_delta_to_state_if_needed(
                 |oplog| encoding::decode_oplog_changes(oplog, parsed),
+                ImportedValues::Exact,
                 origin,
             ),
             EncodeMode::Auto => {
@@ -764,11 +766,14 @@ impl LoroDoc {
     pub(crate) fn import_changes_and_apply_delta_to_state_if_needed(
         &self,
         decode_changes: impl FnOnce(&mut OpLog) -> Result<Vec<Change>, LoroError>,
+        values: ImportedValues,
         origin: InternalString,
     ) -> Result<ImportStatus, LoroError> {
         let mut oplog = self.oplog.lock();
         let arena_checkpoint = oplog.arena.checkpoint_for_rollback();
-        let changes = match decode_changes(&mut oplog) {
+        let changes = match decode_changes(&mut oplog)
+            .and_then(|changes| oplog.check_and_trim_known_part_of_changes(changes, values))
+        {
             Ok(changes) => changes,
             Err(e) => {
                 oplog.rollback_arena(arena_checkpoint, &self.state.lock());
@@ -950,6 +955,7 @@ impl LoroDoc {
         self.with_barrier(|| {
             let result = self.import_changes_and_apply_delta_to_state_if_needed(
                 |oplog| crate::encoding::json_schema::decode_json_changes(json, &oplog.arena),
+                ImportedValues::Lossy,
                 Default::default(),
             );
             self.emit_events();

crates/loro-internal/src/oplog.rs

@@ -1,4 +1,6 @@
 mod change_store;
+mod known_history;
+pub(crate) use known_history::ImportedValues;
 pub(crate) mod loro_dag;
 mod pending_changes;