diff --git a/.changeset/movable-list-invalid-move-set.md b/.changeset/movable-list-invalid-move-set.md new file mode 100644 index 000000000..b12c457ee --- /dev/null +++ b/.changeset/movable-list-invalid-move-set.md @@ -0,0 +1,17 @@ +--- +"loro-crdt": patch +--- + +Importing a malformed movable-list `move` or `set` op no longer aborts the process. These ops used to panic inside the document locks, poisoning them, and the process then aborted during unwind. Now: + +- A `move`/`set` whose element id is not an element of the same list in the op's causal history, or a `move` whose index is past the end of the list, makes `import`, `importJsonUpdates` and `importBatch` return an error. Nothing from that import is applied and the document stays usable. This also holds when the op depends on another change in the same import, which used to skip validation. +- Insert, delete and move positions of 1,073,741,822 (about 2^30) or more in a List, MovableList or Text op are rejected when the op is decoded, instead of panicking during import. This is a hard limit on sequence length: it also applies when a document reads its own stored history or a snapshot. No peer could import such a document before either. +- A `move`/`set` that targets an element deleted earlier in its own history is applied the same way a concurrent move/set of a deleted element is: the move brings the element back, and the set is not visible. Every import path, and a replay of the full history, gives the same result. + +Also fixed: + +- `getChangeAtLamport` returned `undefined` or the wrong change for changes that were loaded from a snapshot and not yet read. The change store misread the lamport range of stored blocks, a leftover from the block format change in 1.0. On a snapshot-loaded peer with many blocks, 42 of 669 sampled lamports returned `undefined` and 625 returned the wrong change; all 669 are now correct. +- A rolled-back import could leave a cached change block behind that made the next import of the same changes panic with "counter should be continuous". +- Imports of List and Tree ops no longer copy the peer's last change block on every import to keep a rollback copy. Many small List or Tree imports are about 30% faster. + +Performance: movable-list imports now check every `move`/`set` against the document's history. In release benchmarks that makes them about 2–8% slower for many small imports and 0–4% slower for a 120k-op import where about 70% of the ops are moves and sets. diff --git a/AGENTS.md b/AGENTS.md index 248622b2a..658b6c21e 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). +- 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). - Shallow-snapshot dead-style redaction (null-only, both-expand exclusion, root-whitelisted latest state): [context/shallow-snapshot-style-redaction.md](context/shallow-snapshot-style-redaction.md). diff --git a/context/import-batch-atomicity.md b/context/import-batch-atomicity.md index cd20fe882..784478775 100644 --- a/context/import-batch-atomicity.md +++ b/context/import-batch-atomicity.md @@ -32,7 +32,9 @@ is still at its pre-batch version and the `OpLog` is the only thing that moved. the `OpLog` therefore makes the two agree again, which is what lets the doc stay attached. `import_batch` opens `OpLog::begin_import_rollback` before the loop and `BatchImportGuard::finish` either commits it or rolls the whole batch back and returns -the state-apply error. +the state-apply error. Before that checkout, `finish` also validates the movable-list element references +of every blob in the batch ([movable-list-op-validation.md](movable-list-op-validation.md)). +A failure there rolls the batch back the same way. Consequences to keep in mind: diff --git a/context/movable-list-op-validation.md b/context/movable-list-op-validation.md new file mode 100644 index 000000000..84cb775b7 --- /dev/null +++ b/context/movable-list-op-validation.md @@ -0,0 +1,169 @@ +# Movable List `Move`/`Set` Validation on Import + +Verified against code 2026-09-28 (merged with main `6e294c87`). + +Imported movable-list `Move { from, to, elem_id }` and `Set { elem_id, value }` ops +come from other peers, so they are external input. Several shapes of them used +to panic while the doc held its locks (`pos.unwrap()`/`value_id.unwrap()` in +`MovableListState::apply_diff_and_convert`, `idlp_to_id(..).unwrap()` in +`MovableListHistoryCache::last_pos`, `last_value(..).unwrap()` in +`MovableListDiffCalculator::calculate_diff`, `convert_index(..).unwrap()` for +an overrunning list delta, and the diff tracker's B-tree for huge positions). +The panic poisoned a `LoroMutex`, and the process aborted when a destructor hit +the poisoned lock during unwind. + +Tests: +- `crates/loro/tests/movable_list_invalid_ops.rs`. +- The binary and `import_batch` cases in `crates/loro-internal/src/tests/import_atomicity.rs`. +- The `ChangeStore` rollback and KV lookup tests in `src/oplog/change_store.rs`. +- `crates/loro-wasm/tests/movable_list_invalid_ops.test.ts`. + +## Rejected with `Err`: ops with no meaning + +1. **Positions no real sequence can reach.** `InnerListOp::check_positions` + (`src/container/list/list_op.rs`) runs when binary (`outdated_encode_reordered::decode_op`) + and JSON (`json_schema::decode_op`) ops are decoded. + - It rejects any position of a List/MovableList/Text op at or past + `UNKNOWN_SPAN_LEN - 1` (`src/container/richtext/tracker.rs`). That is the length + of the tracker's placeholder span for unreplayed history. + - Positions past it panicked inside the tracker before `validate_diff` could run. + - This check needs no history, so it covers every import path, including + detached imports. + - **It is a hard document limit, not only an import check.** `decode_op` is also + how a doc parses its own stored change blocks and snapshot blocks + (`block_encode.rs`), so a sequence position ≥ 1,073,741,822 is rejected there + too. That is deliberate: + - Snapshot and update blocks are external input as well. + - A sequence that long (≈1.07e9 items, ≥1 GB of text) cannot be diffed or + imported by any peer anyway: the tracker panics before this check existed. + - Skipping the check for "own" blocks would need a trust flag threaded through + block decoding. +2. **Unknown or out-of-history element.** `OpLog::validate_movable_list_elem_refs_in_import_scope` + (`src/oplog.rs`) checks each `Move`/`Set`. `elem_id` must be an `Insert` op in the + same container that lies in the op's causal history (see the causal check below). + It returns `LoroError::DecodeError`. + - The references are recorded by `OpLog::insert_new_change` into the open + `ImportRollback`, so they cover: + - directly imported changes, + - pending changes the import unlocks, + - every blob of an `import_batch`. + - The validator reads the recorded list, so it never re-reads the imported range + from the change store. + - `OpLog::resolve_movable_list_elem` finds the element with one + `ChangeStore::get_change_by_lamport_lte` lookup. That lookup scans both parsed + and KV-only blocks, so a miss means the lamport is not in the stored history. + - What an element resolves to (insert op, container) does not depend on the op, so + each pass caches it per `elem_id`. Moves/sets keep hitting the same elements. + - The causal check reads the cached start version of the op's DAG node + (`AppDag::ensure_vv_for`); earlier ops of the same node are by the same peer. + `AppDag::get_vv` would clone and insert into a version vector per op. + - Only on a shallow doc does a miss fall back to the shallow-root state + (`ContainerHistoryCache::shallow_root_has_movable_list_elem`): an op after the + root can only see pre-root elements that are still alive at the root. + - Call sites: + - the attached branch of `import_changes_and_apply_delta_to_state_if_needed` + (when `rollback_enabled`); + - its detached branch (it opens its own scope when the preflight asks for one + and no batch scope is open); + - `BatchImportGuard::finish`, which validates the whole batch once before the + closing checkout and rolls the whole batch back on failure + ([import-batch-atomicity.md](import-batch-atomicity.md)). +3. **Out-of-bounds `from`/`to` that survive the import.** `MovableListState::validate_diff` + bounds-checks the list delta in op-index space (dead list items count), the same + way `ListState::validate_diff` does. + - This check needs state, so a detached import cannot run it (see the gaps below). + - It checks the composed delta, so an out-of-bounds move that a later op in the + same import cancels (e.g. `move to: 9` followed by `delete pos: 9`) is accepted. + Every path gives the same result for it. + +### Why every such import gets a rollback scope + +`ImportChangesPreflight` (`OpLog::preflight_import_changes`) and +`PendingChanges::has_state_apply_rollback_ops` set `needs_state_apply_rollback` +for List, MovableList and Tree ops. +- The preflight inspects the ops of **every** new change, including ones whose + deps are not in the DAG yet. Those deps may be earlier changes of the same import, + which then unlock them during the import. +- It used to skip such changes before looking at their ops. That let + `[C1: map-only change, C2: forged op depending on C1]` skip both the rollback + scope and the validation. On that path a cross-container move was accepted + silently and put one element in two lists. + +## Applied with CRDT semantics: move/set of a deleted element + +A `Move`/`Set` that is causally after the delete of its element cannot come from +the public API. It is still **accepted**, with the same meaning as a *concurrent* +move/set of a deleted element. The move creates a new position, so the element +comes back with its last value. The set changes a value nobody can see. Why: + +- It gives a forger nothing new. The forger can declare deps from before the + delete and get the same effect from a legitimate concurrent move. +- Rejecting it consistently is not affordable. Checkout-mode replay (`checkout`, + `import_batch`'s closing reattach, concurrent imports, detached → `attach`) would + have to know whether the element is alive at the op's version. That needs an + index from list items to the deletes that cover them. If only the linear path + rejected it, `import` and `import_batch` of the same bytes would disagree. + +The one path that could not apply it is the forward fast path +(`DiffMode::Linear`/`ImportGreaterUpdates`). There the movable-list diff only +carries the fields the op touched, and `apply_diff_and_convert` fills in the rest +from the element in `DocState`. Detection and fallback: +- `MovableListState::references_absent_elem` / `DocState::needs_checkout_diff` + detect a delta that moves/sets an element missing from the state. +- `recalc_in_checkout_mode_if_needed` (`src/loro.rs`) recomputes that import's + diff with `DiffCalculator::new(true)` (Persist, so always Checkout mode). This + matches a full replay. +- Honest imports never trigger it. +- `validate_diff` still returns `Err` for such a delta as a backstop. + +## Change-store pitfalls this work exposed + +- **`decode_block_range` read a version varint that blocks do not have.** + `encode_block` writes a postcard `EncodedBlock` with no version prefix, so every + field was read one position off. + - History: `3d2d9d9c` (2024-09, "refactor: optimize block encoder") removed the + `version` field from the block, its encoder and the full decoder, but not from + `decode_block_range`. + - Every `loro-crdt@1.0.0*` release contains that commit, so no 1.x build ever wrote + version-prefixed blocks and the fix cannot make stored data unreadable. + + Consequences: + - Blocks that do not start at counter 0 were skipped. + - Block `0@P` got its lamport length as its lamport start. + - Lamport lookups on KV-only blocks were wrong: + `get_change_with_lamport_lte` / JS `getChangeAtLamport` after a snapshot load + (this was also broken on main), and the element validator above. + - Existing tests missed it because they parsed every block first. +- **Rollback must not leave an old block in front of the next insert.** + `insert_change_inner` merges a change into the cached block right before it. + - Failure sequence: an import creates a peer's newest block without loading the + older ones; a read during the scope (such as the validator's lamport lookup) + caches an older KV block; rollback removes the newest block. The next insert of + the same change then panicked with "counter should be continuous". + - `ChangeStore::rollback_import` therefore evicts the flushed blocks of every peer + the rollback touched. They reload from KV on demand. +- **Cheap rollback records.** `ChangeStoreRollback` keeps a `BlockShape` for each + unflushed pre-scope block an import appended to: change count, the last change's + op count and its last op. Rollback truncates back to it. + - Imports only append (push changes, or push ops that may merge into the last op), + so the shape is enough. + - Flushed blocks need no record: their KV copy is the pre-scope version. + - The earlier record was an `Arc` of the whole block, which made the next append + copy the block's changes on every import under a scope. That made small + movable-list imports about 40% slower, and List/Tree imports on main already + paid it (they are about 45% faster now). + +## Known gaps (not fixed here) + +- A `Move` whose `from` does not point at its element is accepted. The tracker + removes whatever list item sits at `from`, so another element can disappear. + The result is the same on every path, but it does not match any honest op. +- An explicitly detached doc that imports an op which only state validation + rejects (an out-of-bounds list insert, or a movable-list move out of bounds) + panics on `attach()`/`checkout_to_latest`. Those return `()` and `expect` the + checkout. This affects every container type and predates this change. +- The oplog inside a `FastSnapshot` is not validated op by op, because that would + decode every block. A forged snapshot can still reach the diff calculator's + unwraps on a later checkout. +- A forged change parked as pending fails the import that later unlocks it, so + that import is rolled back, the same as for list bounds errors. diff --git a/crates/loro-internal/AGENTS.md b/crates/loro-internal/AGENTS.md index 6498d1324..69ac67328 100644 --- a/crates/loro-internal/AGENTS.md +++ b/crates/loro-internal/AGENTS.md @@ -44,6 +44,9 @@ over graceful degradation. mergeable container regression tests. - `src/tests/import_atomicity.rs`: import rollback and malformed-input regressions. +- 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). - `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/container/list/list_op.rs b/crates/loro-internal/src/container/list/list_op.rs index 4bee2bf12..5e6d7875c 100644 --- a/crates/loro-internal/src/container/list/list_op.rs +++ b/crates/loro-internal/src/container/list/list_op.rs @@ -2,12 +2,13 @@ use std::ops::Range; use append_only_bytes::BytesSlice; use enum_as_inner::EnumAsInner; -use loro_common::{ContainerType, HasId, HasIdSpan, IdLp, LoroValue, ID}; +use loro_common::{ContainerType, HasId, HasIdSpan, IdLp, LoroError, LoroResult, LoroValue, ID}; use rle::{HasLength, Mergable, Sliceable}; use serde::{Deserialize, Serialize}; use crate::{ container::richtext::richtext_state::unicode_to_utf8_index, + container::richtext::tracker::UNKNOWN_SPAN_LEN, container::richtext::TextStyleInfoFlag, op::{ListSlice, SliceRange}, InternalString, @@ -88,6 +89,37 @@ impl ListOp<'_> { } impl InnerListOp { + /// Reject positions that no real sequence can reach. + /// + /// Diff calculation replays ops on a tracker whose unreplayed history is one + /// placeholder span of [`UNKNOWN_SPAN_LEN`] items, and positions past it panic + /// inside the tracker. Positions are op-index based (deleted items included), so + /// a valid document stays many orders of magnitude below this. Smaller + /// out-of-bounds positions are rejected later by state validation. + pub(crate) fn check_positions(&self) -> LoroResult<()> { + const MAX_POS: i64 = UNKNOWN_SPAN_LEN as i64 - 1; + let in_range = |pos: i64| (0..=MAX_POS).contains(&pos); + let ok = match self { + InnerListOp::Insert { pos, .. } => in_range(*pos as i64), + InnerListOp::InsertText { pos, .. } => in_range(*pos as i64), + InnerListOp::Delete(span) => { + in_range(span.start() as i64) && in_range(span.last() as i64) + } + InnerListOp::Move { from, to, .. } => in_range(*from as i64) && in_range(*to as i64), + InnerListOp::StyleStart { start, end, .. } => { + in_range(*start as i64) && in_range(*end as i64) + } + InnerListOp::Set { .. } | InnerListOp::StyleEnd => true, + }; + if ok { + Ok(()) + } else { + Err(LoroError::DecodeError( + format!("Sequence op position is out of range: {self:?}").into_boxed_str(), + )) + } + } + pub fn new_del(id: ID, pos: usize, len: isize) -> Self { assert!(len != 0); Self::Delete(DeleteSpanWithId { diff --git a/crates/loro-internal/src/container/richtext.rs b/crates/loro-internal/src/container/richtext.rs index ffe4aa339..20bbfb1b9 100644 --- a/crates/loro-internal/src/container/richtext.rs +++ b/crates/loro-internal/src/container/richtext.rs @@ -17,7 +17,7 @@ pub(crate) mod str_slice; mod style_range_map; #[cfg(all(test, miri))] mod tinyvec; -mod tracker; +pub(crate) mod tracker; use crate::{change::Lamport, delta::StyleMeta, utils::string_slice::StringSlice, InternalString}; use fugue_span::*; diff --git a/crates/loro-internal/src/container/richtext/tracker.rs b/crates/loro-internal/src/container/richtext/tracker.rs index 53e3fe89a..751b92c2c 100644 --- a/crates/loro-internal/src/container/richtext/tracker.rs +++ b/crates/loro-internal/src/container/richtext/tracker.rs @@ -19,6 +19,12 @@ use super::{ mod crdt_rope; mod id_to_cursor; + +/// Length of the placeholder span that stands for history the tracker has not +/// replayed (`Tracker::new_with_unknown`). Op positions must stay below it; see +/// `InnerListOp::check_positions`. +pub(crate) const UNKNOWN_SPAN_LEN: u32 = u32::MAX / 4; + pub(crate) use crdt_rope::CrdtRopeDelta; #[derive(Debug)] @@ -46,7 +52,7 @@ impl Tracker { }; let result = this.rope.tree.push(FugueSpan { - content: RichtextChunk::new_unknown(u32::MAX / 4), + content: RichtextChunk::new_unknown(UNKNOWN_SPAN_LEN), id: IdFull::new(UNKNOWN_PEER_ID, 0, 0), real_id: None, status: Status::default(), @@ -56,7 +62,7 @@ impl Tracker { }); this.id_to_cursor.insert_without_split( ID::new(UNKNOWN_PEER_ID, 0), - id_to_cursor::Cursor::new_insert(result.leaf, u32::MAX as usize / 4), + id_to_cursor::Cursor::new_insert(result.leaf, UNKNOWN_SPAN_LEN as usize), ); this } diff --git a/crates/loro-internal/src/encoding/json_schema.rs b/crates/loro-internal/src/encoding/json_schema.rs index 2f846c313..80fea8d8e 100644 --- a/crates/loro-internal/src/encoding/json_schema.rs +++ b/crates/loro-internal/src/encoding/json_schema.rs @@ -1049,6 +1049,9 @@ fn decode_op(op: json::JsonOp, arena: &SharedArena, peers: &Option>) } } // Note: The Future Type need try to parse Op from the unknown content }; + if let InnerContent::List(list_op) = &content { + list_op.check_positions()?; + } Ok(Op { counter, container: idx, diff --git a/crates/loro-internal/src/encoding/outdated_encode_reordered.rs b/crates/loro-internal/src/encoding/outdated_encode_reordered.rs index 703393976..4a4ba9b9e 100644 --- a/crates/loro-internal/src/encoding/outdated_encode_reordered.rs +++ b/crates/loro-internal/src/encoding/outdated_encode_reordered.rs @@ -472,6 +472,10 @@ pub(crate) fn decode_op( }), }; + if let crate::op::InnerContent::List(list_op) = &content { + list_op.check_positions()?; + } + Ok(content) } diff --git a/crates/loro-internal/src/history_cache.rs b/crates/loro-internal/src/history_cache.rs index c0cad3ed6..dcfaadc3f 100644 --- a/crates/loro-internal/src/history_cache.rs +++ b/crates/loro-internal/src/history_cache.rs @@ -352,6 +352,35 @@ impl ContainerHistoryCache { } } + /// Whether the movable list `idx` holds `elem_id` in the shallow root state. + /// + /// Elements created before the shallow root cannot be resolved from the trimmed + /// history; the checkout index only knows the ones still alive at the root + /// (`MovableListHistoryCache::record_shallow_root_state`). + pub(crate) fn shallow_root_has_movable_list_elem( + &self, + idx: ContainerIdx, + elem_id: IdLp, + ) -> bool { + let Some(state) = self.shallow_root_state.as_ref() else { + return false; + }; + let mut store = state.store.lock(); + let Some(c) = store.get_mut(idx) else { + return false; + }; + let ctx = ContainerCreationContext { + configure: &Default::default(), + peer: 0, + }; + match c.get_state_mut(idx, ctx) { + crate::state::State::MovableListState(l) => { + l.elements().contains_key(&elem_id.compact()) + } + _ => false, + } + } + pub(crate) fn find_text_chunks_in( &self, idx: ContainerIdx, diff --git a/crates/loro-internal/src/loro.rs b/crates/loro-internal/src/loro.rs index 7c712be5b..e8b28cf38 100644 --- a/crates/loro-internal/src/loro.rs +++ b/crates/loro-internal/src/loro.rs @@ -781,7 +781,21 @@ impl LoroDoc { } if self.is_detached() { + // An enclosing `import_batch` scope validates the whole batch before it + // reattaches (`BatchImportGuard::finish`). + let owns_rollback = + preflight.needs_state_apply_rollback && !oplog.has_import_rollback(); + if owns_rollback { + oplog.begin_import_rollback_with_arena(arena_checkpoint); + } let result = encoding::apply_decoded_changes_to_oplog(&mut oplog, changes); + if owns_rollback { + if let Err(e) = oplog.validate_movable_list_elem_refs_in_import_scope() { + oplog.rollback_import(); + return Err(e); + } + oplog.commit_import_rollback(); + } if result.has_deps_before_shallow_root { return Err(LoroError::ImportUpdatesThatDependsOnOutdatedVersion); } @@ -824,6 +838,14 @@ impl LoroDoc { let result = encoding::apply_decoded_changes_to_oplog(&mut oplog, changes); if &old_vv != oplog.vv() { + // The preflight enables rollback whenever the imported or unlocked + // pending changes hold movable-list ops, so other imports skip the scan. + if rollback_enabled { + if let Err(e) = oplog.validate_movable_list_elem_refs_in_import_scope() { + oplog.rollback_import(); + return Err(e); + } + } let mut diff = DiffCalculator::new(false); // Applying may have unlocked pending changes; the isolated fast path is only valid // when exactly the candidate batch was appended on top of the old version. @@ -861,6 +883,13 @@ impl LoroDoc { ) }; let mut state = self.state.lock(); + let diff = recalc_in_checkout_mode_if_needed( + &mut state, + &oplog, + &old_vv, + &old_frontiers, + diff, + ); if let Err(e) = state.apply_diff( InternalDocDiff { origin, @@ -2659,6 +2688,34 @@ struct BatchImportGuard<'a> { finished: bool, } +/// Forward import diffs of movable lists take untouched element fields from +/// `DocState`. When a (forged) op targets an element the state no longer holds, +/// recompute the diff in Checkout mode, which resolves elements from history the +/// same way `checkout`, `import_batch` and snapshot replay do. +/// See `context/movable-list-op-validation.md`. +fn recalc_in_checkout_mode_if_needed( + state: &mut DocState, + oplog: &OpLog, + old_vv: &VersionVector, + old_frontiers: &Frontiers, + diff: Vec, +) -> Vec { + if !state.needs_checkout_diff(&diff) { + return diff; + } + + DiffCalculator::new(true) + .calc_diff_internal( + oplog, + old_vv, + old_frontiers, + oplog.vv(), + oplog.dag.get_frontiers(), + None, + ) + .0 +} + impl BatchImportGuard<'_> { /// Leave batch-import mode, reattach if the doc was attached, and renew the txn. /// @@ -2688,7 +2745,13 @@ impl BatchImportGuard<'_> { // flight poisons the mutex. let mut checkout = Ok(()); if self.was_attached { - checkout = doc._checkout_to_latest_without_commit(true); + // The blobs were imported while detached inside this scope, so their + // movable-list element references are validated here, once per batch. + let validation = doc + .oplog + .lock() + .validate_movable_list_elem_refs_in_import_scope(); + checkout = validation.and_then(|()| doc._checkout_to_latest_without_commit(true)); if let Err(e) = &checkout { // `DocState::apply_diff` validates before mutating, so the state is // still at its pre-batch version; undoing the batch in the `OpLog` diff --git a/crates/loro-internal/src/oplog.rs b/crates/loro-internal/src/oplog.rs index 68e888c0d..4d5bcd250 100644 --- a/crates/loro-internal/src/oplog.rs +++ b/crates/loro-internal/src/oplog.rs @@ -24,14 +24,14 @@ use crate::encoding::decode_oplog; use crate::encoding::{ImportStatus, ParsedHeaderAndBody}; use crate::history_cache::ContainerHistoryCache; use crate::id::{Counter, PeerID, ID}; -use crate::op::{FutureInnerContent, ListSlice, RawOpContent, RemoteOp, RichOp}; +use crate::op::{FutureInnerContent, InnerContent, ListSlice, RawOpContent, RemoteOp, RichOp}; use crate::span::{HasCounterSpan, HasLamportSpan}; use crate::version::{Frontiers, ImVersionVector, VersionVector}; use crate::LoroError; use change_store::{BlockOpRef, ChangeStoreRollback}; use loro_common::{ContainerType, HasIdSpan, IdLp, IdSpan}; use rle::{HasLength, RleVec, Sliceable}; -use rustc_hash::FxHashSet; +use rustc_hash::{FxHashMap, FxHashSet}; use smallvec::SmallVec; pub use self::loro_dag::{AppDag, AppDagNode, FrontiersNotIncluded}; @@ -65,10 +65,26 @@ pub struct OpLog { } pub(crate) struct ImportRollback { - old_vv: VersionVector, arena: SharedArenaRollback, change_store: ChangeStoreRollback, pending: PendingChangesRollback, + /// Movable-list `Move`/`Set` element references inserted in this scope and not + /// validated yet (`OpLog::validate_movable_list_elem_refs_in_import_scope`). + movable_list_refs: Vec, +} + +#[derive(Clone, Copy)] +struct ResolvedElem { + target: ID, + container: ContainerIdx, + is_insert: bool, +} + +struct MovableListElemRef { + container: ContainerIdx, + op_id: ID, + op_lamport: Lamport, + elem_id: IdLp, } #[derive(Debug, Default, Clone, Copy)] @@ -243,7 +259,8 @@ impl OpLog { let rollback_old_vv = self .import_rollback .as_ref() - .and_then(|x| (!x.old_vv.is_empty()).then_some(&x.old_vv)); + .map(|x| x.change_store.old_vv()) + .filter(|vv| !vv.is_empty()); self.dag .handle_new_change(&change, from_local, rollback_old_vv); self.history_cache @@ -251,6 +268,23 @@ impl OpLog { .insert_by_new_change(&change, true, true); self.register_container_and_parent_link(&change); if let Some(rollback) = self.import_rollback.as_mut() { + for op in change.ops.iter() { + if op.container.get_type() != ContainerType::MovableList { + continue; + } + if let InnerContent::List( + list_op::InnerListOp::Move { elem_id, .. } + | list_op::InnerListOp::Set { elem_id, .. }, + ) = &op.content + { + rollback.movable_list_refs.push(MovableListElemRef { + container: op.container, + op_id: ID::new(change.id.peer, op.counter), + op_lamport: change.lamport + (op.counter - change.id.counter) as Lamport, + elem_id: *elem_id, + }); + } + } self.change_store.insert_change_with_rollback( change, true, @@ -270,13 +304,12 @@ impl OpLog { pub(crate) fn begin_import_rollback_with_arena(&mut self, arena: SharedArenaRollback) { debug_assert!(self.import_rollback.is_none()); - let old_vv = self.vv().clone(); self.dag.begin_import_rollback(); self.import_rollback = Some(ImportRollback { - old_vv: old_vv.clone(), arena, - change_store: ChangeStoreRollback::new(old_vv), + change_store: ChangeStoreRollback::new(self.vv().clone()), pending: Default::default(), + movable_list_refs: Vec::new(), }); } @@ -322,23 +355,24 @@ impl OpLog { continue; } - if self - .dag - .get_change_lamport_from_deps(&change.deps) - .is_none() - { - continue; - } - - ans.applies_to_dag = true; + // Inspect the ops even when the deps are not in the DAG yet: they may be + // earlier changes of this same import, which then unlock this one. if change.ops.iter().any(|op| { matches!( op.container.get_type(), - ContainerType::List | ContainerType::Tree + ContainerType::List | ContainerType::MovableList | ContainerType::Tree ) }) { ans.needs_state_apply_rollback = true; } + + if self + .dag + .get_change_lamport_from_deps(&change.deps) + .is_some() + { + ans.applies_to_dag = true; + } } // Any newly applied change can unlock pending changes whose ops are not @@ -1154,6 +1188,98 @@ impl OpLog { )) } + /// Reject the movable-list `Move`/`Set` ops inserted since the open import rollback + /// scope began whose `elem_id` is not an element inserted into the same list within + /// the op's causal history. + /// + /// Honest peers can only target elements they can see, and the movable-list diff + /// calculator relies on it (`last_pos`/`last_value` lookups). The refs are recorded + /// by `insert_new_change` while the scope is open, which covers directly imported + /// changes, unlocked pending changes and every blob of an `import_batch`. See + /// `context/movable-list-op-validation.md`. + pub(crate) fn validate_movable_list_elem_refs_in_import_scope( + &mut self, + ) -> Result<(), LoroError> { + let Some(rollback) = self.import_rollback.as_mut() else { + return Ok(()); + }; + let refs = std::mem::take(&mut rollback.movable_list_refs); + // Moves and sets keep hitting the same elements, and what an element id + // resolves to does not depend on the op, so resolve each one once. + let mut elems: FxHashMap> = FxHashMap::default(); + for r in refs { + let elem = *elems + .entry(r.elem_id) + .or_insert_with(|| self.resolve_movable_list_elem(r.elem_id)); + if !self.is_visible_movable_list_elem(&r, elem) { + return Err(LoroError::DecodeError( + format!( + "Movable list op {} targets element {}, which is not in the list's \ + causal history", + r.op_id, r.elem_id + ) + .into_boxed_str(), + )); + } + } + + Ok(()) + } + + /// Where `elem_id`'s insert op lives, or `None` if its lamport is not in the stored + /// history. The lookup searches both parsed and KV-only blocks of the peer, so + /// `None` means the element never existed or was trimmed before a shallow root. + fn resolve_movable_list_elem(&self, elem_id: IdLp) -> Option { + let change = self.change_store.get_change_by_lamport_lte(elem_id)?; + if change.lamport_end() <= elem_id.lamport { + return None; + } + let target = ID::new( + change.id.peer, + (elem_id.lamport - change.lamport) as Counter + change.id.counter, + ); + let op = change.get_op_with_counter(target.counter)?; + Some(ResolvedElem { + target, + container: op.container, + is_insert: matches!( + op.content, + InnerContent::List(list_op::InnerListOp::Insert { .. }) + ), + }) + } + + fn is_visible_movable_list_elem( + &self, + r: &MovableListElemRef, + elem: Option, + ) -> bool { + if r.elem_id.lamport >= r.op_lamport { + return false; + } + + let Some(elem) = elem else { + // History before a shallow root is trimmed. An op after the root can + // only see pre-root elements that are still alive at the root. + return !self.dag.shallow_since_vv().is_empty() + && self.with_history_cache(|h| { + h.shallow_root_has_movable_list_elem(r.container, r.elem_id) + }); + }; + if !elem.is_insert || elem.container != r.container { + return false; + } + + // Earlier ops of the op's own DAG node are by the same peer, so the node's + // start version covers every other peer's part of its causal history. + let target = elem.target; + (target.peer == r.op_id.peer && target.counter < r.op_id.counter) + || self + .dag + .get(r.op_id) + .is_some_and(|node| self.dag.ensure_vv_for(&node).includes_id(target)) + } + #[allow(unused)] pub(crate) fn id_to_idlp(&self, id_start: ID) -> IdLp { let change = self.get_change_at(id_start).unwrap(); diff --git a/crates/loro-internal/src/oplog/change_store.rs b/crates/loro-internal/src/oplog/change_store.rs index 6e947ab9d..c45d347f6 100644 --- a/crates/loro-internal/src/oplog/change_store.rs +++ b/crates/loro-internal/src/oplog/change_store.rs @@ -133,7 +133,27 @@ struct ChangeStoreInner { #[derive(Debug)] pub(crate) struct ChangeStoreRollback { old_vv: VersionVector, - blocks_before_mutation: BTreeMap>, + /// Pre-scope blocks that an import appended to, keyed by block id. `None` means the + /// block was flushed, so its KV copy is the pre-scope version. + blocks_before_mutation: BTreeMap>, +} + +/// What an unflushed block looked like before an import appended to it. +/// +/// Imports only append to a block (`ChangesBlock::push_change`): they push changes, +/// or push ops onto its last change, where an op may merge into the last op. So the +/// shape only needs the change count, the last change's op count and its last op. +/// Keeping an `Arc` of the whole block instead made the next append copy the block's +/// changes, and cloning the last change costs as much once remote changes keep +/// merging into it. +#[derive(Debug)] +struct BlockShape { + counter_end: Counter, + lamport_end: Lamport, + estimated_size: usize, + n_changes: usize, + /// `(merged op count, last op)` of the last change. + last_change_ops: Option<(usize, Option)>, } impl ChangeStoreRollback { @@ -144,13 +164,33 @@ impl ChangeStoreRollback { } } - fn record_block_before_mutation(&mut self, id: ID, block: Arc) { + /// The version when the import scope began. + pub(crate) fn old_vv(&self) -> &VersionVector { + &self.old_vv + } + + fn record_block_before_mutation(&mut self, id: ID, block: &ChangesBlock) { let old_end = self.old_vv.get(&id.peer).copied().unwrap_or(0); if id.counter >= old_end { return; } - self.blocks_before_mutation.entry(id).or_insert(block); + self.blocks_before_mutation.entry(id).or_insert_with(|| { + if block.flushed { + return None; + } + let changes = block + .content + .try_changes() + .expect("an unflushed block always holds parsed changes"); + Some(BlockShape { + counter_end: block.counter_range.1, + lamport_end: block.lamport_range.1, + estimated_size: block.estimated_size, + n_changes: changes.len(), + last_change_ops: changes.last().map(|c| (c.ops.len(), c.ops.last().cloned())), + }) + }); } } @@ -355,13 +395,57 @@ impl ChangeStore { // The name set may already include names from changes this rollback removes. That is // fine: stale names only make `old_history_may_touch_root_names` conservatively true. let mut inner = self.inner.lock(); + let mut touched_peers = FxHashSet::default(); inner.mem_parsed_kv.retain(|id, _| { let old_end = rollback.old_vv.get(&id.peer).copied().unwrap_or(0); - id.counter < old_end + let keep = id.counter < old_end; + if !keep { + touched_peers.insert(id.peer); + } + keep }); - for (id, block) in rollback.blocks_before_mutation { - inner.mem_parsed_kv.insert(id, block); + for (id, shape) in rollback.blocks_before_mutation { + touched_peers.insert(id.peer); + let Some(shape) = shape else { + // Flushed before the scope: the KV copy is the pre-scope block. + inner.mem_parsed_kv.remove(&id); + continue; + }; + let block = inner + .mem_parsed_kv + .get_mut(&id) + .expect("a block appended to during the scope stays cached"); + let block = Arc::make_mut(block); + let changes = Arc::make_mut( + block + .content + .changes_mut(&self.arena) + .expect("an unflushed block always holds parsed changes"), + ); + changes.truncate(shape.n_changes); + if let Some((n_ops, last_op)) = shape.last_change_ops { + let ops = changes.last_mut().unwrap().ops.vec_mut(); + ops.truncate(n_ops); + if let Some(last_op) = last_op { + *ops.last_mut().unwrap() = last_op; + } + } + block.counter_range.1 = shape.counter_end; + block.lamport_range.1 = shape.lamport_end; + block.estimated_size = shape.estimated_size; + } + + // `insert_change_inner` merges a change into the cached block right before it. + // A read during the scope (e.g. an lamport lookup) may have cached an older + // KV block of a touched peer, and removing the scope's newer blocks leaves it + // in front of the next insert with a counter gap ("counter should be + // continuous"). Blocks that are flushed are identical to their KV copy, so + // evict them; they are reloaded on demand. + if !touched_peers.is_empty() { + inner + .mem_parsed_kv + .retain(|id, block| !block.flushed || !touched_peers.contains(&id.peer)); } } @@ -1001,7 +1085,7 @@ mod mut_inner_kv { } if let Some(rollback) = &mut rollback { - rollback.record_block_before_mutation(*_id, block.clone()); + rollback.record_block_before_mutation(*_id, block); } match block.push_change( @@ -2032,6 +2116,136 @@ mod test { } } + /// Peer 1 history split into many blocks, encoded into a fresh store's KV only + /// (nothing parsed into `mem_parsed_kv`), plus the change that comes next. + fn kv_only_store_and_next_change() -> (ChangeStore, Counter, Change) { + let doc = LoroDoc::new_auto_commit(); + doc.set_peer_id(1).unwrap(); + for i in 0..100 { + let text = doc.get_text(format!("t{i}").as_str()); + text.insert(0, &"x".repeat(30), PosType::Unicode).unwrap(); + doc.commit_then_renew(); + } + let (bytes, end) = { + let oplog = doc.oplog().lock(); + let end = oplog.vv().get(&1).copied().unwrap(); + let bytes = oplog + .change_store + .encode_all(oplog.vv(), oplog.dag.frontiers()); + (bytes, end) + }; + doc.get_text("t0").insert(0, "y", PosType::Unicode).unwrap(); + doc.commit_then_renew(); + let next = { + let oplog = doc.oplog().lock(); + let change = oplog.get_change_at(ID::new(1, end)).unwrap(); + (*change).clone() + }; + let store = ChangeStore::new_for_test(); + let _ = store.import_all(bytes).unwrap(); + // Test builds of `import_all` parse each peer's last block; release builds do not. + store.inner.lock().mem_parsed_kv.clear(); + (store, end, next) + } + + #[test] + fn lamport_lookup_reads_kv_only_blocks() { + // Regression: `decode_block_range` read a version varint the block encoding + // does not have, shifting every field. KV-only blocks that did not start at + // counter 0 were skipped and block `0@P` got its lamport length as its start. + let (store, end, _) = kv_only_store_and_next_change(); + for l in (0..end as Lamport).step_by(13) { + store.inner.lock().mem_parsed_kv.clear(); + let change = store + .get_change_by_lamport_lte(IdLp::new(1, l)) + .unwrap_or_else(|| panic!("lamport {l} should be found")); + assert!(change.lamport <= l); + assert!(l < change.lamport + change.atom_len() as Lamport); + } + } + + #[test] + fn rollback_evicts_older_blocks_cached_during_the_scope() { + // An import creates the peer's newest block without loading the older ones, + // then a read caches an old KV block (the element lookup of the movable-list + // validator does this). Rolling back removes the newest block; the cached old + // block must not end up right before the next insert of the same change. + let (store, end, next) = kv_only_store_and_next_change(); + let mut old_vv = VersionVector::new(); + old_vv.insert(1, end); + let mut rollback = ChangeStoreRollback::new(old_vv); + store.insert_change_with_rollback(next.clone(), true, false, &mut rollback); + assert!(store.get_change(ID::new(1, 0)).is_some()); + store.rollback_import(rollback); + store.insert_change(next, true, false); + assert!(store.get_change(ID::new(1, end)).is_some()); + assert!(store.get_change(ID::new(1, 0)).is_some()); + } + + #[test] + fn rollback_restores_unflushed_block_that_was_appended_to() { + // The rollback keeps only the shape of a block an import appended to and + // truncates back to it. Appends both push changes and merge ops into the last + // change, so the restored store must encode exactly like one that never saw + // the rolled-back changes. + let doc = LoroDoc::new_auto_commit(); + doc.set_peer_id(1).unwrap(); + // Keep the commits as separate changes; the store below merges them itself. + doc.set_change_merge_interval(-1); + let text = doc.get_text("t"); + for i in 0..40 { + text.insert(i, "a", PosType::Unicode).unwrap(); + if i % 3 == 0 { + doc.get_map("m").insert("k", i as i64).unwrap(); + } + doc.commit_then_renew(); + } + let mut changes = Vec::new(); + let arena = { + let oplog = doc.oplog().lock(); + oplog + .change_store + .visit_all_changes(&mut |c| changes.push(c.clone())); + oplog.arena.clone() + }; + // The changes reference the doc's containers and values. + let new_store = || ChangeStore::new_mem(&arena, Arc::new(AtomicI64::new(0))); + assert!(changes.len() > 10); + let (first, rest) = changes.split_at(changes.len() / 2); + let vv_of = |cs: &[Change]| { + let mut vv = VersionVector::new(); + vv.insert(1, cs.last().unwrap().ctr_end()); + vv + }; + + let rolled_back = new_store(); + for c in first { + rolled_back.insert_change(c.clone(), true, false); + } + let mut rollback = ChangeStoreRollback::new(vv_of(first)); + for c in &rest[..rest.len() / 2] { + rolled_back.insert_change_with_rollback(c.clone(), true, false, &mut rollback); + } + rolled_back.rollback_import(rollback); + assert!(rolled_back + .get_change(ID::new(1, vv_of(first)[&1])) + .is_none()); + for c in rest { + rolled_back.insert_change(c.clone(), true, false); + } + + let direct = new_store(); + for c in &changes { + direct.insert_change(c.clone(), true, false); + } + let vv = vv_of(&changes); + let frontiers = Frontiers::from_id(changes.last().unwrap().id_last()); + assert_eq!( + rolled_back.encode_all(&vv, &frontiers), + direct.encode_all(&vv, &frontiers) + ); + } + #[test] fn lamport_lookup_finds_unflushed_mem_blocks() { // Regression: when the lamport binary search bails out, the fallback diff --git a/crates/loro-internal/src/oplog/change_store/block_encode.rs b/crates/loro-internal/src/oplog/change_store/block_encode.rs index f18d1db8f..7d99ece62 100644 --- a/crates/loro-internal/src/oplog/change_store/block_encode.rs +++ b/crates/loro-internal/src/oplog/change_store/block_encode.rs @@ -130,8 +130,6 @@ fn diagnose_block(block: &EncodedBlock) { info!(" values: {} bytes", block.values.len()); } -const VERSION: u16 = 0; - // MARK: encode_block /// It's assume that every change in the block share the same peer. pub fn encode_block(block: &[Change], arena: &SharedArena) -> Vec { @@ -463,19 +461,16 @@ impl ValueDecodedArenasTrait for ValueDecodeArena<'_> { } } +/// Read the counter and lamport ranges from the first fields of an encoded block +/// without decoding the rest. +/// +/// The block is a postcard-encoded [`EncodedBlock`], whose leading `u32` fields are +/// varints, so they can be read as LEB128 in declaration order. There is no version +/// prefix: reading one shifted every field and broke lamport lookups on blocks that +/// only live in the KV store (`ChangeStore::get_change_by_lamport_lte`). pub fn decode_block_range( mut bytes: &[u8], ) -> LoroResult<((Counter, Counter), (Lamport, Lamport))> { - let version = leb128::read::unsigned(&mut bytes).map_err(|e| { - LoroError::DecodeError(format!("Failed to read version: {e}").into_boxed_str()) - })?; - - if version as u16 != VERSION { - return Err(LoroError::DecodeError( - "Version mismatch".to_string().into_boxed_str(), - )); - } - let counter_start = leb128::read::unsigned(&mut bytes).map_err(|e| { LoroError::DecodeError(format!("Failed to read counter start: {e}").into_boxed_str()) })? as Counter; diff --git a/crates/loro-internal/src/oplog/pending_changes.rs b/crates/loro-internal/src/oplog/pending_changes.rs index a7d251f72..35361ffab 100644 --- a/crates/loro-internal/src/oplog/pending_changes.rs +++ b/crates/loro-internal/src/oplog/pending_changes.rs @@ -46,7 +46,7 @@ impl PendingChanges { change.ops.iter().any(|op| { matches!( op.container.get_type(), - ContainerType::List | ContainerType::Tree + ContainerType::List | ContainerType::MovableList | ContainerType::Tree ) }) }) diff --git a/crates/loro-internal/src/state.rs b/crates/loro-internal/src/state.rs index 7bb0db897..5068d729b 100644 --- a/crates/loro-internal/src/state.rs +++ b/crates/loro-internal/src/state.rs @@ -978,6 +978,26 @@ impl DocState { } } + /// Whether a forward import diff has to be recomputed in Checkout mode because a + /// movable-list op targets an element this state no longer holds. + /// See [`MovableListState::references_absent_elem`]. + pub(crate) fn needs_checkout_diff(&mut self, diffs: &[InternalContainerDiff]) -> bool { + diffs.iter().any(|diff| { + let crate::event::DiffVariant::Internal(InternalDiff::MovableList(delta)) = &diff.diff + else { + return false; + }; + match self.store.get_container(diff.idx) { + Some(State::MovableListState(state)) => state.references_absent_elem(delta), + Some(_) => unreachable!("movable list diff for a non movable list container"), + None => delta + .elements + .values() + .any(|elem| elem.pos.is_none() || elem.value_id.is_none()), + } + }) + } + fn validate_diff_batch(&mut self, diffs: &[InternalContainerDiff]) -> LoroResult<()> { for diff in diffs { let crate::event::DiffVariant::Internal(internal_diff) = &diff.diff else { diff --git a/crates/loro-internal/src/state/AGENTS.md b/crates/loro-internal/src/state/AGENTS.md index d35af4227..61f044e80 100644 --- a/crates/loro-internal/src/state/AGENTS.md +++ b/crates/loro-internal/src/state/AGENTS.md @@ -21,6 +21,9 @@ before changing mergeable child behavior. shallow-snapshot export; read [../../../../context/shallow-snapshot-style-redaction.md](../../../../context/shallow-snapshot-style-redaction.md) before changing the richtext snapshot codec or that pass. + `movable_list_state.rs` import validation and the Checkout-mode fallback for + moves of deleted elements: + [../../../../context/movable-list-op-validation.md](../../../../context/movable-list-op-validation.md). - `mergeable.rs`: logical child edge resolution for mergeable containers. - `dead_containers_cache.rs`: dead/alive tracking. Map/list deletions are final; tree and movable-list ones are dropped whenever a move could revive diff --git a/crates/loro-internal/src/state/movable_list_state.rs b/crates/loro-internal/src/state/movable_list_state.rs index 119b89a56..f7b187c9a 100644 --- a/crates/loro-internal/src/state/movable_list_state.rs +++ b/crates/loro-internal/src/state/movable_list_state.rs @@ -5,7 +5,9 @@ use std::sync::Weak; use tracing::{instrument, warn}; use generic_btree::BTree; -use loro_common::{CompactIdLp, ContainerID, IdFull, IdLp, LoroResult, LoroValue, PeerID, ID}; +use loro_common::{ + CompactIdLp, ContainerID, IdFull, IdLp, LoroError, LoroResult, LoroValue, PeerID, ID, +}; use rustc_hash::FxHashMap; use crate::{ @@ -1002,6 +1004,26 @@ impl MovableListState { } } +impl MovableListState { + /// Whether a forward (Linear/ImportGreaterUpdates) diff moves or sets an element + /// that is not in this state. + /// + /// Such diffs only carry the fields an op touched and take the rest from the + /// current element. Honest ops cannot produce this (an element absent here was + /// deleted before every op in the diff), but a forged op can; it has the same + /// meaning as a concurrent move/set of a deleted element, which only a Checkout + /// mode diff resolves. See `context/movable-list-op-validation.md`. + pub(crate) fn references_absent_elem( + &self, + diff: &crate::delta::MovableListInnerDelta, + ) -> bool { + diff.elements.iter().any(|(elem_id, elem)| { + (elem.pos.is_none() || elem.value_id.is_none()) + && !self.inner.elements().contains_key(elem_id) + }) + } +} + impl ContainerState for MovableListState { fn container_idx(&self) -> ContainerIdx { self.idx @@ -1011,6 +1033,58 @@ impl ContainerState for MovableListState { self.list().is_empty() && self.elements().is_empty() } + fn validate_diff(&self, diff: &InternalDiff) -> LoroResult<()> { + let InternalDiff::MovableList(diff) = diff else { + unreachable!() + }; + + // List items are addressed by op index (dead items included). + let mut cursor = 0usize; + let mut projected = self.inner.len_kind(IndexType::ForOp); + for item in diff.list.iter() { + match item { + DeltaItem::Retain { retain, .. } => { + cursor += retain; + if cursor > projected { + return Err(LoroError::DecodeError( + format!("movable list diff retains {cursor} items but state only has {projected}") + .into_boxed_str(), + )); + } + } + DeltaItem::Insert { insert, .. } => { + if cursor > projected { + return Err(LoroError::DecodeError( + format!("movable list diff inserts at {cursor} but state only has {projected}") + .into_boxed_str(), + )); + } + cursor += insert.len(); + projected += insert.len(); + } + DeltaItem::Delete { delete, .. } => { + if cursor + delete > projected { + return Err(LoroError::DecodeError( + format!("movable list diff deletes {delete} at {cursor} but state only has {projected}") + .into_boxed_str(), + )); + } + projected -= delete; + } + } + } + + // Import callers recompute such diffs in Checkout mode first + // (`DocState::needs_checkout_diff`); reaching here means that was skipped. + if self.references_absent_elem(diff) { + return Err(LoroError::internal( + "movable list diff lacks the value or position of an element absent from state", + )); + } + + Ok(()) + } + // How we apply the diff is coupled with the [DiffMode] we used to calculate the diff. // So be careful when you modify this function. #[instrument(skip_all)] diff --git a/crates/loro-internal/src/tests/import_atomicity.rs b/crates/loro-internal/src/tests/import_atomicity.rs index a3e70db45..bcd8cccf0 100644 --- a/crates/loro-internal/src/tests/import_atomicity.rs +++ b/crates/loro-internal/src/tests/import_atomicity.rs @@ -711,3 +711,155 @@ fn outdated_update_on_shallow_doc_is_dropped_not_pending() { "outdated changes must be dropped, not parked as pending" ); } + +/// Binary update whose movable-list move targets an element that never existed. +/// The public import paths reject it, so the carrier inserts the decoded change +/// straight into its oplog to obtain the bytes. +fn binary_update_moving_unknown_movable_list_elem() -> (LoroDoc, Vec) { + let doc = LoroDoc::new_auto_commit(); + doc.set_peer_id(1).unwrap(); + let list = doc.get_movable_list("list"); + list.insert(0, "a").unwrap(); + list.insert(1, "b").unwrap(); + doc.commit_then_renew(); + + let json = serde_json::json!({ + "schema_version": 1, + "start_version": {}, + "peers": ["1"], + "changes": [{ + "id": "2@0", "timestamp": 0, "deps": ["1@0"], "lamport": 2, "msg": null, + "ops": [{ + "container": "cid:root-list:MovableList", + "content": {"type": "move", "from": 0, "to": 1, "elem_id": "L99@0"}, + "counter": 2 + }] + }] + }); + let bytes = binary_update_bypassing_validation(&doc, json); + (doc, bytes) +} + +/// Encode the JSON `changes` on top of `doc` as a binary update. The carrier inserts +/// the decoded changes straight into its oplog, so no import validation runs. +fn binary_update_bypassing_validation(doc: &LoroDoc, json: serde_json::Value) -> Vec { + let carrier = LoroDoc::new(); + carrier + .import(&doc.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + let json: JsonSchema = serde_json::from_value(json).unwrap(); + { + let mut oplog = carrier.oplog().lock(); + let changes = + crate::encoding::json_schema::decode_json_changes(json, &oplog.arena).unwrap(); + crate::encoding::apply_decoded_changes_to_oplog(&mut oplog, changes); + } + carrier + .export(ExportMode::updates(&doc.oplog_vv())) + .unwrap() +} + +/// Peer 2 edits a map, and peer 3 moves an unknown element in a change that depends on +/// peer 2's. The import preflight used to skip peer 3's ops because its deps were not +/// in the DAG yet, so the blob was imported without a rollback scope or validation. +#[test] +fn binary_forged_move_behind_same_blob_dependency_is_rejected() { + let (doc, _) = binary_update_moving_unknown_movable_list_elem(); + let json = serde_json::json!({ + "schema_version": 1, + "start_version": {}, + "peers": ["1", "2", "3"], + "changes": [ + { + "id": "0@1", "timestamp": 0, "deps": ["1@0"], "lamport": 2, "msg": null, + "ops": [{ + "container": "cid:root-meta:Map", + "content": {"type": "insert", "key": "k", "value": 1}, + "counter": 0 + }] + }, + { + "id": "0@2", "timestamp": 0, "deps": ["0@1"], "lamport": 3, "msg": null, + "ops": [{ + "container": "cid:root-list:MovableList", + "content": {"type": "move", "from": 0, "to": 1, "elem_id": "L99@0"}, + "counter": 0 + }] + } + ] + }); + let bad = binary_update_bypassing_validation(&doc, json); + let vv_before = doc.oplog_vv(); + let frontiers_before = doc.oplog_frontiers(); + let state_before = doc.get_deep_value(); + + let err = doc.import(&bad).unwrap_err(); + assert!(matches!(err, LoroError::DecodeError(_)), "{err:?}"); + assert_doc_unchanged(&doc, &vv_before, &frontiers_before, &state_before); + + let err = doc.import_batch(std::slice::from_ref(&bad)).unwrap_err(); + assert!(matches!(err, LoroError::DecodeError(_)), "{err:?}"); + assert!(!doc.is_detached()); + assert_doc_unchanged(&doc, &vv_before, &frontiers_before, &state_before); + + doc.detach(); + let err = doc.import(&bad).unwrap_err(); + assert!(matches!(err, LoroError::DecodeError(_)), "{err:?}"); + doc.attach(); + assert_doc_unchanged(&doc, &vv_before, &frontiers_before, &state_before); + + doc.get_movable_list("list").mov(0, 1).unwrap(); + doc.commit_then_renew(); + assert_eq!(doc.state_frontiers(), doc.oplog_frontiers()); +} + +#[test] +fn binary_move_of_unknown_movable_list_elem_is_rejected_atomically() { + let (doc, bad) = binary_update_moving_unknown_movable_list_elem(); + let vv_before = doc.oplog_vv(); + let frontiers_before = doc.oplog_frontiers(); + let state_before = doc.get_deep_value(); + + let err = doc.import(&bad).unwrap_err(); + assert!(matches!(err, LoroError::DecodeError(_)), "{err:?}"); + assert_doc_unchanged(&doc, &vv_before, &frontiers_before, &state_before); + + doc.detach(); + let err = doc.import(&bad).unwrap_err(); + assert!(matches!(err, LoroError::DecodeError(_)), "{err:?}"); + doc.attach(); + assert_doc_unchanged(&doc, &vv_before, &frontiers_before, &state_before); + + doc.get_movable_list("list").mov(0, 1).unwrap(); + doc.commit_then_renew(); + assert_eq!(doc.state_frontiers(), doc.oplog_frontiers()); +} + +/// Blobs inside an attached `import_batch` are imported detached, so their element +/// references are validated once in `BatchImportGuard::finish`. +#[test] +fn import_batch_with_move_of_unknown_movable_list_elem_rolls_back() { + let (doc, bad) = binary_update_moving_unknown_movable_list_elem(); + let valid = LoroDoc::new_auto_commit(); + valid.set_peer_id(2).unwrap(); + valid + .import(&doc.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + valid.get_movable_list("list").push("v".into()).unwrap(); + valid.commit_then_renew(); + let valid_update = valid.export(ExportMode::updates(&doc.oplog_vv())).unwrap(); + + let vv_before = doc.oplog_vv(); + let frontiers_before = doc.oplog_frontiers(); + let state_before = doc.get_deep_value(); + + let err = doc.import_batch(&[valid_update, bad]).unwrap_err(); + assert!(matches!(err, LoroError::DecodeError(_)), "{err:?}"); + assert!(!doc.is_detached()); + assert!(!doc.oplog().lock().batch_importing); + assert_doc_unchanged(&doc, &vv_before, &frontiers_before, &state_before); + + doc.get_movable_list("list").mov(0, 1).unwrap(); + doc.commit_then_renew(); + assert_eq!(doc.state_frontiers(), doc.oplog_frontiers()); +} diff --git a/crates/loro-wasm/tests/movable_list_invalid_ops.test.ts b/crates/loro-wasm/tests/movable_list_invalid_ops.test.ts new file mode 100644 index 000000000..58102666c --- /dev/null +++ b/crates/loro-wasm/tests/movable_list_invalid_ops.test.ts @@ -0,0 +1,216 @@ +import { describe, expect, it } from "vitest"; +import { LoroDoc } from "../bundler/index"; + +// Malformed movable-list ops used to panic inside the doc locks, which traps the +// WASM instance ("RuntimeError: unreachable"). They must throw a readable error +// and leave the doc unchanged and usable. See context/movable-list-op-validation.md. + +/** Peer 1: list = [a, b, c] (lamports 0..=2), then delete `b` (counter 3). */ +function base(): LoroDoc { + const doc = new LoroDoc(); + doc.setPeerId(1); + const list = doc.getMovableList("list"); + list.insert(0, "a"); + list.insert(1, "b"); + list.insert(2, "c"); + doc.commit(); + list.delete(1, 1); + doc.commit(); + return doc; +} + +function forged(content: unknown): string { + return JSON.stringify({ + schema_version: 1, + start_version: {}, + peers: ["1"], + changes: [ + { + id: "4@0", + timestamp: 0, + deps: ["3@0"], + lamport: 4, + msg: null, + ops: [{ container: "cid:root-list:MovableList", content, counter: 4 }], + }, + ], + }); +} + +/** A map-only change, then `content` in a change that depends on it. */ +function forgedAfterMapChange(container: string, content: unknown): string { + return JSON.stringify({ + schema_version: 1, + start_version: {}, + peers: ["1"], + changes: [ + { + id: "4@0", + timestamp: 0, + deps: ["3@0"], + lamport: 4, + msg: null, + ops: [ + { + container: "cid:root-meta:Map", + content: { type: "insert", key: "k", value: 1 }, + counter: 4, + }, + ], + }, + { + id: "5@0", + timestamp: 0, + deps: ["4@0"], + lamport: 5, + msg: null, + ops: [{ container, content, counter: 5 }], + }, + ], + }); +} + +function expectRejected(doc: LoroDoc, json: string) { + const value = doc.toJSON(); + const vv = doc.oplogVersion().toJSON(); + let error: unknown; + try { + doc.importJsonUpdates(json); + } catch (e) { + error = e; + } + // LoroError reaches JS as its message. + expect(error).toBeDefined(); + expect(String(error)).toMatch(/Decode error/); + expect(String(error)).not.toMatch(/unreachable/); + expect(doc.toJSON()).toEqual(value); + expect(doc.oplogVersion().toJSON()).toEqual(vv); + + // Still usable. + const list = doc.getMovableList("list"); + list.push("local"); + doc.commit(); + const remote = new LoroDoc(); + remote.import(doc.export({ mode: "snapshot" })); + remote.getMovableList("list").push("remote"); + remote.commit(); + doc.import(remote.export({ mode: "update", from: doc.oplogVersion() })); + expect(doc.toJSON()).toEqual(remote.toJSON()); +} + +describe("malformed movable list ops", () => { + it("rejects moves and sets of unknown elements", () => { + expectRejected( + base(), + forged({ type: "move", from: 0, to: 1, elem_id: "L99@0" }), + ); + expectRejected( + base(), + forged({ type: "set", elem_id: "L99@0", value: "z" }), + ); + }); + + it("rejects forged ops that depend on a change in the same import", () => { + expectRejected( + base(), + forgedAfterMapChange("cid:root-list:MovableList", { + type: "move", + from: 0, + to: 1, + elem_id: "L99@0", + }), + ); + expectRejected( + base(), + forgedAfterMapChange("cid:root-list:MovableList", { + type: "move", + from: 9, + to: 0, + elem_id: "L0@0", + }), + ); + // `L0@0` belongs to `list`; moving it inside `other` must not put it in both. + expectRejected( + base(), + forgedAfterMapChange("cid:root-other:MovableList", { + type: "move", + from: 0, + to: 0, + elem_id: "L0@0", + }), + ); + }); + + it("rejects huge positions instead of trapping", () => { + const edge = Math.floor(0xffffffff / 4); + for (const content of [ + { type: "move", from: edge, to: 0, elem_id: "L0@0" }, + { type: "move", from: 0, to: edge, elem_id: "L0@0" }, + { type: "move", from: 0, to: 0xffffffff, elem_id: "L0@0" }, + { type: "insert", pos: edge, value: ["z"] }, + { type: "delete", pos: edge, len: 1, start_id: "0@0" }, + ]) { + expectRejected(base(), forged(content)); + } + }); + + it("applies a move of a deleted element like a concurrent move", () => { + const doc = base(); + doc.importJsonUpdates( + forged({ type: "move", from: 0, to: 1, elem_id: "L1@0" }), + ); + const replayed = new LoroDoc(); + replayed.import(doc.export({ mode: "update" })); + expect(doc.toJSON()).toEqual({ list: ["c", "b"] }); + expect(replayed.toJSON()).toEqual(doc.toJSON()); + }); +}); + +describe("docs loaded from a snapshot", () => { + it("import moves and sets of old elements from other peers", () => { + for (const set of [false, true]) { + const p2 = new LoroDoc(); + p2.setPeerId(2); + p2.getMovableList("list").insert(0, "a"); + p2.getMovableList("list").insert(1, "b"); + p2.commit(); + const p1 = new LoroDoc(); + p1.setPeerId(1); + p1.import(p2.export({ mode: "update" })); + p1.getMovableList("list").push("x"); + p1.commit(); + + const b = new LoroDoc(); + b.import(p1.export({ mode: "snapshot" })); + const vv = b.oplogVersion(); + if (set) { + p1.getMovableList("list").set(0, "A"); + } else { + p1.getMovableList("list").move(0, 2); + } + p1.commit(); + b.import(p1.export({ mode: "update", from: vv })); + expect(b.toJSON()).toEqual(p1.toJSON()); + } + }); + + it("getChangeAtLamport finds changes stored in the snapshot", () => { + const src = new LoroDoc(); + src.setPeerId(7); + const text = src.getText("t"); + for (let i = 0; i < 300; i++) { + text.insert(0, `${i}-abcdefghijklmnopqrstuvwxyz`); + src.commit(); + } + const end = src.oplogVersion().get("7")!; + for (let lamport = 0; lamport < end; lamport += 97) { + const loaded = new LoroDoc(); + loaded.import(src.export({ mode: "snapshot" })); + const expected = src.getChangeAtLamport("7", lamport); + const got = loaded.getChangeAtLamport("7", lamport); + expect(expected).toBeDefined(); + expect(got?.counter).toBe(expected!.counter); + expect(got?.lamport).toBe(expected!.lamport); + } + }); +}); diff --git a/crates/loro/tests/change_store_test.rs b/crates/loro/tests/change_store_test.rs index 0c7d669b9..a1f512086 100644 --- a/crates/loro/tests/change_store_test.rs +++ b/crates/loro/tests/change_store_test.rs @@ -50,3 +50,36 @@ fn test_compact_change_store() { doc.compact_change_store(); doc.checkout(&ID::new(0, 60).into()).unwrap(); } + +/// `get_change_with_lamport_lte` (`getChangeAtLamport` in JS) read the block range +/// of KV-only blocks one field off, so after loading a snapshot most lookups +/// returned `None` or the wrong change. +#[test] +fn get_change_with_lamport_lte_after_snapshot_import() { + let src = LoroDoc::new(); + src.set_peer_id(7).unwrap(); + let text = src.get_text("t"); + for i in 0..300 { + text.insert(0, &format!("{i}-abcdefghijklmnopqrstuvwxyz")) + .unwrap(); + src.commit(); + } + let end = src.oplog_vv().get(&7).copied().unwrap() as u32; + for lamport in (0..end).step_by(97) { + let expected = src.with_oplog(|o| { + o.get_change_with_lamport_lte(7, lamport) + .map(|c| (c.id(), c.lamport())) + }); + // Fresh load for every lookup, so the answer comes from the KV blocks. + let fresh = LoroDoc::new(); + fresh + .import(&src.export(ExportMode::Snapshot).unwrap()) + .unwrap(); + let got = fresh.with_oplog(|o| { + o.get_change_with_lamport_lte(7, lamport) + .map(|c| (c.id(), c.lamport())) + }); + assert_eq!(got, expected, "lamport {lamport}"); + assert!(expected.is_some()); + } +} diff --git a/crates/loro/tests/movable_list_invalid_ops.rs b/crates/loro/tests/movable_list_invalid_ops.rs new file mode 100644 index 000000000..1e7e6515d --- /dev/null +++ b/crates/loro/tests/movable_list_invalid_ops.rs @@ -0,0 +1,641 @@ +//! Forged or corrupted movable-list `Move`/`Set` ops used to panic inside the +//! state/oplog locks, poisoning them and aborting the process. Ops without a +//! meaning are now rejected with `Err` and leave the document untouched and +//! usable; a move/set of a deleted element is applied with the same semantics as +//! a concurrent one, identically on every import path. +//! See `context/movable-list-op-validation.md`. + +use loro::{ExportMode, LoroDoc, LoroValue, ToJson, VersionVector}; +use serde_json::{json, Value}; + +/// Peer 1 inserts `[a, b, c]` (counters/lamports 0..=2) and then deletes `b` +/// (counter 3), so the list is `[a, c]` and `b`'s element id is `L1@1`. +fn base() -> LoroDoc { + let doc = LoroDoc::new(); + doc.set_peer_id(1).unwrap(); + let list = doc.get_movable_list("list"); + list.insert(0, "a").unwrap(); + list.insert(1, "b").unwrap(); + list.insert(2, "c").unwrap(); + doc.commit(); + list.delete(1, 1).unwrap(); + doc.commit(); + doc +} + +/// A single change by peer 1, causally after `base()`, holding one op on `container`. +/// Element ids in `content` use the peer index into `["1", "2"]`. +fn forged_on(container: &str, content: Value) -> String { + json!({ + "schema_version": 1, + "start_version": {}, + "peers": ["1", "2"], + "changes": [{ + "id": "4@0", + "timestamp": 0, + "deps": ["3@0"], + "lamport": 4, + "msg": null, + "ops": [{"container": container, "content": content, "counter": 4}] + }] + }) + .to_string() +} + +fn forged(content: Value) -> String { + forged_on("cid:root-list:MovableList", content) +} + +fn move_deleted() -> Value { + json!({"type": "move", "from": 0, "to": 1, "elem_id": "L1@0"}) +} + +struct Snapshot { + value: LoroValue, + vv: VersionVector, +} + +impl Snapshot { + fn of(doc: &LoroDoc) -> Self { + Self { + value: doc.get_deep_value(), + vv: doc.oplog_vv(), + } + } + + fn assert_unchanged(&self, doc: &LoroDoc) { + assert_eq!(doc.get_deep_value(), self.value); + assert_eq!(doc.oplog_vv(), self.vv); + } +} + +/// The doc still accepts local edits and valid remote updates after a rejected import. +fn assert_usable(doc: &LoroDoc) { + let list = doc.get_movable_list("list"); + let len = list.len(); + list.push("local").unwrap(); + list.mov(0, list.len() - 1).unwrap(); + doc.commit(); + assert_eq!(list.len(), len + 1); + + let remote = LoroDoc::new(); + remote.set_peer_id(9).unwrap(); + remote + .import(&doc.export(ExportMode::Snapshot).unwrap()) + .unwrap(); + remote.get_movable_list("list").push("remote").unwrap(); + remote.commit(); + doc.import(&remote.export(ExportMode::updates(&doc.oplog_vv())).unwrap()) + .unwrap(); + assert_eq!(doc.get_deep_value(), remote.get_deep_value()); +} + +/// `base()` plus an edit by peer 2 that the forged op (deps `3@1`) cannot see, so +/// importing the forged op is concurrent with the doc's version. +fn base_with_concurrent_edit() -> LoroDoc { + let doc = base(); + doc.set_peer_id(2).unwrap(); + doc.get_movable_list("list").insert(0, "x").unwrap(); + doc.commit(); + doc +} + +fn assert_rejected(doc: &LoroDoc, json: String) { + let before = Snapshot::of(doc); + let err = doc.import_json_updates(json).unwrap_err(); + assert!( + matches!(err, loro::LoroError::DecodeError(_)), + "unexpected error: {err:?}" + ); + before.assert_unchanged(doc); + assert_usable(doc); +} + +/// Import `json` on top of `base()` through every path and check they agree with +/// a doc that replays the whole history at once. +fn assert_same_result_on_every_import_path(json: String, expected: Value) { + let carrier = LoroDoc::new(); + carrier + .import(&base().export(ExportMode::all_updates()).unwrap()) + .unwrap(); + // Detached imports only touch the oplog, which gives us the op as binary. + carrier.detach(); + carrier.import_json_updates(json.clone()).unwrap(); + let update = carrier + .export(ExportMode::updates_in_range(vec![loro::IdSpan::new( + 1, 4, 5, + )])) + .unwrap(); + let all = carrier.export(ExportMode::all_updates()).unwrap(); + + let replayed = LoroDoc::new(); + replayed.import(&all).unwrap(); + assert_eq!(replayed.get_deep_value().to_json_value(), expected); + + let from_snapshot = LoroDoc::new(); + from_snapshot + .import(&base().export(ExportMode::Snapshot).unwrap()) + .unwrap(); + let detached = base(); + detached.detach(); + let docs = [base(), base(), base(), from_snapshot, detached]; + docs[0].import_json_updates(json.clone()).unwrap(); + docs[1].import(&update).unwrap(); + docs[2].import_batch(&[update.clone()]).unwrap(); + docs[3].import_json_updates(json).unwrap(); + docs[4].import(&update).unwrap(); + docs[4].attach(); + for doc in docs { + assert_eq!(doc.get_deep_value(), replayed.get_deep_value()); + assert_eq!(doc.oplog_vv(), replayed.oplog_vv()); + assert_usable(&doc); + } +} + +#[test] +fn move_of_deleted_element_is_applied_like_a_concurrent_move() { + // `from: 0` points at `a` in the op's version, so `a`'s position is consumed + // and `b` comes back at index 1, exactly as for a concurrent move. + assert_same_result_on_every_import_path(forged(move_deleted()), json!({"list": ["c", "b"]})); +} + +#[test] +fn set_of_deleted_element_is_applied_like_a_concurrent_set() { + assert_same_result_on_every_import_path( + forged(json!({"type": "set", "elem_id": "L1@0", "value": "z"})), + json!({"list": ["a", "c"]}), + ); +} + +#[test] +fn move_or_set_of_unknown_element_is_rejected() { + let ops = [ + json!({"type": "move", "from": 0, "to": 1, "elem_id": "L99@0"}), + json!({"type": "set", "elem_id": "L99@0", "value": "z"}), + // Peer 2 has no ops at all in `base()`. + json!({"type": "move", "from": 0, "to": 1, "elem_id": "L0@1"}), + // Lamport 3 is `base()`'s delete op, not an element. + json!({"type": "set", "elem_id": "L3@0", "value": "z"}), + ]; + for op in ops { + assert_rejected(&base(), forged(op.clone())); + assert_rejected(&base_with_concurrent_edit(), forged(op)); + } +} + +#[test] +fn move_of_element_from_another_container_is_rejected() { + let doc = base(); + let other = doc.get_movable_list("other"); + other.insert(0, "o").unwrap(); + doc.commit(); + // `other`'s element is `L4@1`; target it from `list` in a change after it. + let json = json!({ + "schema_version": 1, + "start_version": {}, + "peers": ["1"], + "changes": [{ + "id": "5@0", "timestamp": 0, "deps": ["4@0"], "lamport": 5, "msg": null, + "ops": [{ + "container": "cid:root-list:MovableList", + "content": {"type": "move", "from": 0, "to": 1, "elem_id": "L4@0"}, + "counter": 5 + }] + }] + }) + .to_string(); + assert_rejected(&doc, json); +} + +#[test] +fn move_of_element_outside_causal_history_is_rejected() { + // Peer 2 inserts `y` concurrently with `base()`'s delete; the forged op by + // peer 1 depends only on `3@1` and so cannot have seen `y` (`L3@2`). + let doc = base(); + let concurrent = LoroDoc::new(); + concurrent.set_peer_id(2).unwrap(); + let mut before_delete = VersionVector::default(); + before_delete.insert(1, 3); + concurrent + .import( + &doc.export(ExportMode::updates_till(&before_delete)) + .unwrap(), + ) + .unwrap(); + concurrent.get_movable_list("list").push("y").unwrap(); + concurrent.commit(); + doc.import(&concurrent.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + assert_eq!( + doc.get_deep_value().to_json_value(), + json!({"list": ["a", "c", "y"]}) + ); + + assert_rejected( + &doc, + forged(json!({"type": "move", "from": 0, "to": 1, "elem_id": "L3@1"})), + ); +} + +#[test] +fn move_with_out_of_bounds_index_is_rejected() { + let ops = [ + json!({"type": "move", "from": 9, "to": 0, "elem_id": "L0@0"}), + json!({"type": "move", "from": 0, "to": 9, "elem_id": "L0@0"}), + json!({"type": "move", "from": 5, "to": 1, "elem_id": "L1@0"}), + ]; + for op in ops { + assert_rejected(&base(), forged(op.clone())); + assert_rejected(&base_with_concurrent_edit(), forged(op)); + } +} + +#[test] +fn detached_import_of_unknown_element_is_rejected() { + let doc = base(); + doc.detach(); + let before = Snapshot::of(&doc); + doc.import_json_updates(forged( + json!({"type": "move", "from": 0, "to": 1, "elem_id": "L99@0"}), + )) + .unwrap_err(); + before.assert_unchanged(&doc); + doc.attach(); + assert_usable(&doc); +} + +#[test] +fn import_batch_with_out_of_bounds_move_rolls_back_and_stays_attached() { + // Detached imports only touch the oplog (state bounds are checked when the + // doc reattaches), which lets us export the invalid op as binary. + let carrier = base(); + carrier.detach(); + carrier + .import_json_updates(forged( + json!({"type": "move", "from": 9, "to": 0, "elem_id": "L0@0"}), + )) + .unwrap(); + let forged_update = carrier + .export(ExportMode::updates_in_range(vec![loro::IdSpan::new( + 1, 4, 5, + )])) + .unwrap(); + + let doc = base(); + let valid = LoroDoc::new(); + valid.set_peer_id(5).unwrap(); + valid + .import(&doc.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + valid.get_movable_list("list").push("v").unwrap(); + valid.commit(); + let valid_update = valid.export(ExportMode::updates(&doc.oplog_vv())).unwrap(); + + let before = Snapshot::of(&doc); + doc.import(&forged_update).unwrap_err(); + before.assert_unchanged(&doc); + + doc.import_batch(&[valid_update, forged_update]) + .unwrap_err(); + assert!(!doc.is_detached()); + before.assert_unchanged(&doc); + assert_usable(&doc); +} + +#[test] +fn concurrent_delete_and_move_resurrects_moved_element() { + // Legitimate concurrency: a move of an element concurrently deleted by + // another peer is kept, because the move created a new position for it. + let a = LoroDoc::new(); + a.set_peer_id(1).unwrap(); + let list = a.get_movable_list("list"); + list.insert(0, "a").unwrap(); + list.insert(1, "b").unwrap(); + list.insert(2, "c").unwrap(); + a.commit(); + let b = LoroDoc::new(); + b.set_peer_id(2).unwrap(); + b.import(&a.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + + a.get_movable_list("list").delete(1, 1).unwrap(); + a.commit(); + b.get_movable_list("list").mov(1, 2).unwrap(); + b.get_movable_list("list").set(2, "B").unwrap(); + b.commit(); + + a.import(&b.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + b.import(&a.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + assert_eq!(a.get_deep_value(), b.get_deep_value()); + assert_eq!( + a.get_deep_value().to_json_value(), + json!({"list": ["a", "c", "B"]}) + ); + + // A peer that only now learns about both branches reaches the same state. + let c = LoroDoc::new(); + c.import(&a.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + assert_eq!(c.get_deep_value(), a.get_deep_value()); +} + +#[test] +fn move_and_set_on_shallow_root_element_still_import() { + let doc = LoroDoc::new(); + doc.set_peer_id(1).unwrap(); + let list = doc.get_movable_list("list"); + list.insert(0, "a").unwrap(); + list.insert(1, "b").unwrap(); + list.insert(2, "c").unwrap(); + doc.commit(); + list.insert(3, "d").unwrap(); + doc.commit(); + + let shallow = LoroDoc::new(); + shallow + .import( + &doc.export(ExportMode::shallow_snapshot(&doc.oplog_frontiers())) + .unwrap(), + ) + .unwrap(); + let vv = shallow.oplog_vv(); + + // Elements created before the shallow root are moved/set after it. + list.mov(0, 3).unwrap(); + list.set(0, "B").unwrap(); + doc.commit(); + shallow + .import(&doc.export(ExportMode::updates(&vv)).unwrap()) + .unwrap(); + assert_eq!(shallow.get_deep_value(), doc.get_deep_value()); + + // Unknown elements are still rejected on a shallow doc. + let json = json!({ + "schema_version": 1, + "start_version": {}, + "peers": ["1"], + "changes": [{ + "id": "6@0", "timestamp": 0, "deps": ["5@0"], "lamport": 6, "msg": null, + "ops": [{ + "container": "cid:root-list:MovableList", + "content": {"type": "move", "from": 0, "to": 1, "elem_id": "L99@0"}, + "counter": 6 + }] + }] + }) + .to_string(); + assert_rejected(&shallow, json); +} + +#[test] +fn move_of_element_deleted_before_shallow_root_is_rejected() { + // The trimmed history of a shallow doc only keeps elements alive at the root, + // and no valid op after the root can see `b`, which was deleted before it. + let src = base(); + let shallow_snapshot = src + .export(ExportMode::shallow_snapshot(&src.oplog_frontiers())) + .unwrap(); + for op in [ + move_deleted(), + json!({"type": "set", "elem_id": "L1@0", "value": "z"}), + json!({"type": "move", "from": 0, "to": 1, "elem_id": "L99@0"}), + ] { + let doc = LoroDoc::new(); + doc.import(&shallow_snapshot).unwrap(); + assert_rejected(&doc, forged(op)); + } + + // `a` is alive at the root, so it can still be moved and set. + let doc = LoroDoc::new(); + doc.import(&shallow_snapshot).unwrap(); + doc.import_json_updates(forged( + json!({"type": "set", "elem_id": "L0@0", "value": "A"}), + )) + .unwrap(); + assert_eq!( + doc.get_deep_value().to_json_value(), + json!({"list": ["A", "c"]}) + ); +} + +/// Two changes after `base()`: `C1` edits only a map, and `C2` holds `forged_op` on +/// `container` and depends on `C1`. When the import started, `C2`'s deps were not in +/// the DAG yet, and the import preflight used to skip its ops, so no rollback scope +/// (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 { + let (peers, c1_id, c2_id, c2_deps, c2_counter) = if two_peers { + (json!(["1", "2", "3"]), "0@1", "0@2", "0@1", 0) + } else { + (json!(["1"]), "4@0", "5@0", "4@0", 5) + }; + let c1_counter = if two_peers { 0 } else { 4 }; + json!({ + "schema_version": 1, + "start_version": {}, + "peers": peers, + "changes": [ + { + "id": c1_id, "timestamp": 0, "deps": ["3@0"], "lamport": 4, "msg": null, + "ops": [{ + "container": "cid:root-meta:Map", + "content": {"type": "insert", "key": "k", "value": 1}, + "counter": c1_counter + }] + }, + { + "id": c2_id, "timestamp": 0, "deps": [c2_deps], "lamport": 5, "msg": null, + "ops": [{"container": container, "content": forged_op, "counter": c2_counter}] + } + ] + }) + .to_string() +} + +#[test] +fn forged_op_depending_on_a_change_in_the_same_import_is_rejected() { + let other = "cid:root-other:MovableList"; + // The last field says whether the op is invalid without looking at state, so + // a detached import (which never touches state) can reject it too. + let cases = [ + ( + "list", + json!({"type": "move", "from": 0, "to": 1, "elem_id": "L99@0"}), + true, + ), + ( + "list", + json!({"type": "set", "elem_id": "L99@0", "value": "z"}), + true, + ), + ( + "list", + json!({"type": "move", "from": 9, "to": 0, "elem_id": "L0@0"}), + false, + ), + // `L0@0` is an element of `list`, not of `other`. + ( + "other", + json!({"type": "move", "from": 0, "to": 0, "elem_id": "L0@0"}), + true, + ), + ]; + for (target, op, history_only) in cases { + let container = if target == "list" { + "cid:root-list:MovableList" + } else { + other + }; + for two_peers in [false, true] { + let json = forged_after_map_change(container, op.clone(), two_peers); + assert_rejected(&base(), json.clone()); + assert_rejected(&base_with_concurrent_edit(), json.clone()); + if !history_only { + continue; + } + + let detached = base(); + detached.detach(); + let before = Snapshot::of(&detached); + detached.import_json_updates(json).unwrap_err(); + before.assert_unchanged(&detached); + detached.attach(); + assert_usable(&detached); + } + } +} + +#[test] +fn cross_container_move_is_not_applied_to_two_lists() { + // Used to be accepted silently when it rode behind a same-import dependency: + // the element ended up in both lists and a full replay rejected the history. + let doc = base(); + doc.get_movable_list("other").push("o").unwrap(); + doc.commit(); + let before = Snapshot::of(&doc); + // `other`'s element is `L4@0`; move it inside `list`. + let json = json!({ + "schema_version": 1, + "start_version": {}, + "peers": ["1"], + "changes": [ + { + "id": "5@0", "timestamp": 0, "deps": ["4@0"], "lamport": 5, "msg": null, + "ops": [{ + "container": "cid:root-meta:Map", + "content": {"type": "insert", "key": "k", "value": 1}, + "counter": 5 + }] + }, + { + "id": "6@0", "timestamp": 0, "deps": ["5@0"], "lamport": 6, "msg": null, + "ops": [{ + "container": "cid:root-list:MovableList", + "content": {"type": "move", "from": 0, "to": 1, "elem_id": "L4@0"}, + "counter": 6 + }] + } + ] + }) + .to_string(); + doc.import_json_updates(json).unwrap_err(); + before.assert_unchanged(&doc); + assert_usable(&doc); +} + +#[test] +fn huge_positions_are_rejected_before_diff_calculation() { + // Positions at or past the diff tracker's placeholder span used to panic inside + // the tracker with the doc locks held. + let edge = (u32::MAX / 4) as u64; + let ops = [ + json!({"type": "move", "from": edge, "to": 0, "elem_id": "L0@0"}), + json!({"type": "move", "from": 0, "to": edge, "elem_id": "L0@0"}), + json!({"type": "move", "from": 0, "to": u32::MAX, "elem_id": "L0@0"}), + json!({"type": "insert", "pos": edge, "value": ["z"]}), + json!({"type": "delete", "pos": edge, "len": 1, "start_id": "0@0"}), + json!({"type": "delete", "pos": 0, "len": edge + 1, "start_id": "0@0"}), + // Just below the limit: an ordinary out-of-bounds error from state validation. + json!({"type": "move", "from": 0, "to": edge - 1, "elem_id": "L0@0"}), + json!({"type": "move", "from": edge - 1, "to": 0, "elem_id": "L0@0"}), + ]; + for op in ops { + assert_rejected(&base(), forged(op.clone())); + assert_rejected(&base_with_concurrent_edit(), forged(op.clone())); + assert_rejected( + &base(), + forged_after_map_change("cid:root-list:MovableList", op, false), + ); + } + + let doc = base(); + doc.get_list("plain").push("p").unwrap(); + doc.get_text("text").insert(0, "t").unwrap(); + doc.commit(); + for (container, op) in [ + ( + "cid:root-plain:List", + json!({"type": "insert", "pos": edge, "value": ["z"]}), + ), + ( + "cid:root-text:Text", + json!({"type": "insert", "pos": edge, "text": "z"}), + ), + ( + "cid:root-text:Text", + json!({"type": "delete", "pos": edge, "len": 1, "start_id": "5@0"}), + ), + ] { + let json = json!({ + "schema_version": 1, "start_version": {}, "peers": ["1"], + "changes": [{"id": "6@0", "timestamp": 0, "deps": ["5@0"], "lamport": 6, "msg": null, + "ops": [{"container": container, "content": op, "counter": 6}]}] + }) + .to_string(); + assert_rejected(&doc, json); + } +} + +#[test] +fn reversed_delete_at_start_is_still_accepted() { + let doc = base(); + let json = forged(json!({"type": "delete", "pos": 0, "len": -1, "start_id": "0@0"})); + doc.import_json_updates(json).unwrap(); + assert_eq!(doc.get_deep_value().to_json_value(), json!({"list": ["c"]})); +} + +#[test] +fn move_and_set_of_old_element_import_on_snapshot_loaded_doc() { + // The element's insert lives only in a KV block of the loaded snapshot. The + // lamport lookup used to misread KV-only blocks, so this honest update was + // rejected as "not in the list's causal history". + for set in [false, true] { + let p2 = LoroDoc::new(); + p2.set_peer_id(2).unwrap(); + let l = p2.get_movable_list("list"); + l.insert(0, "a").unwrap(); + l.insert(1, "b").unwrap(); + p2.commit(); + let p1 = LoroDoc::new(); + p1.set_peer_id(1).unwrap(); + p1.import(&p2.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + p1.get_movable_list("list").push("x").unwrap(); + p1.commit(); + + let b = LoroDoc::new(); + b.import(&p1.export(ExportMode::Snapshot).unwrap()).unwrap(); + let vv = b.oplog_vv(); + if set { + p1.get_movable_list("list").set(0, "A").unwrap(); + } else { + p1.get_movable_list("list").mov(0, 2).unwrap(); + } + p1.commit(); + b.import(&p1.export(ExportMode::updates(&vv)).unwrap()) + .unwrap(); + assert_eq!(b.get_deep_value(), p1.get_deep_value()); + } +}