diff --git a/.changeset/import-reused-peer-id.md b/.changeset/import-reused-peer-id.md new file mode 100644 index 000000000..8ced6f676 --- /dev/null +++ b/.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. diff --git a/AGENTS.md b/AGENTS.md index 708a805f9..0d59af1de 100644 --- a/AGENTS.md +++ b/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). diff --git a/context/import-peer-id-reuse.md b/context/import-peer-id-reuse.md new file mode 100644 index 000000000..638f8f612 --- /dev/null +++ b/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`. diff --git a/crates/examples/examples/import_known_history_perf.rs b/crates/examples/examples/import_known_history_perf.rs new file mode 100644 index 000000000..6677ab2f4 --- /dev/null +++ b/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(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>, change: loro::JsonChange) -> loro::JsonSchema { + loro::JsonSchema { + schema_version: 1, + start_version: Default::default(), + peers: peers.clone(), + changes: vec![change], + } +} diff --git a/crates/loro-internal/AGENTS.md b/crates/loro-internal/AGENTS.md index a54380f90..e68036673 100644 --- a/crates/loro-internal/AGENTS.md +++ b/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). diff --git a/crates/loro-internal/src/arena.rs b/crates/loro-internal/src/arena.rs index 03c92fc4d..4faa19afa 100644 --- a/crates/loro-internal/src/arena.rs +++ b/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, b: Range) -> 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] diff --git a/crates/loro-internal/src/encoding/fast_snapshot.rs b/crates/loro-internal/src/encoding/fast_snapshot.rs index 727770ab6..cf0542160 100644 --- a/crates/loro-internal/src/encoding/fast_snapshot.rs +++ b/crates/loro-internal/src/encoding/fast_snapshot.rs @@ -383,7 +383,6 @@ pub(crate) fn encode_updates_in_range( pub(crate) fn decode_updates(oplog: &mut OpLog, body: Bytes) -> Result, 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 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, LoroError>, + values: ImportedValues, origin: InternalString, ) -> Result { 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(); diff --git a/crates/loro-internal/src/oplog.rs b/crates/loro-internal/src/oplog.rs index d07beaa21..663f8cd51 100644 --- a/crates/loro-internal/src/oplog.rs +++ b/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; diff --git a/crates/loro-internal/src/oplog/change_store.rs b/crates/loro-internal/src/oplog/change_store.rs index 7edd054a7..0b023fc1e 100644 --- a/crates/loro-internal/src/oplog/change_store.rs +++ b/crates/loro-internal/src/oplog/change_store.rs @@ -358,6 +358,12 @@ impl ChangeStore { self.external_kv.lock().export_all() } + /// Decode the changes of a snapshot imported into a non-empty doc. + /// + /// Changes this doc already has are kept, because + /// `OpLog::check_and_trim_known_part_of_changes` compares them with the local + /// history before trimming them. When nothing is new, return no changes and + /// skip the clones. pub(crate) fn decode_snapshot_for_updates( bytes: Bytes, arena: &SharedArena, @@ -365,46 +371,22 @@ impl ChangeStore { ) -> Result, LoroError> { let change_store = ChangeStore::new_mem(arena, Arc::new(AtomicI64::new(0))); let _ = change_store.import_all(bytes)?; - let mut changes = Vec::new(); + let mut has_new = false; change_store.visit_all_changes(&mut |c| { - let cnt_threshold = self_vv.get(&c.id.peer).copied().unwrap_or(0); - if c.id.counter >= cnt_threshold { - changes.push(c.clone()); - return; - } - - let change_end = c.ctr_end(); - if change_end > cnt_threshold { - changes.push(c.slice((cnt_threshold - c.id.counter) as usize, c.atom_len())); - } + has_new |= c.ctr_end() > self_vv.get(&c.id.peer).copied().unwrap_or(0); }); + let mut changes = Vec::new(); + if has_new { + change_store.visit_all_changes(&mut |c| changes.push(c.clone())); + } Ok(changes) } - pub(crate) fn decode_block_bytes( - bytes: Bytes, - arena: &SharedArena, - self_vv: &VersionVector, - ) -> LoroResult> { - let mut ans = ChangesBlockBytes::new(bytes).parse(arena)?; - if ans.is_empty() { - return Ok(ans); - } - - let start = self_vv.get(&ans[0].peer()).copied().unwrap_or(0); - ans.retain_mut(|c| { - if c.id.counter >= start { - true - } else if c.ctr_end() > start { - *c = c.slice((start - c.id.counter) as usize, c.atom_len()); - true - } else { - false - } - }); - - Ok(ans) + /// Decode an update block. Changes the doc already has are kept; see + /// [`Self::decode_snapshot_for_updates`]. + pub(crate) fn decode_block_bytes(bytes: Bytes, arena: &SharedArena) -> LoroResult> { + ChangesBlockBytes::new(bytes).parse(arena) } /// Rolls back the store and the arena (to `arena`, the checkpoint taken when the import diff --git a/crates/loro-internal/src/oplog/known_history.rs b/crates/loro-internal/src/oplog/known_history.rs new file mode 100644 index 000000000..a5ad2fd09 --- /dev/null +++ b/crates/loro-internal/src/oplog/known_history.rs @@ -0,0 +1,384 @@ +//! Imported changes that reuse op ids the doc already has. +//! +//! 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, the "known" part is different history, and the rest is then applied to +//! a state it was not written against: it panicked inside state apply (poisoning +//! the doc) or silently scrambled it (loro-dev/loro#1118). See +//! `context/import-peer-id-reuse.md`. + +use loro_common::{Counter, HasCounterSpan, LoroError, LoroResult, PeerID, ID}; +use rle::{HasLength, Sliceable}; +use rustc_hash::FxHashMap; + +use crate::{ + arena::SharedArena, + change::Change, + container::list::list_op::InnerListOp, + op::{InnerContent, Op}, + version::Frontiers, + OpLog, +}; + +#[cfg(feature = "counter")] +use crate::op::FutureInnerContent; + +/// How faithfully the import format carries op values. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ImportedValues { + /// Binary encodings round-trip every value. + Exact, + /// JSON updates may have gone through JSON text, which does not round-trip + /// every value (`NaN` becomes `null`, binary may come back as a list). Values + /// are then not compared; everything else still is. + Lossy, +} + +impl OpLog { + /// Check the decoded `changes` against the history this doc already has for the + /// same op ids, then drop the parts it already has. + /// + /// Returns [`LoroError::UsedOpID`] with the first conflicting id when an imported + /// change differs from the local one (deps or op content), and + /// [`LoroError::DecodeError`] when a change skips counters of its peer (see + /// [`Self::check_changes_do_not_skip_counters`]). It reads the op log only, so a + /// caller can return the error before mutating anything. + /// + /// The comparison runs only when the import brings something new: that is what + /// would be applied on top of a conflicting prefix. An import made only of + /// known changes is a no-op either way and stays as cheap as before. + pub(crate) fn check_and_trim_known_part_of_changes( + &self, + mut changes: Vec, + values: ImportedValues, + ) -> LoroResult> { + let vv = self.vv(); + let known_end = |c: &Change| vv.get(&c.id.peer).copied().unwrap_or(0); + if changes.iter().any(|c| c.ctr_end() > known_end(c)) { + for change in changes.iter() { + if change.id.counter < known_end(change) { + self.check_known_part_of_change(change, known_end(change), values)?; + } + } + } + + changes.retain_mut(|c| { + let end = known_end(c); + if c.id.counter >= end { + true + } else if c.ctr_end() > end { + *c = c.slice((end - c.id.counter) as usize, c.atom_len()); + true + } else { + false + } + }); + self.check_changes_do_not_skip_counters(&changes)?; + Ok(changes) + } + + /// A change of peer `p` starting at counter `c > 0` is written after `p@c-1`, so + /// `p@c-1` is in the causal past of its deps. When all its deps are present but + /// `p@c-1` is not, the change skips counters and is invalid. The main pass of + /// `import_changes_to_oplog` applies any change whose deps are present and + /// would hit the counter assertion in `AppDag::update_version_on_new_change` + /// under the op log lock. (Changes parked as pending already wait for `p@c-1`, + /// see `remote_change_apply_state`.) + /// + /// This replays that pass on counter ends only: changes are visited in the same + /// order, and a change counts as applied when its deps are. + fn check_changes_do_not_skip_counters(&self, changes: &[Change]) -> LoroResult<()> { + let vv = self.vv(); + let mut ends: FxHashMap = FxHashMap::default(); + for change in changes { + let end_of = |ends: &FxHashMap, peer: &PeerID| { + ends.get(peer) + .or_else(|| vv.get(peer)) + .copied() + .unwrap_or(0) + }; + let peer = change.id.peer; + let end = end_of(&ends, &peer); + if change.ctr_end() <= end || self.dag.import_deps_before_shallow_root(&change.deps) { + continue; + } + + if !change + .deps + .iter() + .all(|dep| dep.counter < end_of(&ends, &dep.peer)) + { + // Parked as pending. + continue; + } + + if change.id.counter > end { + return Err(LoroError::DecodeError( + format!( + "Invalid change {}: its deps are present but {} is not, so it skips counters of its peer", + change.id, + ID::new(peer, change.id.counter - 1), + ) + .into_boxed_str(), + )); + } + + ends.insert(peer, change.ctr_end()); + } + + Ok(()) + } + + fn check_known_part_of_change( + &self, + change: &Change, + known_end: Counter, + values: ImportedValues, + ) -> LoroResult<()> { + let peer = change.id.peer; + // History before the shallow root is not stored, so it cannot be compared. + let shallow_start = self.shallow_since_vv().get(&peer).copied().unwrap_or(0); + let end = change.ctr_end().min(known_end); + let mut ctr = change.id.counter.max(shallow_start); + while ctr < end { + // Missing local history cannot be compared; keep the old behavior for it. + let Some(local) = self.change_store.get_change(ID::new(peer, ctr)) else { + break; + }; + let seg_end = end.min(local.ctr_end()); + if seg_end <= ctr { + break; + } + + if deps_at(change, ctr) != deps_at(&local, ctr) { + return Err(LoroError::UsedOpID { + id: ID::new(peer, ctr), + }); + } + + if let Some(bad) = first_mismatch(&self.arena, values, change, &local, ctr, seg_end) { + return Err(LoroError::UsedOpID { + id: ID::new(peer, bad), + }); + } + + ctr = seg_end; + } + + Ok(()) + } +} + +/// The deps of the op at `ctr`. Inside a change every op depends on the previous +/// one, and changes only merge when that holds, so this does not depend on how +/// either side split its history into changes. +fn deps_at(change: &Change, ctr: Counter) -> Frontiers { + if ctr == change.id.counter { + change.deps.clone() + } else { + Frontiers::from_id(ID::new(change.id.peer, ctr - 1)) + } +} + +/// Walk the ops of `a` and `b` over `[start, end)` and return the first counter at +/// which they differ. Each side may split or merge the ops differently, so both +/// are cut at every boundary of either side before comparing. +fn first_mismatch( + arena: &SharedArena, + values: ImportedValues, + a: &Change, + b: &Change, + start: Counter, + end: Counter, +) -> Option { + let mut a_ops = OpCursor::new(a, start); + let mut b_ops = OpCursor::new(b, start); + let mut ctr = start; + while ctr < end { + let (Some(a_op), Some(b_op)) = (a_ops.op_at(ctr), b_ops.op_at(ctr)) else { + return Some(ctr); + }; + let len = (end - ctr) + .min(a_op.ctr_end() - ctr) + .min(b_op.ctr_end() - ctr); + let a_slice = slice_op(a_op, ctr, len); + let b_slice = slice_op(b_op, ctr, len); + if !op_eq(arena, values, &a_slice, &b_slice) { + // Error path only: find the exact atom. + let offset = (0..len) + .find(|&i| { + !op_eq( + arena, + values, + &slice_op(a_op, ctr + i, 1), + &slice_op(b_op, ctr + i, 1), + ) + }) + .unwrap_or(0); + return Some(ctr + offset); + } + ctr += len; + } + + None +} + +struct OpCursor<'a> { + ops: &'a [Op], + index: usize, +} + +impl<'a> OpCursor<'a> { + fn new(change: &'a Change, start: Counter) -> Self { + let ops = change.ops.vec().as_slice(); + let index = ops.partition_point(|op| op.ctr_end() <= start); + Self { ops, index } + } + + /// The op containing `ctr`. Counters only move forward. + fn op_at(&mut self, ctr: Counter) -> Option<&'a Op> { + while let Some(op) = self.ops.get(self.index) { + if op.ctr_end() > ctr { + return (op.counter <= ctr).then_some(op); + } + self.index += 1; + } + None + } +} + +fn slice_op(op: &Op, ctr: Counter, len: Counter) -> std::borrow::Cow<'_, Op> { + let from = (ctr - op.counter) as usize; + let to = from + len as usize; + if from == 0 && to == op.atom_len() { + std::borrow::Cow::Borrowed(op) + } else { + std::borrow::Cow::Owned(op.slice(from, to)) + } +} + +/// Whether two ops with the same id range are the same op. Compares what the op +/// means, not how it is stored: arena offsets and the direction of a one-atom +/// delete are representation details. +fn op_eq(arena: &SharedArena, values: ImportedValues, a: &Op, b: &Op) -> bool { + if a.container != b.container || a.atom_len() != b.atom_len() { + return false; + } + + let exact = values == ImportedValues::Exact; + + match (&a.content, &b.content) { + (InnerContent::List(a), InnerContent::List(b)) => list_op_eq(arena, exact, a, b), + (InnerContent::Map(a), InnerContent::Map(b)) => { + a.key == b.key + && a.value.is_some() == b.value.is_some() + && (!exact || a.value == b.value) + } + (InnerContent::Tree(a), InnerContent::Tree(b)) => a == b, + (InnerContent::Future(a), InnerContent::Future(b)) => match (a, b) { + #[cfg(feature = "counter")] + (FutureInnerContent::Counter(a), FutureInnerContent::Counter(b)) => { + !exact || a == b || (a.is_nan() && b.is_nan()) + } + ( + crate::op::FutureInnerContent::Unknown { + prop: a_prop, + value: a_value, + }, + crate::op::FutureInnerContent::Unknown { + prop: b_prop, + value: b_value, + }, + ) => a_prop == b_prop && (!exact || a_value == b_value), + #[allow(unreachable_patterns)] + _ => false, + }, + _ => false, + } +} + +fn list_op_eq(arena: &SharedArena, exact: bool, a: &InnerListOp, b: &InnerListOp) -> bool { + match (a, b) { + ( + InnerListOp::Insert { + slice: a_slice, + pos: a_pos, + }, + InnerListOp::Insert { + slice: b_slice, + pos: b_pos, + }, + ) => { + a_pos == b_pos + && (!exact || arena.value_slices_eq(a_slice.to_range(), b_slice.to_range())) + } + ( + InnerListOp::InsertText { + slice: a_slice, + unicode_len: a_len, + pos: a_pos, + .. + }, + InnerListOp::InsertText { + slice: b_slice, + unicode_len: b_len, + pos: b_pos, + .. + }, + ) => a_pos == b_pos && a_len == b_len && a_slice[..] == b_slice[..], + (InnerListOp::Delete(a), InnerListOp::Delete(b)) => { + // A one-atom delete has no direction: `signed_len` 1 and -1 delete the + // same element, and which one is stored depends on how it was sliced. + a.id_start == b.id_start + && a.span.pos == b.span.pos + && (a.span.signed_len == b.span.signed_len + || (a.span.signed_len.abs() == 1 && b.span.signed_len.abs() == 1)) + } + ( + InnerListOp::Move { + from: a_from, + elem_id: a_elem, + to: a_to, + }, + InnerListOp::Move { + from: b_from, + elem_id: b_elem, + to: b_to, + }, + ) => a_from == b_from && a_elem == b_elem && a_to == b_to, + ( + InnerListOp::Set { + elem_id: a_elem, + value: a_value, + }, + InnerListOp::Set { + elem_id: b_elem, + value: b_value, + }, + ) => a_elem == b_elem && (!exact || a_value == b_value), + ( + InnerListOp::StyleStart { + start: a_start, + end: a_end, + key: a_key, + value: a_value, + info: a_info, + }, + InnerListOp::StyleStart { + start: b_start, + end: b_end, + key: b_key, + value: b_value, + info: b_info, + }, + ) => { + a_start == b_start + && a_end == b_end + && a_key == b_key + && (!exact || a_value == b_value) + && a_info == b_info + } + (InnerListOp::StyleEnd, InnerListOp::StyleEnd) => true, + _ => false, + } +} diff --git a/crates/loro-wasm/tests/import_reused_peer_id.test.ts b/crates/loro-wasm/tests/import_reused_peer_id.test.ts new file mode 100644 index 000000000..aafee4a3e --- /dev/null +++ b/crates/loro-wasm/tests/import_reused_peer_id.test.ts @@ -0,0 +1,76 @@ +import { describe, expect, it } from "vitest"; +import { LoroDoc } from "../bundler/index"; + +// Two docs that shared a peer id wrote different ops under the same ids. Importing +// one into the other used to trap the WASM instance ("RuntimeError: unreachable") +// or scramble the text. It must throw and leave the doc unchanged and usable. +// See loro-dev/loro#1118 and context/import-peer-id-reuse.md. + +function history(): LoroDoc { + const doc = new LoroDoc(); + doc.setPeerId(7); + doc.getMap("m").set("k", 0); + doc.getText("t").insert(0, "a"); + doc.commit(); // 7@0, 7@1 + return doc; +} + +function conflicting(text: string): LoroDoc { + const doc = new LoroDoc(); + doc.setPeerId(7); + doc.getText("t").insert(0, text); + doc.commit(); + return doc; +} + +function expectRejected(doc: LoroDoc, bytes: Uint8Array) { + const value = doc.toJSON(); + const version = doc.version().toJSON(); + expect(() => doc.import(bytes)).toThrow(/has been used/); + expect(doc.toJSON()).toEqual(value); + expect(doc.version().toJSON()).toEqual(version); + + // Still usable: local edits and legitimate remote updates apply. + doc.getText("after").insert(0, "local"); + doc.commit(); + const remote = new LoroDoc(); + remote.setPeerId(1000); + remote.import(doc.export({ mode: "snapshot" })); + remote.getMap("remote").set("k", "v"); + remote.commit(); + doc.import(remote.export({ mode: "update", from: doc.version() })); + expect(doc.toJSON()).toEqual(remote.toJSON()); +} + +describe("importing history that reuses local op ids", () => { + it("throws instead of trapping (issue repro)", () => { + const other = conflicting("xyz"); // 7@0..=2 + expectRejected(history(), other.export({ mode: "update" })); + expectRejected(history(), other.export({ mode: "snapshot" })); + }); + + it("throws instead of scrambling the text", () => { + const doc = new LoroDoc(); + doc.setPeerId(7); + doc.getText("t").insert(0, "hello"); + doc.commit(); + expectRejected(doc, conflicting("hellO world").export({ mode: "update" })); + expect(doc.getText("t").toString()).toBe("hello"); + }); + + it("still accepts overlapping re-imports of the same history", () => { + const source = new LoroDoc(); + source.setPeerId(1); + const text = source.getText("t"); + text.insert(0, "hello world"); + source.commit(); + const target = new LoroDoc(); + target.import(source.export({ mode: "update" })); + text.delete(0, 6); + text.insert(0, "hi "); + source.commit(); + target.import(source.export({ mode: "update" })); + target.import(source.export({ mode: "snapshot" })); + expect(target.toJSON()).toEqual(source.toJSON()); + }); +}); diff --git a/crates/loro/tests/import_reused_peer_id.rs b/crates/loro/tests/import_reused_peer_id.rs new file mode 100644 index 000000000..796492a70 --- /dev/null +++ b/crates/loro/tests/import_reused_peer_id.rs @@ -0,0 +1,463 @@ +//! Importing history whose op ids are already used locally by *different* ops +//! (two clients sharing a peer id) must fail with an `Err` before anything is +//! applied, instead of panicking, poisoning the doc, or scrambling its state. +//! See loro-dev/loro#1118. + +use loro::{ + ExportMode, LoroDoc, LoroError, LoroResult, LoroValue, ToJson, TreeParentId, VersionVector, ID, +}; +use pretty_assertions::assert_eq; + +struct Before { + value: serde_json::Value, + vv: VersionVector, + frontiers: loro::Frontiers, +} + +fn snapshot_of(doc: &LoroDoc) -> Before { + Before { + value: doc.get_deep_value().to_json_value(), + vv: doc.oplog_vv(), + frontiers: doc.oplog_frontiers(), + } +} + +fn assert_unchanged(doc: &LoroDoc, before: &Before) { + assert_eq!(doc.get_deep_value().to_json_value(), before.value); + assert_eq!(doc.oplog_vv(), before.vv); + assert_eq!(doc.oplog_frontiers(), before.frontiers); + assert_eq!(doc.state_frontiers(), before.frontiers); +} + +fn assert_used_op_id(err: LoroError, expected: ID) { + match err { + LoroError::UsedOpID { id } => assert_eq!(id, expected), + other => panic!("expected UsedOpID({expected}), got {other:?}"), + } +} + +/// After a rejected import the doc keeps working: it accepts local edits and +/// legitimate remote updates, and it can be exported and dropped. +fn assert_still_usable(doc: &LoroDoc) { + doc.get_text("after").insert(0, "local").unwrap(); + doc.commit(); + + let remote = LoroDoc::new(); + remote.set_peer_id(1000).unwrap(); + remote + .import(&doc.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + remote.get_map("remote").insert("k", "v").unwrap(); + remote.commit(); + doc.import(&remote.export(ExportMode::updates(&doc.oplog_vv())).unwrap()) + .unwrap(); + assert_eq!( + doc.get_deep_value().to_json_value(), + remote.get_deep_value().to_json_value() + ); + + let copy = LoroDoc::new(); + copy.import(&doc.export(ExportMode::Snapshot).unwrap()) + .unwrap(); + assert_eq!( + copy.get_deep_value().to_json_value(), + doc.get_deep_value().to_json_value() + ); +} + +/// The reproduction from loro-dev/loro#1118: `history` and `other` both wrote +/// 7@0..=1, and `other` went on to 7@2. The import used to trim the known part of +/// `other`'s change and apply its tail (an insert at position 2) on top of +/// `history`'s text, which has length 1. +#[test] +fn import_overlapping_change_with_different_content_is_rejected() -> LoroResult<()> { + let history = LoroDoc::new(); + history.set_peer_id(7)?; + history.get_map("m").insert("k", 0)?; + history.get_text("t").insert(0, "a")?; + history.commit(); // 7@0, 7@1 + + let other = LoroDoc::new(); + other.set_peer_id(7)?; + other.get_text("t").insert(0, "xyz")?; + other.commit(); // 7@0..=2 + + let before = snapshot_of(&history); + for update in [ + other.export(ExportMode::all_updates())?, + other.export(ExportMode::Snapshot)?, + ] { + let err = history.import(&update).unwrap_err(); + assert_used_op_id(err, ID::new(7, 0)); + assert_unchanged(&history, &before); + + let err = history.import_batch(&[update]).unwrap_err(); + assert_used_op_id(err, ID::new(7, 0)); + assert_unchanged(&history, &before); + assert!(!history.is_detached()); + } + + let json = other.export_json_updates(&Default::default(), &other.oplog_vv()); + let err = history.import_json_updates(json).unwrap_err(); + assert_used_op_id(err, ID::new(7, 0)); + assert_unchanged(&history, &before); + + assert_still_usable(&history); + Ok(()) +} + +/// Same shape, but the shifted tail lands on a valid position, so the import used +/// to succeed and silently scramble the text. +#[test] +fn import_overlapping_change_with_same_shape_is_rejected() -> LoroResult<()> { + let history = LoroDoc::new(); + history.set_peer_id(7)?; + history.get_text("t").insert(0, "hello")?; + history.commit(); // 7@0..=4 + + let other = LoroDoc::new(); + other.set_peer_id(7)?; + other.get_text("t").insert(0, "hellO world")?; + other.commit(); // 7@0..=10 + + let before = snapshot_of(&history); + let err = history + .import(&other.export(ExportMode::all_updates())?) + .unwrap_err(); + assert_used_op_id(err, ID::new(7, 4)); + assert_unchanged(&history, &before); + assert_still_usable(&history); + Ok(()) +} + +/// The conflicting part is a separate change that is entirely known locally, and +/// the new part starts exactly at the local counter end, so there is no partially +/// overlapping change to trim. +#[test] +fn import_known_conflicting_change_followed_by_new_change_is_rejected() -> LoroResult<()> { + let history = LoroDoc::new(); + history.set_peer_id(7)?; + history.get_text("t").insert(0, "ab")?; + history.commit(); // 7@0..=1 + + let other = LoroDoc::new(); + other.set_peer_id(7)?; + other.get_text("t").insert(0, "xy")?; + other.commit(); // 7@0..=1 + // Put a change from another peer in between so 7@2 starts a new change. + let helper = LoroDoc::new(); + helper.set_peer_id(9)?; + helper.import(&other.export(ExportMode::all_updates())?)?; + helper.get_map("m").insert("k", 1)?; + helper.commit(); + other.import(&helper.export(ExportMode::all_updates())?)?; + other.get_text("t").insert(2, "z")?; + other.commit(); // 7@2, deps [7@1, 9@0] + + let before = snapshot_of(&history); + let err = history + .import(&other.export(ExportMode::all_updates())?) + .unwrap_err(); + assert_used_op_id(err, ID::new(7, 0)); + assert_unchanged(&history, &before); + assert_still_usable(&history); + Ok(()) +} + +/// A peer-id collision in history the import does not build on (nothing new is +/// imported) is not detected, and the import stays a no-op. +#[test] +fn import_with_only_known_conflicting_changes_is_a_no_op() -> LoroResult<()> { + let history = LoroDoc::new(); + history.set_peer_id(7)?; + history.get_text("t").insert(0, "ab")?; + history.commit(); + + let other = LoroDoc::new(); + other.set_peer_id(7)?; + other.get_text("t").insert(0, "xy")?; + other.commit(); + + let before = snapshot_of(&history); + let status = history.import(&other.export(ExportMode::all_updates())?)?; + assert!(status.success.is_empty()); + assert!(status.pending.is_none()); + assert_unchanged(&history, &before); + Ok(()) +} + +/// Builds a doc with every kind of op, syncing `replica` piecewise along the way so +/// its change store splits and merges the history differently from `source`'s. +fn build_source_and_piecewise_replica() -> LoroResult<(LoroDoc, LoroDoc)> { + let source = LoroDoc::new(); + source.set_peer_id(1)?; + let replica = LoroDoc::new(); + replica.set_peer_id(2)?; + let sync = |from: &LoroDoc, to: &LoroDoc| -> LoroResult<()> { + to.import(&from.export(ExportMode::updates(&to.oplog_vv()))?)?; + Ok(()) + }; + + let text = source.get_text("text"); + text.insert(0, "hello wörld 😀")?; + source.commit(); + sync(&source, &replica)?; + text.delete(3, 4)?; // forward + for i in (0..3).rev() { + text.delete(i, 1)?; // backward, single-atom deletes + } + text.mark(0..2, "bold", true)?; + source.commit(); + sync(&source, &replica)?; + + let list = source.get_list("list"); + list.push(1)?; + list.push("two")?; + list.push(LoroValue::Double(f64::NAN))?; + list.delete(0, 1)?; + let map = source.get_map("map"); + map.insert("nan", f64::NAN)?; + map.insert("bin", vec![1u8, 2, 3])?; + map.delete("bin")?; + source.commit(); + sync(&source, &replica)?; + + let mov = source.get_movable_list("mov"); + mov.push("a")?; + mov.push("b")?; + mov.push("c")?; + mov.mov(0, 2)?; + mov.set(1, "B")?; + mov.delete(0, 1)?; + let tree = source.get_tree("tree"); + let root = tree.create(TreeParentId::Root)?; + let child = tree.create(root)?; + tree.mov(child, TreeParentId::Root)?; + tree.delete(root)?; + let counter = source.get_counter("counter"); + counter.increment(1.5)?; + counter.decrement(0.5)?; + source.commit(); + + // Changes from the replica's own peer in between, then more from `source`. + sync(&source, &replica)?; + replica.get_text("text").insert(0, ">")?; + replica.commit(); + sync(&replica, &source)?; + text.insert(1, "more")?; + text.delete(0, 2)?; + source.commit(); + Ok((source, replica)) +} + +/// Re-importing history the doc already has must still succeed, whatever the +/// export format and however the change store split or merged it. +#[test] +fn reimporting_known_history_with_new_changes_still_succeeds() -> LoroResult<()> { + let (source, replica) = build_source_and_piecewise_replica()?; + let expected = source.get_deep_value().to_json_value(); + let replica_vv = replica.oplog_vv(); + assert_ne!(replica_vv, source.oplog_vv()); + + let full_updates = source.export(ExportMode::all_updates())?; + let snapshot = source.export(ExportMode::Snapshot)?; + let json = source.export_json_updates(&Default::default(), &source.oplog_vv()); + let json_updates = serde_json::to_string(&json).unwrap(); + + for (name, import) in [ + ( + "updates", + Box::new(|doc: &LoroDoc| doc.import(&full_updates).map(|_| ())) + as Box LoroResult<()>>, + ), + ( + "snapshot", + Box::new(|doc: &LoroDoc| doc.import(&snapshot).map(|_| ())), + ), + ( + "batch", + Box::new(|doc: &LoroDoc| { + doc.import_batch(&[full_updates.clone(), snapshot.clone()]) + .map(|_| ()) + }), + ), + ( + "json", + Box::new(|doc: &LoroDoc| doc.import_json_updates(json_updates.as_str()).map(|_| ())), + ), + ] { + let doc = LoroDoc::new(); + doc.import(&replica.export(ExportMode::Snapshot)?)?; + import(&doc).unwrap_or_else(|e| panic!("{name}: {e:?}")); + assert_eq!(doc.get_deep_value().to_json_value(), expected, "{name}"); + + // The replica itself, whose change store holds the history split up. + let doc = LoroDoc::new(); + doc.import(&replica.export(ExportMode::all_updates())?)?; + doc.import(&replica.export(ExportMode::updates(&doc.oplog_vv()))?)?; + import(&doc).unwrap_or_else(|e| panic!("{name} (updates base): {e:?}")); + assert_eq!(doc.get_deep_value().to_json_value(), expected, "{name}"); + } + + // And in the other direction: the merged history into the piecewise one. + replica.import(&full_updates)?; + assert_eq!(replica.get_deep_value().to_json_value(), expected); + Ok(()) +} + +/// Two exports of the same peer's history that were cut at different points, +/// imported so that every imported change straddles the local counter end. +#[test] +fn reimporting_overlapping_ranges_of_the_same_history_succeeds() -> LoroResult<()> { + let source = LoroDoc::new(); + source.set_peer_id(1)?; + let text = source.get_text("text"); + let mut versions = vec![source.oplog_vv()]; + for i in 0..20 { + text.insert(text.len_unicode(), &format!("{i},"))?; + if i % 3 == 0 { + text.delete(0, 1)?; + } + source.commit(); + versions.push(source.oplog_vv()); + } + + let target = LoroDoc::new(); + for window in versions.windows(3) { + // Export [v_i, v_{i+2}) while the target already has up to v_{i+1}. + let update = source.export(ExportMode::updates(&window[0]))?; + target.import(&update)?; + } + assert_eq!( + target.get_deep_value().to_json_value(), + source.get_deep_value().to_json_value() + ); + Ok(()) +} + +/// The issue also noted that dropping the doc aborted the process when the +/// failed import panicked inside `catch_unwind`. With an `Err` there is no panic; +/// make sure dropping inside the closure is fine. +#[test] +fn rejected_import_inside_catch_unwind_can_drop_the_doc() { + let result = std::panic::catch_unwind(|| { + let history = LoroDoc::new(); + history.set_peer_id(7).unwrap(); + history.get_map("m").insert("k", 0).unwrap(); + history.get_text("t").insert(0, "a").unwrap(); + history.commit(); + + let other = LoroDoc::new(); + other.set_peer_id(7).unwrap(); + other.get_text("t").insert(0, "xyz").unwrap(); + other.commit(); + + let r = history.import(&other.export(ExportMode::all_updates()).unwrap()); + drop(history); + r.is_err() + }); + assert_eq!(result.ok(), Some(true)); +} + +/// JSON updates for `doc`'s last change, moved to `peer` at `counter` with `deps`. +fn forged_json_change(doc: &LoroDoc, peer: u64, counter: i32, deps: Vec) -> loro::JsonSchema { + let mut json = doc.export_json_updates(&Default::default(), &doc.oplog_vv()); + json.changes.drain(..json.changes.len() - 1); + let change = json.changes.last_mut().unwrap(); + let shift = counter - change.id.counter; + change.id.counter = counter; + change.deps = deps; + for op in change.ops.iter_mut() { + op.counter += shift; + } + json.peers = Some(vec![peer]); + change.id.peer = 0; + json +} + +/// A change whose deps are all present but whose own peer's previous counter is +/// not skips counters. It used to trip a DAG assertion under the op log lock, +/// poisoning the doc (and aborting when it was dropped). +#[test] +fn import_change_that_skips_counters_is_rejected() -> LoroResult<()> { + let writer = LoroDoc::new(); + writer.set_peer_id(8)?; + writer.get_text("t").insert(0, "xyz")?; + writer.commit(); + + // Into a doc that has none of peer 7, and into one that has 7@0..=1. + for local_len in [0, 2] { + let doc = LoroDoc::new(); + doc.set_peer_id(7)?; + if local_len > 0 { + doc.get_text("t").insert(0, &"a".repeat(local_len))?; + doc.commit(); + } + let before = snapshot_of(&doc); + let err = doc + .import_json_updates(forged_json_change(&writer, 7, 5, vec![])) + .unwrap_err(); + assert!( + matches!(&err, LoroError::DecodeError(msg) if msg.contains("skips counters")), + "{err:?}" + ); + assert_unchanged(&doc, &before); + doc.set_peer_id(100)?; + assert_still_usable(&doc); + } + + // The deps are satisfied by an earlier change of the same import. + let helper = LoroDoc::new(); + helper.set_peer_id(9)?; + helper.get_map("m").insert("k", 1)?; + helper.commit(); + let mut json = helper.export_json_updates(&Default::default(), &helper.oplog_vv()); + let forged = forged_json_change(&writer, 7, 5, vec![ID::new(1, 0)]); + // Peer index 1 is peer 7, index 0 is peer 9. + json.peers = Some(vec![9, 7]); + let mut change = forged.changes.into_iter().next().unwrap(); + change.id.peer = 1; + change.deps = vec![ID::new(0, 0)]; + json.changes.push(change); + + let doc = LoroDoc::new(); + let before = snapshot_of(&doc); + let err = doc.import_json_updates(json).unwrap_err(); + assert!(matches!(err, LoroError::DecodeError(_)), "{err:?}"); + assert_unchanged(&doc, &before); + assert_still_usable(&doc); + Ok(()) +} + +/// Known history below a shallow root cannot be compared and must not be +/// rejected for that. +#[test] +fn reimporting_into_a_shallow_doc_still_succeeds() -> LoroResult<()> { + let source = LoroDoc::new(); + source.set_peer_id(1)?; + let text = source.get_text("text"); + for i in 0..10 { + text.insert(0, &format!("{i}"))?; + source.commit(); + } + let shallow = LoroDoc::new(); + shallow.import(&source.export(ExportMode::shallow_snapshot(&source.oplog_frontiers()))?)?; + text.insert(0, "new")?; + source.commit(); + + shallow.import(&source.export(ExportMode::updates(&shallow.shallow_since_vv().to_vv()))?)?; + assert_eq!( + shallow.get_deep_value().to_json_value(), + source.get_deep_value().to_json_value() + ); + + // The full history, which straddles the shallow root, plus one more change. + text.insert(0, "more")?; + source.commit(); + shallow.import(&source.export(ExportMode::all_updates())?)?; + assert_eq!( + shallow.get_deep_value().to_json_value(), + source.get_deep_value().to_json_value() + ); + Ok(()) +} diff --git a/crates/loro/tests/movable_list_invalid_ops.rs b/crates/loro/tests/movable_list_invalid_ops.rs index 1e7e6515d..991082d62 100644 --- a/crates/loro/tests/movable_list_invalid_ops.rs +++ b/crates/loro/tests/movable_list_invalid_ops.rs @@ -426,8 +426,10 @@ fn move_of_element_deleted_before_shallow_root_is_rejected() { /// (and no validation) was opened. With `two_peers`, `C1` is by peer 2 and `C2` by /// peer 3. fn forged_after_map_change(container: &str, forged_op: Value, two_peers: bool) -> String { + // Peer 2 is taken by `base_with_concurrent_edit`; reusing its `2@0` for other + // content is rejected as `UsedOpID` before the movable-list check runs. let (peers, c1_id, c2_id, c2_deps, c2_counter) = if two_peers { - (json!(["1", "2", "3"]), "0@1", "0@2", "0@1", 0) + (json!(["1", "4", "3"]), "0@1", "0@2", "0@1", 0) } else { (json!(["1"]), "4@0", "5@0", "4@0", 5) }; diff --git a/docs/encoding.md b/docs/encoding.md index 5d52cc82c..138b013df 100644 --- a/docs/encoding.md +++ b/docs/encoding.md @@ -543,11 +543,13 @@ The updates APIs differ only in which changes are selected: shallow root), while `UpdatesInRange` emits the normalized requested ID spans. Both produce the exact body above. -On import, each decoded block is filtered against the receiver's current +On import, the decoded changes are filtered against the receiver's current VersionVector. A change entirely before the receiver's exclusive counter end is -dropped; a change crossing that boundary is sliced at the boundary. The -remaining changes from every block are then sorted by starting Lamport value -before application. Block order in the byte stream is therefore not application +dropped; a change crossing that boundary is sliced at the boundary. Before +that, when the import brings anything new, the part of each change the receiver +already has is compared with its local history, and a mismatch (a reused peer +id) fails the import with `UsedOpID`. The changes from every block are sorted by +starting Lamport value before application. Block order in the byte stream is therefore not application order. These selection and ordering rules are API behavior around the wire format; they do not add fields to it. @@ -555,8 +557,9 @@ Selection: [`ChangeStore::export_blocks_from`](../crates/loro-internal/src/oplog/change_store.rs#L543-L577), [`export_blocks_in_range`](../crates/loro-internal/src/oplog/change_store.rs#L203-L233). Import filtering and ordering: -[`ChangeStore::decode_block_bytes`](../crates/loro-internal/src/oplog/change_store.rs#L278-L300), -[`fast_snapshot.rs::decode_updates`](../crates/loro-internal/src/encoding/fast_snapshot.rs#L360-L388). +[`fast_snapshot.rs::decode_updates`](../crates/loro-internal/src/encoding/fast_snapshot.rs), +[`OpLog::check_and_trim_known_part_of_changes`](../crates/loro-internal/src/oplog/known_history.rs) +(see [`context/import-peer-id-reuse.md`](../context/import-peer-id-reuse.md)). ## 7. Postcard change block