diff --git a/.changeset/unregistered-tree-meta-parent.md b/.changeset/unregistered-tree-meta-parent.md new file mode 100644 index 000000000..bdfa9de37 --- /dev/null +++ b/.changeset/unregistered-tree-meta-parent.md @@ -0,0 +1,5 @@ +--- +"loro-crdt": patch +--- + +Fix a crash in documents loaded from a snapshot or created by `fork()`: `isDeleted()`, edits, `getContainerById`, and events touching the metadata of a tree node that was created under an already deleted parent threw `RuntimeError: unreachable` ("Parent is not registered"). The parent of such a container is now found from the change that created it. Also, an import rejected by state validation could leave loaded history pointing at containers and values its rollback had removed, so a later export, checkout, or import crashed (#1161). A change block that is present but cannot be parsed now fails with an internal error when a query needs it, instead of the container reading as deleted. diff --git a/AGENTS.md b/AGENTS.md index 1d75766c6..d1277aef0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -46,6 +46,9 @@ Loro is a Rust CRDT workspace with JS/WASM packaging and a MoonBit codec. - Bounded decoded-value cache in `InnerStore` (second-chance FIFO, eviction safety contract, loro-dev/loro#1092): [context/container-value-cache.md](context/container-value-cache.md). +- How the arena learns a container's parent (parsed changes, state KV, the op + log's creator resolver), its locking (including `ChangeStore`'s lock order), + and import rollback: [context/arena-parent-links.md](context/arena-parent-links.md). - Which deletions `DocState::is_deleted` caches, and how tree/movable-list children are revived: [context/dead-container-cache.md](context/dead-container-cache.md). - Movable-list `apply_diff`/undo: delete + re-insert of a child becomes a move, diff --git a/context/arena-parent-links.md b/context/arena-parent-links.md new file mode 100644 index 000000000..b0306264c --- /dev/null +++ b/context/arena-parent-links.md @@ -0,0 +1,204 @@ +# Arena Parent Links + +Verified against code 2026-09-29. + +`SharedArena` (`crates/loro-internal/src/arena.rs`) stores each container's +parent. Liveness (`DocState::is_deleted`), paths (`DocState::get_path`, +`LoroDoc::get_path_to_container`), depth (`SharedArena::get_depth`, used by +diff calculation and `ContainerWrapper::new`), and event routing +(`SharedArena::with_ancestors`) all walk it. + +## Where a parent link comes from + +1. Parsing a change: `register_container_and_parent_link` (`parent.rs`) links + every container an op creates to the op's container. Local ops do the same + through `DocState::set_container_parent_by_raw_op`, and a tree `Create` diff + through `DocState::register_meta_parents_of_created_tree_nodes`. +2. Registering a mergeable root, whose ID encodes its parent. +3. The state KV (`parent_resolver`, installed by `InnerStore::decode` when a + snapshot is loaded): a container's own state entry records its parent. +4. The op log (`creator_resolver`, installed by `OpLog::new` from + `ChangeStore::creator_resolver`): a normal container's ID is the ID of the + op that created it, so loading that op's change block registers the link + through (1). + +`SharedArena::get_parent` and `get_depth` try them in that order. +`DocState::does_container_exist` (behind `get_container`, `has_container`, and +WASM `getContainerById`) calls `SharedArena::find_created_container` for a +normal ID that is not registered yet. + +## Why (4) is needed + +A document loaded from a snapshot, or created by `fork()`, parses its change +blocks lazily, so (1) has not run for most of the history. A full snapshot has +a state entry only for containers that were materialized. The metadata of a +tree node created inside an already deleted subtree (a concurrent child of a +deleted parent) is never materialized, so neither (1) nor (3) knew its parent, +and `is_deleted()`, edits, and events touching it panicked with "Parent is not +registered" under the state lock (loro-dev/loro#1158). Shallow snapshots are +not affected: shallow export ensures every container of the retention walk, +which includes the metadata of deleted nodes. + +## Locking + +The creator resolver is the only access to the change store that does not hold +the document's op log lock. `is_deleted`, `has_container`, +`get_path_to_container`, and `ContainerWrapper::new` reach it under the state +lock, and event emission (`with_ancestors`) under no document lock. So: + +- `ChangeStore` takes its locks in one order, documented on the struct: + `root_history_names`, `external_kv`, `inner`, `external_vv`, and the arena's + lock last (parsing registers containers). Before loro-dev/loro#1159's review, + `get_parsed_block_in`, `flush_and_compact`, and `get_change_by_lamport_lte` + took `inner` before `external_kv`, which was harmless while the op log lock + serialized every caller. With the resolver, `is_deleted` on one thread and + `len_changes`, `export`, or `vv_to_frontiers` on another deadlocked. + `ChangeStore::load_parsed_block` looks for a cached block under `inner` + alone, then takes `external_kv` and `inner` in order and looks again before + it loads the block from the KV store. +- The arena calls the resolver without holding its own lock. + `ArenaContainers::get_depth` runs under the arena lock, so + `SharedArena::get_depth` first resolves the missing links of the ancestor + chain. It fails fast on a cycle, which would otherwise loop. +- The change store never resolves an arena parent while holding `inner`. + `ChangeStore::visit_all_changes` runs its callback under that lock; none of + its callers resolve parents. +- Import rollbacks run under the state lock and roll the arena back under + `inner`; see "Import rollback" below. +- Another thread can register a container at any time without its parent, for + example by creating a handler. Code that looks a container up twice must not + assume the answer stayed the same: `DocState::does_container_exist` decides + from its first lookup. + +It holds weak references to the change store, so the arena does not keep the +store alive. `SharedArena::fork` drops it (it would read the source document's +op log). + +## When there is no parent + +The resolver answers `CreatorOp::Loaded` when a change block holds the op (it +is parsed now, so the parent of every container the op creates is registered) +and `CreatorOp::Absent` when none does. `get_parent` returns `None` for a +normal container that no source knows: + +- no op in the history creates it: an ID from the user that is not a + container, or an op not received yet; +- a shallow document whose creating op was trimmed and whose state does not + hold the container. Shallow export keeps everything that can be alive or + revived, so such a container is dead at the root. + +No path leads to it, so callers treat it like a deleted container: +`is_deleted` answers `true` (and does not cache it), and `get_path` and +`get_depth` answer `None`. Broken invariants still fail fast: + +- The resolver panics when a block holds the op but cannot be decoded or + parsed, instead of answering `Absent`. +- An arena without a creator resolver (not owned by an op log) panics. +- `ContainerWrapper::new` panics for a container without a parent, so a + container never gets state without one. +- A local op's containers are linked by the transaction + (`set_container_parent_by_raw_op`) before the change reaches the op log, so + the resolver cannot supply a link that a local op path forgot. Debug builds + check the links when a local change is committed + (`parent::assert_local_parent_links_registered`). + +## Cost of looking up an ID + +`has_container` / `get_container` / `getContainerById` for a normal ID that is +not registered ask the resolver: + +- No block holds the op (beyond the history, an unknown peer, before a shallow + root): a KV scan and one block header, with no parse. About 0.5 µs per ID. +- A block holds the op: the block is parsed once and stays cached, like any + other lazy access to that part of the history. This happens even when the op + creates no container, because only the op's content tells. The version vector + or DAG can tell only whether the op is in the history, not what it creates. + +Measured with 2k IDs on a 20k-node, 100k-op document, `main` vs the resolver: + +| IDs | `main` | resolver | +|---|---|---| +| in the history, first pass | 1.3 ms | 6.1 ms | +| in the history, second pass | 0.66 ms | 0.78 ms | +| beyond the history | 1.1 ms | 2.1 ms | + +A vv shortcut for IDs beyond the history would have to be exact for every path +that writes the KV store, or a live container would read as deleted. Measured +2026-09-28 (loro-dev/loro#1159). + +## Import rollback + +A failed import can parse old change blocks while it computes its diff +(including through the creator resolver) or validates movable-list ops +(`OpLog::resolve_movable_list_elem`), which registers containers and allocates +their values. The arena rollback (`SharedArena::rollback`) drops registrations +and values made during the import, so `ChangeStore::rollback_import`, after +truncating and evicting blocks (see +[movable-list-op-validation.md](movable-list-op-validation.md)), drops the +parsed changes of the kept blocks that were parsed since the import began; the +next access parses again and registers again. `ChangesBlock::ensure_changes` +records the arena's `ArenaExtent` (containers, values, text) when it parses a +block. A block parsed before the checkpoint can only refer to what was there +then, so it keeps its parsed changes (`SharedArenaRollback::keeps`). Dropping +them all made every failed import an O(blocks) pass and the next history read a +full reparse. Blocks without bytes were built from changes inserted before the +import, so their containers were registered then. Before 2026-09-28 the parsed +ops kept indices and value slices past the truncated arena: exporting the +history hit `unreachable!` in the JSON encoder, and a later checkout panicked on +a missing value (loro-dev/loro#1161). + +The resolver can run during any import, because it does not take the op log +lock. So every arena rollback of a failed import goes through the change store +and runs under the state lock: + +- Under `inner`: `ChangeStore::rollback_arena` rolls the arena back and drops + the parsed changes of blocks parsed since the checkpoint. `rollback_import` + ends with it. The + early returns of `LoroDoc::import_changes_and_apply_delta_to_state_if_needed` + (a decode error, or updates that depend on history before the shallow root) + call it through `OpLog::rollback_arena`; they used to call + `SharedArena::rollback` directly. Blocks are parsed under `inner`, so a + block the resolver loads is either dropped by the rollback or parsed against + the rolled-back arena. A failed snapshot import replaces the change store: + `ChangeStore::retire` rolls the arena back and makes the old store answer + `Absent`, so a resolver that reached it before the swap does not register the + discarded history. +- Under the state lock: `OpLog::rollback_import`, `rollback_arena`, and + `reset_to_empty_for_failed_snapshot_import` take `&DocState` as proof. A + state-locked query (`has_container`, `is_deleted`, `get_path_to_container`) + can register containers through the resolver and then use their indices + before it returns; the rollback must not free them in between. Without the + state lock, a loom model hit `get_depth` on a freed index. + +Only the state-locked queries are covered. A handler created on another thread +during the import keeps its index, and `DocState` is not rolled back: its store +and dead-container cache keep entries at the freed indices, and the next +container registered at one inherits them. `get_path_to_container` answers +`None` if its registration is already gone. See loro-dev/loro#1164 +(pre-existing). + +## Testing pitfall + +Under `cfg(test)`, `ChangeStore::import_all` parses the last change of every +peer at load. Unit tests in `loro-internal` therefore do not see lazily parsed +blocks unless the change is not the peer's last; the `loro` crate's tests run +the real path. + +## Tests + +- `crates/loro/tests/unregistered_container_parent.rs`: repros for every entry + point with snapshot, fork, and shallow sources, the dumped state from the + review harness, the import rollbacks (including #1161's missing values), a + thread stress of queries against history readers + (`UNREGISTERED_PARENT_THREAD_TRIALS`), and a random comparison of every tree + meta against a full-history import (`UNREGISTERED_PARENT_SEEDS=0..300`). +- `crates/loro/tests/unregistered_container_parent.rs` + `queries_race_with_failing_imports`: queries against imports that fail at + decode or depend on history before a shallow root, swept across the import's + measured duration. +- `crates/loro/tests/multi_thread_test.rs` (run by `pnpm test-loom`): the lock + order (`resolving_a_meta_parent_while_another_thread_reads_the_history`) and + an import that fails at decode + (`resolving_a_meta_parent_while_an_import_fails`), as loom models. +- Unit tests in `arena.rs` (the resolver runs outside the arena lock, the + `None` / panic / cycle cases) and `parent.rs` (the local link check). diff --git a/context/movable-list-op-validation.md b/context/movable-list-op-validation.md index 84cb775b7..575dd298e 100644 --- a/context/movable-list-op-validation.md +++ b/context/movable-list-op-validation.md @@ -142,6 +142,9 @@ from the element in `DocState`. Detection and fallback: 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. + - The lookup's parse also registers containers that the arena rollback forgets, + so rollback then drops the parsed changes of the blocks parsed since the + import began ([arena-parent-links.md](arena-parent-links.md#import-rollback)). - **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. diff --git a/crates/loro-internal/AGENTS.md b/crates/loro-internal/AGENTS.md index 69ac67328..a54380f90 100644 --- a/crates/loro-internal/AGENTS.md +++ b/crates/loro-internal/AGENTS.md @@ -19,6 +19,11 @@ over graceful degradation. or op/value encoding. - `src/oplog/` and `src/dag/`: change storage, dependency ordering, pending changes, version vectors/frontiers, shallow roots, and history traversal. + `ChangeStore` is also read without the op log lock, so keep its lock order: + [../../context/arena-parent-links.md](../../context/arena-parent-links.md). +- `src/arena.rs`: container IDs, indices, and parent links. How a parent is + found in a lazily loaded document, and the locking rules: + [../../context/arena-parent-links.md](../../context/arena-parent-links.md). - `src/state.rs` and `src/state/`: materialized document state, container stores, diff application, checkout/replay, deep value, dead-container tracking, and mergeable container visibility. Read `src/state/AGENTS.md` and diff --git a/crates/loro-internal/src/arena.rs b/crates/loro-internal/src/arena.rs index 563f7b46b..03c92fc4d 100644 --- a/crates/loro-internal/src/arena.rs +++ b/crates/loro-internal/src/arena.rs @@ -14,7 +14,7 @@ use crate::{ LoroValue, }; use append_only_bytes::BytesSlice; -use loro_common::PeerID; +use loro_common::{PeerID, ID}; use rustc_hash::FxHashMap; use std::fmt; use std::{ @@ -25,6 +25,22 @@ use std::{ pub(crate) struct LoadAllFlag; type ParentResolver = dyn Fn(ContainerID) -> Option + Send + Sync + 'static; +/// Loads the change holding the op with the given ID, if the op log has it. Parsing a change +/// registers the parent link of every container its ops create, and a normal container's ID is +/// the ID of the op that created it. It panics if the op log has the change but cannot parse +/// it. See `context/arena-parent-links.md`. +type CreatorResolver = dyn Fn(&SharedArena, ID) -> CreatorOp + Send + Sync + 'static; + +/// What the op log's history knows about an op ID, answered by the [`CreatorResolver`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum CreatorOp { + /// The change holding the op is parsed, so the parent of every container the op creates + /// is registered. + Loaded, + /// The history has no op with this ID: it was not received or committed yet, or it is + /// before the shallow root. + Absent, +} #[derive(Default)] struct ArenaContainers { @@ -49,6 +65,14 @@ struct ArenaContainers { /// Locking: the resolver may read the state KV store. Code that loads from KV must snapshot /// KV data and release the KV lock before taking the arena lock. parent_resolver: Option>, + /// Set by the op log. Used after `parent_resolver` when a container's parent is still + /// unknown, e.g. the meta of a tree node that was never alive in a document loaded from a + /// snapshot, whose change blocks are parsed lazily. + /// + /// Locking: it takes the change store's locks and parses changes, which registers + /// containers in this arena, so it must be called without holding the arena lock. It runs + /// without the op log lock; see `ChangeStore::creator_resolver`. + creator_resolver: Option>, } #[derive(Default)] @@ -86,6 +110,24 @@ pub(crate) struct SharedArenaRollback { str: StrArenaCheckpoint, } +/// How far the arena reached when something that refers into it was built, e.g. a parsed +/// change block: it can only refer to containers, values, and text allocated before. +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct ArenaExtent { + containers: usize, + values: usize, + str_bytes: usize, +} + +impl SharedArenaRollback { + /// Whether rolling back to this checkpoint keeps everything within `extent`. + pub(crate) fn keeps(&self, extent: ArenaExtent) -> bool { + extent.containers <= self.container_len + && extent.values <= self.values_len + && extent.str_bytes <= self.str.bytes_len() + } +} + #[derive(Debug)] pub struct StrAllocResult { /// unicode start @@ -267,6 +309,8 @@ impl SharedArena { root_c_idx: containers.root_c_idx.clone(), top_level_root_c_idx: containers.top_level_root_c_idx.clone(), parent_resolver: containers.parent_resolver.clone(), + // It reads the source document's op log. + creator_resolver: None, } }), values: Mutex::new(self.inner.values.lock().clone()), @@ -275,6 +319,15 @@ impl SharedArena { } } + /// See [`ArenaExtent`]. + pub(crate) fn extent(&self) -> ArenaExtent { + ArenaExtent { + containers: self.inner.containers.read().container_idx_to_id.len(), + values: self.inner.values.lock().len(), + str_bytes: self.inner.str.lock().bytes_len(), + } + } + pub(crate) fn checkpoint_for_rollback(&self) -> SharedArenaRollback { let containers = self.inner.containers.read(); let container_len = containers.container_idx_to_id.len(); @@ -292,6 +345,9 @@ impl SharedArena { } } + /// Call it through the op log's change store (`ChangeStore::rollback_arena` and friends), + /// which does it under the lock the creator resolver parses under and drops the parsed + /// changes that may refer to what this removes. See `context/arena-parent-links.md`. pub(crate) fn rollback(&self, checkpoint: SharedArenaRollback) { let mut containers = self.inner.containers.write(); let removed_ids = containers @@ -433,33 +489,88 @@ impl SharedArena { }); } + /// The parent of `child`, or `None` for a top-level root. + /// + /// Also `None` for a normal container that no op in the history creates (an ID from the + /// user that is not a container, an op not received yet, or a container that is not in + /// the state of a shallow document and whose creating op was trimmed): no path leads to + /// it, so it is treated like a container whose parent no longer holds it. A history that + /// has the op but cannot parse it panics instead (see [`CreatorResolver`]). pub fn get_parent(&self, child: ContainerIdx) -> Option { - let (child_id, resolver) = { + match self.resolve_parent(child) { + Some(parent) => parent, + None => { + assert!( + self.inner.containers.read().creator_resolver.is_some(), + "InternalError: Parent is not registered" + ); + None + } + } + } + + /// Finds and registers `child`'s parent edge: the registered edge, else the state KV + /// (`parent_resolver`), else the change that created it (`creator_resolver`). + /// + /// Returns `Some(parent)` when the edge is known (`parent` is `None` for a top-level root) + /// and `None` when no source knows the container. + fn resolve_parent(&self, child: ContainerIdx) -> Option> { + let (child_id, resolver, creator_resolver) = { let containers = self.inner.containers.read(); let child_id = containers.container_id(child).unwrap(); if child_id.is_root() && !child_id.is_mergeable() { // TODO: PERF: we can speed this up by use a special bit in ContainerIdx to indicate // whether the target is a root container - return None; + return Some(None); } // Try fast path first if let Some(p) = containers.parents.get(&child).copied() { - return p; + return Some(p); } - // Fallback: try to resolve parent lazily via the resolver if provided. - (child_id, containers.parent_resolver.clone()) + // Fallback: try to resolve parent lazily via the resolvers if provided. + ( + child_id, + containers.parent_resolver.clone(), + containers.creator_resolver.clone(), + ) }; if let Some(resolver) = resolver { if let Some(parent_id) = resolver(child_id.clone()) { let parent_idx = self.register_container(&parent_id); self.set_parent(child, Some(parent_idx)); - return Some(parent_idx); + return Some(Some(parent_idx)); } } - panic!("InternalError: Parent is not registered") + if let (Some(resolver), ContainerID::Normal { peer, counter, .. }) = + (creator_resolver, &child_id) + { + return match resolver(self, ID::new(*peer, *counter)) { + // Registered unless the op does not create this container. + CreatorOp::Loaded => self.get_registered_parent(child), + CreatorOp::Absent => None, + }; + } + + None + } + + /// The index of a normal container that is not registered yet but that an op in the + /// history creates. Loading that op's change registers the container and its parent. + pub(crate) fn find_created_container(&self, id: &ContainerID) -> Option { + if let Some(idx) = self.id_to_idx(id) { + return Some(idx); + } + let ContainerID::Normal { peer, counter, .. } = id else { + return None; + }; + let resolver = self.inner.containers.read().creator_resolver.clone()?; + match resolver(self, ID::new(*peer, *counter)) { + CreatorOp::Loaded => self.id_to_idx(id), + CreatorOp::Absent => None, + } } /// Return the parent edge already stored in the arena without invoking the lazy resolver. @@ -707,6 +818,41 @@ impl SharedArena { // TODO: this can return a u16 directly now, since the depths are always valid pub(crate) fn get_depth(&self, container: ContainerIdx) -> Option { + if let Some(d) = self.inner.containers.read().depth[container.to_index() as usize] { + return Some(d); + } + + // `ArenaContainers::get_depth` runs under the arena lock, where the creator resolver + // cannot run, so resolve the missing parent links first. + let mut c = container; + // Every step goes one level up, and no valid chain is deeper than `u16::MAX`. + let mut steps = 0usize; + loop { + steps += 1; + assert!( + steps <= u16::MAX as usize + 1, + "InternalError: the parent links of {:?} form a cycle", + self.idx_to_id(container) + ); + { + let containers = self.inner.containers.read(); + if containers.depth[c.to_index() as usize].is_some() { + break; + } + match containers.parents.get(&c).copied() { + Some(Some(p)) => { + c = p; + continue; + } + Some(None) => break, + None => {} + } + } + match self.resolve_parent(c) { + Some(Some(p)) => c = p, + _ => break, + } + } self.inner.containers.write().get_depth(container) } @@ -784,6 +930,14 @@ impl SharedArena { self.inner.containers.write().parent_resolver = resolver.map(|f| Arc::new(f) as Arc); } + + /// Register the op log's [`CreatorResolver`]. See `creator_resolver` in `ArenaContainers`. + pub(crate) fn set_creator_resolver(&self, resolver: F) + where + F: Fn(&SharedArena, ID) -> CreatorOp + Send + Sync + 'static, + { + self.inner.containers.write().creator_resolver = Some(Arc::new(resolver)); + } } #[cfg(test)] @@ -851,6 +1005,67 @@ mod tests { ); } + /// The creator resolver parses changes, which registers containers in the arena, so + /// `get_parent` and `get_depth` must call it without holding the arena lock. + #[test] + fn creator_resolver_runs_outside_the_arena_lock() { + let arena = SharedArena::new(); + let root = ContainerID::new_root("tree", ContainerType::Tree); + let meta = ContainerID::new_normal(ID::new(1, 3), ContainerType::Map); + let meta_for_resolver = meta.clone(); + arena.set_creator_resolver(move |arena: &SharedArena, id: ID| { + assert_eq!(id, ID::new(1, 3)); + // What parsing the change does: register the container and its parent. + let parent = arena.register_container(&root); + let child = arena.register_container(&meta_for_resolver); + arena.set_parent(child, Some(parent)); + CreatorOp::Loaded + }); + + let meta_idx = arena.register_container(&meta); + assert_eq!(arena.get_depth(meta_idx).map(|d| d.get()), Some(2)); + let root_idx = arena.id_to_idx(&ContainerID::new_root("tree", ContainerType::Tree)); + assert_eq!(arena.get_parent(meta_idx), root_idx); + } + + #[test] + fn a_container_that_no_op_creates_has_no_parent() { + for answer in [CreatorOp::Loaded, CreatorOp::Absent] { + // `Loaded`: the op exists but creates something else. + let arena = SharedArena::new(); + arena.set_creator_resolver(move |_: &SharedArena, _: ID| answer); + let id = ContainerID::new_normal(ID::new(1, 3), ContainerType::Map); + let idx = arena.register_container(&id); + assert_eq!(arena.get_parent(idx), None); + assert_eq!(arena.get_depth(idx), None); + let missing = ContainerID::new_normal(ID::new(1, 4), ContainerType::Text); + assert_eq!(arena.find_created_container(&missing), None); + } + } + + #[test] + #[should_panic(expected = "form a cycle")] + fn a_parent_cycle_fails_fast() { + let arena = SharedArena::new(); + let a = + arena.register_container(&ContainerID::new_normal(ID::new(1, 0), ContainerType::Map)); + let b = + arena.register_container(&ContainerID::new_normal(ID::new(1, 1), ContainerType::Map)); + arena.set_parent(a, Some(b)); + arena.set_parent(b, Some(a)); + arena.get_depth(a); + } + + /// Without an op log to ask, an unknown parent is still an internal error. + #[test] + #[should_panic(expected = "Parent is not registered")] + fn unknown_parent_without_a_creator_resolver_panics() { + let arena = SharedArena::new(); + let idx = + arena.register_container(&ContainerID::new_normal(ID::new(1, 3), ContainerType::Map)); + arena.get_parent(idx); + } + #[test] fn set_parent_does_not_resolve_missing_parent_depth() { let arena = SharedArena::new(); diff --git a/crates/loro-internal/src/arena/str_arena.rs b/crates/loro-internal/src/arena/str_arena.rs index db7de2a4b..30f6c97ce 100644 --- a/crates/loro-internal/src/arena/str_arena.rs +++ b/crates/loro-internal/src/arena/str_arena.rs @@ -26,7 +26,17 @@ struct Index { unicode: u32, } +impl StrArenaCheckpoint { + pub(super) fn bytes_len(&self) -> usize { + self.bytes_len + } +} + impl StrArena { + pub(super) fn bytes_len(&self) -> usize { + self.bytes.len() + } + #[inline] pub fn is_empty(&self) -> bool { self.len.bytes == 0 diff --git a/crates/loro-internal/src/encoding/fast_snapshot.rs b/crates/loro-internal/src/encoding/fast_snapshot.rs index 18bdc4f47..727770ab6 100644 --- a/crates/loro-internal/src/encoding/fast_snapshot.rs +++ b/crates/loro-internal/src/encoding/fast_snapshot.rs @@ -251,7 +251,7 @@ pub(crate) fn decode_snapshot_inner( if let Err(e) = decode_result { state.reset_to_empty_for_failed_snapshot_import(); - oplog.reset_to_empty_for_failed_snapshot_import(arena_checkpoint); + oplog.reset_to_empty_for_failed_snapshot_import(arena_checkpoint, &state); return Err(e); } drop(state); @@ -260,12 +260,10 @@ pub(crate) fn decode_snapshot_inner( doc.set_detached(true); if let Err(e) = doc._checkout_to_latest_without_commit_as_import(false, checkout_origin) { doc.set_detached(false); - doc.app_state() - .lock() - .reset_to_empty_for_failed_snapshot_import(); - doc.oplog() - .lock() - .reset_to_empty_for_failed_snapshot_import(arena_checkpoint); + let mut oplog = doc.oplog().lock(); + let mut state = doc.app_state().lock(); + state.reset_to_empty_for_failed_snapshot_import(); + oplog.reset_to_empty_for_failed_snapshot_import(arena_checkpoint, &state); return Err(e); } debug_assert_eq!(doc.state_frontiers(), doc.oplog_frontiers()); diff --git a/crates/loro-internal/src/loro.rs b/crates/loro-internal/src/loro.rs index 9b4eeee70..350c999d3 100644 --- a/crates/loro-internal/src/loro.rs +++ b/crates/loro-internal/src/loro.rs @@ -723,13 +723,13 @@ impl LoroDoc { }, diff_mode, ) { - oplog.end_import_rollback(owns_rollback, false); + oplog.rollback_owned_import(owns_rollback, &state); return Err(e); } } match result { Ok(result) => { - oplog.end_import_rollback(owns_rollback, true); + oplog.commit_owned_import_rollback(owns_rollback); Ok(result) } Err(e) => { @@ -738,18 +738,22 @@ impl LoroDoc { // also applied to state; rollback is for failed state apply // or decode errors that made no visible oplog progress. let keep_prefix = &old_vv != oplog.vv(); - oplog.end_import_rollback(owns_rollback, keep_prefix); + if keep_prefix { + oplog.commit_owned_import_rollback(owns_rollback); + } else { + oplog.rollback_owned_import(owns_rollback, &self.state.lock()); + } Err(e) } } } else { match f(&mut oplog) { Ok(result) => { - oplog.end_import_rollback(owns_rollback, true); + oplog.commit_owned_import_rollback(owns_rollback); Ok(result) } Err(e) => { - oplog.end_import_rollback(owns_rollback, false); + oplog.rollback_owned_import(owns_rollback, &self.state.lock()); Err(e) } } @@ -767,7 +771,7 @@ impl LoroDoc { let changes = match decode_changes(&mut oplog) { Ok(changes) => changes, Err(e) => { - oplog.arena.rollback(arena_checkpoint); + oplog.rollback_arena(arena_checkpoint, &self.state.lock()); return Err(e); } }; @@ -776,7 +780,7 @@ impl LoroDoc { if preflight.has_deps_before_shallow_root && (self.is_detached() || !preflight.applies_to_dag) { - oplog.arena.rollback(arena_checkpoint); + oplog.rollback_arena(arena_checkpoint, &self.state.lock()); return Err(LoroError::ImportUpdatesThatDependsOnOutdatedVersion); } @@ -791,7 +795,7 @@ impl LoroDoc { 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(); + oplog.rollback_import(&self.state.lock()); return Err(e); } oplog.commit_import_rollback(); @@ -807,7 +811,7 @@ impl LoroDoc { let pending_root_containers = pending_root_containers_to_materialize(&oplog, &changes); let result = encoding::apply_decoded_changes_to_oplog(&mut oplog, changes); if result.has_deps_before_shallow_root { - oplog.arena.rollback(arena_checkpoint); + oplog.rollback_arena(arena_checkpoint, &self.state.lock()); return Err(LoroError::ImportUpdatesThatDependsOnOutdatedVersion); } @@ -842,7 +846,7 @@ impl LoroDoc { // 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(); + oplog.rollback_import(&self.state.lock()); return Err(e); } } @@ -900,7 +904,7 @@ impl LoroDoc { diff_mode, ) { if rollback_enabled { - oplog.rollback_import(); + oplog.rollback_import(&state); return Err(e); } @@ -2733,7 +2737,9 @@ impl LoroDoc { state.ensure_container(id); } } - let idx = state.arena.id_to_idx(id).unwrap(); + // The registration above can already be gone: a failed import on another thread + // rolls back every registration made since it began (loro-dev/loro#1164). + let idx = state.arena.id_to_idx(id)?; state.get_path(idx) } @@ -3210,7 +3216,7 @@ impl BatchImportGuard<'_> { // still at its pre-batch version; undoing the batch in the `OpLog` // makes the two agree again, which is what lets the doc stay attached. tracing::warn!("import_batch cannot reattach, rolling the batch back: {e}"); - doc.oplog.lock().rollback_import(); + doc.oplog.lock().rollback_import(&doc.state.lock()); // The shared diff calculator cached ranges against the rolled-back // history; drop that cache instead of reusing stale entries. *doc.diff_calculator.lock() = DiffCalculator::new(true); diff --git a/crates/loro-internal/src/oplog.rs b/crates/loro-internal/src/oplog.rs index 4d5bcd250..dacb7f605 100644 --- a/crates/loro-internal/src/oplog.rs +++ b/crates/loro-internal/src/oplog.rs @@ -26,6 +26,7 @@ use crate::history_cache::ContainerHistoryCache; use crate::id::{Counter, PeerID, ID}; use crate::op::{FutureInnerContent, InnerContent, ListSlice, RawOpContent, RemoteOp, RichOp}; use crate::span::{HasCounterSpan, HasLamportSpan}; +use crate::state::DocState; use crate::version::{Frontiers, ImVersionVector, VersionVector}; use crate::LoroError; use change_store::{BlockOpRef, ChangeStoreRollback}; @@ -151,6 +152,7 @@ impl OpLog { let arena = SharedArena::new(); let cfg = Configure::default(); let change_store = ChangeStore::new_mem(&arena, cfg.merge_interval_in_s.clone()); + arena.set_creator_resolver(change_store.creator_resolver()); Self { visible_op_count, history_cache: Mutex::new(ContainerHistoryCache::new(change_store.clone(), None)), @@ -266,6 +268,10 @@ impl OpLog { self.history_cache .lock() .insert_by_new_change(&change, true, true); + #[cfg(debug_assertions)] + if from_local { + crate::parent::assert_local_parent_links_registered(&self.arena, &change); + } self.register_container_and_parent_link(&change); if let Some(rollback) = self.import_rollback.as_mut() { for op in change.ops.iter() { @@ -328,18 +334,20 @@ impl OpLog { self.import_rollback = None; } - /// Close an import rollback scope this caller owns: commit it when `keep`, - /// roll everything back otherwise. No-op when `owns` is false — the scope - /// then belongs to an outer owner such as `import_batch`. - pub(crate) fn end_import_rollback(&mut self, owns: bool, keep: bool) { - if !owns { - return; + /// Commit an import rollback scope this caller owns. No-op when `owns` is false — the + /// scope then belongs to an outer owner such as `import_batch`. + pub(crate) fn commit_owned_import_rollback(&mut self, owns: bool) { + if owns { + self.commit_import_rollback(); } + } - if keep { - self.commit_import_rollback(); - } else { - self.rollback_import(); + /// Roll back an import rollback scope this caller owns (see [`Self::rollback_import`]). + /// No-op when `owns` is false — the scope then belongs to an outer owner such as + /// `import_batch`. + pub(crate) fn rollback_owned_import(&mut self, owns: bool, state: &DocState) { + if owns { + self.rollback_import(state); } } @@ -402,27 +410,45 @@ impl OpLog { ans } - pub(crate) fn rollback_import(&mut self) { + /// Undo the open import rollback scope, including the arena. + /// + /// `_state` shows that the caller holds the state lock. A reader under that lock can + /// register containers through the creator resolver, which runs without the op log lock, + /// and use their indices before it returns; the arena rollback must not free them in + /// between. See `context/arena-parent-links.md`. + pub(crate) fn rollback_import(&mut self, _state: &DocState) { let Some(rollback) = self.import_rollback.take() else { return; }; - self.change_store.rollback_import(rollback.change_store); + // Also rolls back the arena; see `ChangeStore::rollback_import`. + self.change_store + .rollback_import(rollback.change_store, rollback.arena); self.dag.rollback_import(); rollback.pending.rollback(&mut self.pending_changes); self.history_cache.lock().free_all(); - self.arena.rollback(rollback.arena); self.refresh_visible_op_count(); } + /// Rolls the arena back to a checkpoint taken before an import that failed before its + /// import rollback scope began. See [`ChangeStore::rollback_arena`], and + /// [`Self::rollback_import`] for `_state`. + pub(crate) fn rollback_arena(&self, arena_checkpoint: SharedArenaRollback, _state: &DocState) { + self.change_store.rollback_arena(arena_checkpoint); + } + + /// See [`Self::rollback_import`] for `_state`. pub(crate) fn reset_to_empty_for_failed_snapshot_import( &mut self, arena_checkpoint: SharedArenaRollback, + _state: &DocState, ) { let arena = self.arena.clone(); let configure = self.configure.clone(); - arena.rollback(arena_checkpoint); + // Also rolls back the arena; see `ChangeStore::retire`. + self.change_store.retire(arena_checkpoint); let change_store = ChangeStore::new_mem(&arena, configure.merge_interval_in_s.clone()); + arena.set_creator_resolver(change_store.creator_resolver()); self.history_cache = Mutex::new(ContainerHistoryCache::new(change_store.clone(), None)); self.dag = AppDag::new(change_store.clone()); self.change_store = change_store; diff --git a/crates/loro-internal/src/oplog/change_store.rs b/crates/loro-internal/src/oplog/change_store.rs index c45d347f6..7edd054a7 100644 --- a/crates/loro-internal/src/oplog/change_store.rs +++ b/crates/loro-internal/src/oplog/change_store.rs @@ -4,7 +4,7 @@ use self::block_encode::{ use super::{loro_dag::AppDagNodeInner, AppDagNode}; use crate::sync::Mutex; use crate::{ - arena::SharedArena, + arena::{ArenaExtent, CreatorOp, SharedArena, SharedArenaRollback}, change::Change, estimated_size::EstimatedSize, kv_store::KvStore, @@ -50,6 +50,15 @@ const MAX_ROOT_HISTORY_NAME_BYTES: usize = 256 * 1024; /// The [Change] should be continuous for each peer. /// - However, the first block of a peer can have counter > 0 so that we can trim the history. /// +/// # Locking +/// +/// Take the locks in this order, and release them before calling out: `root_history_names`, +/// `external_kv`, `inner`, `external_vv`. Parsing a block takes the arena's lock under them. +/// Most callers hold the document's op log lock, which serializes them, but the arena's +/// creator resolver ([`ChangeStore::creator_resolver`]) does not, so a method that takes two +/// of these locks in another order can deadlock against it. See +/// `context/arena-parent-links.md`. +/// /// # Encoding Schema /// /// It's based on the underlying KV store. @@ -128,6 +137,9 @@ struct ChangeStoreInner { start_frontiers: Frontiers, /// It's more like a parsed cache for binary_kv. mem_parsed_kv: BTreeMap>, + /// Set by [`ChangeStore::retire`]: the op log replaced this store, so loading from it + /// must not register anything in the arena any more. + retired: bool, } #[derive(Debug)] @@ -203,6 +215,9 @@ pub(crate) struct ChangesBlock { estimated_size: usize, flushed: bool, content: ChangesBlockContent, + /// Where the arena reached when `content` was parsed from its bytes (`Both`). See + /// [`ChangeStore::rollback_arena`]. + parsed_extent: ArenaExtent, } #[derive(Clone)] @@ -231,6 +246,7 @@ impl ChangeStore { start_vv: ImVersionVector::new(), start_frontiers: Frontiers::default(), mem_parsed_kv: BTreeMap::new(), + retired: false, })), arena: a.clone(), external_vv: Arc::new(Mutex::new(VersionVector::new())), @@ -391,7 +407,13 @@ impl ChangeStore { Ok(ans) } - pub(crate) fn rollback_import(&self, rollback: ChangeStoreRollback) { + /// Rolls back the store and the arena (to `arena`, the checkpoint taken when the import + /// began). See [`Self::rollback_arena`]. + pub(crate) fn rollback_import( + &self, + rollback: ChangeStoreRollback, + arena: SharedArenaRollback, + ) { // 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(); @@ -447,6 +469,52 @@ impl ChangeStore { .mem_parsed_kv .retain(|id, block| !block.flushed || !touched_peers.contains(&id.peer)); } + + self.rollback_arena_in(&mut inner, arena); + } + + /// Rolls the arena back to `arena`, a checkpoint taken before a failed import, and drops + /// the parsed changes of the cached blocks that were parsed since. Every arena rollback must + /// go through here (or [`Self::rollback_import`] / [`Self::retire`]). + /// + /// Parsing a block registers the containers its ops use and allocates their values, and + /// the arena rollback drops what was registered or allocated after the checkpoint, so a + /// block parsed in between may hold indices and value slices that no longer exist or that + /// new registrations reuse. Such a block keeps only its bytes, so the next access parses and + /// registers again. A block parsed before the checkpoint can only refer to what was there + /// then (its `parsed_extent`), and keeps its parsed changes. A block without bytes was built + /// in memory from changes inserted before the import, whose containers were registered + /// then. + /// + /// This happens under `inner`, where blocks are parsed. The creator resolver parses + /// without the op log lock, so otherwise it could parse a block after the bytes are + /// restored and before the arena is rolled back. + pub(crate) fn rollback_arena(&self, arena: SharedArenaRollback) { + let mut inner = self.inner.lock(); + self.rollback_arena_in(&mut inner, arena); + } + + fn rollback_arena_in(&self, inner: &mut ChangeStoreInner, arena: SharedArenaRollback) { + for block in inner.mem_parsed_kv.values_mut() { + if let ChangesBlockContent::Both(_, bytes) = &block.content { + if !arena.keeps(block.parsed_extent) { + let bytes = bytes.clone(); + Arc::make_mut(block).content = ChangesBlockContent::Bytes(bytes); + } + } + } + self.arena.rollback(arena); + } + + /// Rolls the arena back to `arena` after the op log replaced this store (a failed + /// snapshot import), and stops loading from it. A creator resolver that reached this store + /// before the replacement then finds nothing instead of registering containers of the + /// discarded history in the arena. + pub(crate) fn retire(&self, arena: SharedArenaRollback) { + let mut inner = self.inner.lock(); + self.arena.rollback(arena); + inner.mem_parsed_kv.clear(); + inner.retired = true; } pub fn get_dag_nodes_that_contains(&self, id: ID) -> Option> { @@ -774,16 +842,18 @@ impl ChangeStore { frontiers: &Frontiers, ) -> Self { self.flush_and_compact(vv, frontiers); + let external_kv = self.external_kv.lock().clone_store(); let inner = self.inner.lock(); Self { inner: Arc::new(Mutex::new(ChangeStoreInner { start_vv: inner.start_vv.clone(), start_frontiers: inner.start_frontiers.clone(), mem_parsed_kv: BTreeMap::new(), + retired: false, })), arena, external_vv: Arc::new(Mutex::new(self.external_vv.lock().clone())), - external_kv: self.external_kv.lock().clone_store(), + external_kv, merge_interval, root_history_names: Arc::new(Mutex::new(RootHistoryNamesState::Uninitialized)), #[cfg(test)] @@ -987,8 +1057,8 @@ mod mut_external_kv { /// Flush the cached change to kv_store pub(crate) fn flush_and_compact(&self, vv: &VersionVector, frontiers: &Frontiers) { - let mut inner = self.inner.lock(); let mut store = self.external_kv.lock(); + let mut inner = self.inner.lock(); let mut external_vv = self.external_vv.lock(); for (id, block) in inner.mem_parsed_kv.iter_mut() { if !block.flushed { @@ -1132,6 +1202,8 @@ mod mut_inner_kv { pub fn get_change_by_lamport_lte(&self, idlp: IdLp) -> Option { // This method is complicated because we impl binary search on top of the range api // It can be simplified + // Lock order: `external_kv` before `inner`. The scan below may need it. + let kv_store = self.external_kv.lock(); let mut inner = self.inner.lock(); let mut iter = inner .mem_parsed_kv @@ -1262,7 +1334,6 @@ mod mut_inner_kv { .map(|(id, _)| *id); let external_block = 'block_scan: { - let kv_store = &self.external_kv.lock(); let scan_end = ID::new(idlp.peer, counter_end).to_bytes(); let iter = kv_store .scan( @@ -1432,56 +1503,115 @@ mod mut_inner_kv { ans } - fn get_parsed_block(&self, id: ID) -> Option> { - let mut inner = self.inner.lock(); - if let Some((_id, block)) = inner.mem_parsed_kv.range_mut(..=id).next_back() { - if block.peer == id.peer && block.counter_range.1 > id.counter { - if let Err(err) = block.ensure_changes(&self.arena) { - warn!(block_id = ?_id, ?err, "failed to parse cached change block"); - return None; - } - return Some(block.clone()); + /// A callback for [`SharedArena::set_creator_resolver`]: loads and parses the change + /// block holding the op with the given ID, if the store has it. Parsing registers the + /// parent link of every container the block's ops create. + /// + /// It holds weak references, so the arena (which this store refers to) does not keep + /// the store alive. + /// + /// Locking: this is the only access to the store that does not hold the document's op + /// log lock (it runs under the state lock, or under none while events are emitted), so + /// it relies on the store's lock order (see [`ChangeStore`]). It takes the arena's lock + /// after the store's, so the arena must call it without holding its own. See + /// `context/arena-parent-links.md`. + pub(crate) fn creator_resolver( + &self, + ) -> impl Fn(&SharedArena, ID) -> CreatorOp + Send + Sync + 'static { + let inner = Arc::downgrade(&self.inner); + let external_kv = Arc::downgrade(&self.external_kv); + move |arena, id| { + let (Some(inner), Some(external_kv)) = (inner.upgrade(), external_kv.upgrade()) + else { + // The op log is gone, and its history with it. + return CreatorOp::Absent; + }; + match Self::load_parsed_block(&inner, &external_kv, arena, id) { + Ok(Some(_)) => CreatorOp::Loaded, + Ok(None) => CreatorOp::Absent, + // Answering "no such op" would report the container as deleted. + Err((block_id, err)) => panic!( + "InternalError: cannot parse change block {block_id}, which holds {id}: \ + {err}" + ), } } + } - let store = self.external_kv.lock(); - let mut iter = store - .scan(Bound::Unbounded, Bound::Included(&id.to_bytes())) - .filter(|(id, _)| id.len() == 12); - - // println!( - // "\nkeys {:?}", - // store - // .scan(Bound::Unbounded, Bound::Included(&id.to_bytes())) - // .filter(|(id, _)| id.len() == 12) - // .map(|(k, _v)| ID::from_bytes(&k)) - // .count() - // ); - // println!("id {:?}", id); - - let (b_id, b_bytes) = iter.next_back()?; - let block_id: ID = ID::from_bytes(&b_id[..]); - let block = match ChangesBlock::from_bytes(b_bytes) { + fn get_parsed_block(&self, id: ID) -> Option> { + match Self::load_parsed_block(&self.inner, &self.external_kv, &self.arena, id) { Ok(block) => block, - Err(err) => { - warn!(?block_id, ?err, "failed to decode external change block"); - return None; + Err((block_id, err)) => { + warn!(?block_id, ?err, "failed to parse change block"); + None } - }; - if block_id.peer == id.peer - && block_id.counter <= id.counter - && block.counter_range.1 > id.counter + } + } + + /// The parsed block holding `id`, `Ok(None)` if the store has no such block, or the ID + /// of a block of `id`'s peer that cannot be decoded or parsed. + fn load_parsed_block( + inner: &Mutex, + external_kv: &Mutex, + arena: &SharedArena, + id: ID, + ) -> Result>, (ID, LoroError)> { + // A cached block needs only `inner`. { - let mut arc_block = Arc::new(block); - if let Err(err) = arc_block.ensure_changes(&self.arena) { - warn!(?block_id, ?err, "failed to parse external change block"); - return None; + let mut inner = inner.lock(); + if inner.retired { + return Ok(None); + } + if let Some(block) = Self::parse_cached_block(&mut inner, arena, id) { + return block.map(Some); } - inner.mem_parsed_kv.insert(block_id, arc_block.clone()); - return Some(arc_block); } - None + // Lock order: `external_kv` before `inner`. Another thread may have loaded the + // block (or retired the store) in between, so look again. + let store = external_kv.lock(); + let mut inner = inner.lock(); + if inner.retired { + return Ok(None); + } + if let Some(block) = Self::parse_cached_block(&mut inner, arena, id) { + return block.map(Some); + } + + let Some((b_id, b_bytes)) = store + .scan(Bound::Unbounded, Bound::Included(&id.to_bytes())) + .rfind(|(id, _)| id.len() == 12) + else { + return Ok(None); + }; + let block_id: ID = ID::from_bytes(&b_id[..]); + if block_id.peer != id.peer { + return Ok(None); + } + let block = ChangesBlock::from_bytes(b_bytes).map_err(|err| (block_id, err))?; + if block.counter_range.1 <= id.counter { + return Ok(None); + } + let mut block = Arc::new(block); + block.ensure_changes(arena).map_err(|err| (block_id, err))?; + inner.mem_parsed_kv.insert(block_id, block.clone()); + Ok(Some(block)) + } + + /// `Some` if a cached block holds `id`: the block, parsed, or why it cannot be parsed. + fn parse_cached_block( + inner: &mut ChangeStoreInner, + arena: &SharedArena, + id: ID, + ) -> Option, (ID, LoroError)>> { + let (block_id, block) = inner.mem_parsed_kv.range_mut(..=id).next_back()?; + if block.peer != id.peer || block.counter_range.1 <= id.counter { + return None; + } + Some(match block.ensure_changes(arena) { + Ok(()) => Ok(block.clone()), + Err(err) => Err((*block_id, err)), + }) } /// Load all the blocks that have overlapped with the given ID range into `inner_mem_parsed_kv` @@ -1675,6 +1805,7 @@ impl ChangesBlock { lamport_range, flushed: true, content, + parsed_extent: ArenaExtent::default(), }) } @@ -1697,6 +1828,7 @@ impl ChangesBlock { estimated_size, content, flushed: false, + parsed_extent: ArenaExtent::default(), } } @@ -1808,6 +1940,7 @@ impl ChangesBlock { let b = bytes.clone(); let this = Arc::make_mut(self); this.content = ChangesBlockContent::Both(Arc::new(changes), b); + this.parsed_extent = a.extent(); Ok(()) } } @@ -1850,11 +1983,6 @@ impl ChangesBlock { } } - #[allow(unused)] - fn get_changes(&mut self, a: &SharedArena) -> LoroResult<&Vec> { - self.content.changes(a) - } - #[allow(unused)] fn id(&self) -> ID { ID::new(self.peer, self.counter_range.0) @@ -1914,19 +2042,6 @@ impl ChangesBlockContent { dag_nodes } - #[allow(unused)] - pub fn changes(&mut self, a: &SharedArena) -> LoroResult<&Vec> { - match self { - ChangesBlockContent::Changes(changes) => Ok(changes), - ChangesBlockContent::Both(changes, _) => Ok(changes), - ChangesBlockContent::Bytes(bytes) => { - let changes = bytes.parse(a)?; - *self = ChangesBlockContent::Both(Arc::new(changes), bytes.clone()); - self.changes(a) - } - } - } - /// Note that this method will invalidate the stored bytes fn changes_mut(&mut self, a: &SharedArena) -> LoroResult<&mut Arc>> { match self { @@ -2164,6 +2279,47 @@ mod test { } } + #[test] + fn rollback_drops_only_blocks_parsed_after_the_checkpoint() { + // A failed import must not keep blocks it parsed (they may refer to what the arena + // rollback drops), but the blocks parsed before it are still valid, and reparsing them + // after every failed import would cost a full history pass. + let (store, end, _) = kv_only_store_and_next_change(); + let is_parsed = |id: ID| { + let inner = store.inner.lock(); + let (_, block) = inner.mem_parsed_kv.range(..=id).next_back().unwrap(); + matches!(block.content, ChangesBlockContent::Both(..)) + }; + let early = ID::new(1, 0); + let late = ID::new(1, end - 1); + store.get_change(early).unwrap(); + let checkpoint = store.arena.checkpoint_for_rollback(); + // Something the failed import registered. + store.arena.register_container(&ContainerID::new_root( + "new", + loro_common::ContainerType::Map, + )); + store.get_change(late).unwrap(); + assert!(is_parsed(early) && is_parsed(late)); + store.rollback_arena(checkpoint); + assert!(is_parsed(early)); + assert!(!is_parsed(late)); + assert!(store.get_change(late).is_some()); + } + + #[test] + fn a_retired_store_resolves_nothing() { + // A resolver that reached the store before a failed snapshot import replaced it + // must not register containers of the discarded history. + let (store, _, _) = kv_only_store_and_next_change(); + let resolve = store.creator_resolver(); + let checkpoint = store.arena.checkpoint_for_rollback(); + assert_eq!(resolve(&store.arena, ID::new(1, 0)), CreatorOp::Loaded); + store.retire(checkpoint); + assert_eq!(resolve(&store.arena, ID::new(1, 0)), CreatorOp::Absent); + assert!(store.inner.lock().mem_parsed_kv.is_empty()); + } + #[test] fn rollback_evicts_older_blocks_cached_during_the_scope() { // An import creates the peer's newest block without loading the older ones, @@ -2174,9 +2330,10 @@ mod test { let mut old_vv = VersionVector::new(); old_vv.insert(1, end); let mut rollback = ChangeStoreRollback::new(old_vv); + let arena = store.arena.checkpoint_for_rollback(); 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.rollback_import(rollback, arena); 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()); @@ -2223,10 +2380,11 @@ mod test { rolled_back.insert_change(c.clone(), true, false); } let mut rollback = ChangeStoreRollback::new(vv_of(first)); + let arena_checkpoint = arena.checkpoint_for_rollback(); for c in &rest[..rest.len() / 2] { rolled_back.insert_change_with_rollback(c.clone(), true, false, &mut rollback); } - rolled_back.rollback_import(rollback); + rolled_back.rollback_import(rollback, arena_checkpoint); assert!(rolled_back .get_change(ID::new(1, vv_of(first)[&1])) .is_none()); diff --git a/crates/loro-internal/src/parent.rs b/crates/loro-internal/src/parent.rs index 031c5ac48..aec7ccad6 100644 --- a/crates/loro-internal/src/parent.rs +++ b/crates/loro-internal/src/parent.rs @@ -35,6 +35,27 @@ pub(super) fn register_container_and_parent_link(arena: &SharedArena, change: &C } } +/// A transaction registers the parent of every container its ops create when it applies them +/// ([`DocState::set_container_parent_by_raw_op`]), before the change is committed. Until then +/// the op log cannot tell the parent, so a container a local op path forgot to link reads as +/// deleted instead of failing (see `SharedArena::get_parent`). Check it on commit. +#[cfg(debug_assertions)] +pub(super) fn assert_local_parent_links_registered(arena: &SharedArena, change: &Change) { + for op in change.ops.iter() { + op.content.visit_created_children(arena, &mut |c| { + let parent = arena + .id_to_idx(c) + .and_then(|idx| arena.get_registered_parent(idx)); + assert_eq!( + parent, + Some(Some(op.container)), + "InternalError: local op {} created {c} without registering its parent", + loro_common::ID::new(change.id.peer, op.counter) + ); + }); + } +} + impl DocState { /// This is used in txn to short cut the process of applying an op to the state. /// So the op here has not been registered on the oplog yet. @@ -108,3 +129,23 @@ impl DocState { } } } + +#[cfg(test)] +mod test { + use loro_common::ID; + + use crate::{arena::SharedArena, LoroDoc, TreeParentId}; + + #[test] + #[cfg(debug_assertions)] + #[should_panic(expected = "without registering its parent")] + fn a_local_op_must_link_the_containers_it_creates() { + let doc = LoroDoc::new_auto_commit(); + doc.set_peer_id(1).unwrap(); + doc.get_tree("tree").create(TreeParentId::Root).unwrap(); + doc.commit_then_renew(); + let change = doc.oplog().lock().get_change_at(ID::new(1, 0)).unwrap(); + // An arena that the transaction did not register the node's meta in. + super::assert_local_parent_links_registered(&SharedArena::new(), &change); + } +} diff --git a/crates/loro-internal/src/state.rs b/crates/loro-internal/src/state.rs index 5068d729b..d483858d1 100644 --- a/crates/loro-internal/src/state.rs +++ b/crates/loro-internal/src/state.rs @@ -1060,11 +1060,14 @@ impl DocState { return true; } - if !is_mergeable { - if let Some(idx) = self.arena.id_to_idx(id) { - if self.arena.get_depth(idx).is_some() { - return true; - } + let registered = if is_mergeable { + None + } else { + self.arena.id_to_idx(id) + }; + if let Some(idx) = registered { + if self.arena.get_depth(idx).is_some() { + return true; } } @@ -1072,6 +1075,18 @@ impl DocState { return true; } + // Decided by the first lookup: another thread (e.g. creating a handler) may register + // the container in between, without its parent. + if !is_mergeable && registered.is_none() { + // A document loaded from a snapshot parses its changes lazily, so a container that + // an op created may not be registered yet. + if let Some(idx) = self.arena.find_created_container(id) { + if self.arena.get_depth(idx).is_some() { + return true; + } + } + } + // An ensured-but-empty mergeable child has no ops or KV state of its own yet; // the parent map's binary child ref is the source of truth for its existence. is_mergeable && self.get_reachable(id) diff --git a/crates/loro-internal/src/state/container_store/container_wrapper.rs b/crates/loro-internal/src/state/container_store/container_wrapper.rs index 81f875851..3abfcb4f8 100644 --- a/crates/loro-internal/src/state/container_store/container_wrapper.rs +++ b/crates/loro-internal/src/state/container_store/container_wrapper.rs @@ -148,7 +148,10 @@ impl ContainerWrapper { let parent = arena .get_parent(idx) .and_then(|p| arena.get_container_id(p)); - let depth = arena.get_depth(idx).unwrap().get() as usize; + let depth = arena + .get_depth(idx) + .expect("InternalError: a container gets state only once its parent is known") + .get() as usize; Self { depth, parent, diff --git a/crates/loro-wasm/tests/unregistered_container_parent.test.ts b/crates/loro-wasm/tests/unregistered_container_parent.test.ts new file mode 100644 index 000000000..922ea3ab3 --- /dev/null +++ b/crates/loro-wasm/tests/unregistered_container_parent.test.ts @@ -0,0 +1,162 @@ +import { readFileSync } from "node:fs"; +import { describe, expect, it } from "vitest"; +import { LoroDoc, LoroMap, TreeID, UndoManager } from "../bundler/index"; + +// loro-dev/loro#1158: in a document loaded from a snapshot (or `fork()`), the +// metadata of a tree node created under an already deleted parent had no known +// parent, so touching it trapped with `RuntimeError: unreachable`. +// See context/arena-parent-links.md. + +/** Peer 2 creates `child` under `parent` while peer 1 deletes `parent`. */ +function history(): { src: LoroDoc; child: TreeID } { + const a = new LoroDoc(); + a.setPeerId(1n); + const tree = a.getTree("tree"); + const parent = tree.createNode(); + a.commit(); + const b = new LoroDoc(); + b.setPeerId(2n); + b.import(a.export({ mode: "update" })); + const child = b.getTree("tree").createNode(parent.id).id; + b.commit(); + tree.delete(parent.id); + a.commit(); + a.import(b.export({ mode: "update" })); + // Later history, so the child's creation is in an older change block. + for (let i = 0; i < 5000; i++) { + a.getMap("m").set(`k${i % 50}`, i); + a.commit(); + } + return { src: a, child }; +} + +function loaded(src: LoroDoc): [string, LoroDoc][] { + const load = (bytes: Uint8Array) => { + const doc = new LoroDoc(); + doc.import(bytes); + doc.setPeerId(9n); + return doc; + }; + const fork = src.fork(); + fork.setPeerId(9n); + return [ + ["updates", load(src.export({ mode: "update" }))], + ["snapshot", load(src.export({ mode: "snapshot" }))], + ["fork", fork], + ]; +} + +const metaId = (node: TreeID) => `cid:${node}:Map`; + +/** An update from another peer that revives `child` and sets `x` in its meta. */ +function revivalAndMetaEdit(src: LoroDoc, child: TreeID): Uint8Array { + const editor = new LoroDoc(); + editor.setPeerId(4n); + editor.import(src.export({ mode: "update" })); + const before = editor.oplogVersion(); + editor.getTree("tree").move(child, undefined); + editor.getTree("tree").getNodeByID(child)!.data.set("x", 1); + editor.commit(); + return editor.export({ mode: "update", from: before }); +} + +describe("metadata of a node created under a deleted parent", () => { + const { src, child } = history(); + + it("is deleted and rejects edits", () => { + for (const [how, doc] of loaded(src)) { + const meta = doc.getTree("tree").getNodeByID(child)!.data; + expect(meta.isDeleted(), how).toBe(true); + expect(() => { + meta.set("k", 1); + doc.commit(); + }, how).toThrow(/deleted/); + expect(doc.getPathToContainer(metaId(child) as any), how).toBeUndefined(); + } + }); + + it("is found by id in a fresh document", () => { + for (const [how, doc] of loaded(src)) { + const meta = doc.getContainerById(metaId(child) as any) as LoroMap; + expect(meta, how).toBeDefined(); + expect(meta.isDeleted(), how).toBe(true); + } + }); + + it("can be edited after reviving the node", () => { + for (const [how, doc] of loaded(src)) { + const tree = doc.getTree("tree"); + const meta = tree.getNodeByID(child)!.data; + expect(meta.isDeleted(), how).toBe(true); + tree.move(child, undefined); + doc.commit(); + expect(meta.isDeleted(), how).toBe(false); + meta.set("k", 1); + doc.commit(); + expect(meta.get("k"), how).toBe(1); + } + }); + + it("emits events for an imported edit", () => { + const update = revivalAndMetaEdit(src, child); + + let expected: string | undefined; + for (const [how, doc] of loaded(src)) { + const seen: string[] = []; + doc.subscribe((e) => { + for (const ev of e.events) + seen.push(`${ev.target} ${JSON.stringify(ev.path)}`); + }); + doc.import(update); + const meta = doc.getTree("tree").getNodeByID(child)!.data; + expect(meta.get("x"), how).toBe(1); + const got = JSON.stringify(seen); + if (expected === undefined) expected = got; + else expect(got, how).toBe(expected); + } + }); + + it("undoes a local edit after importing an edit of the meta", () => { + const update = revivalAndMetaEdit(src, child); + let expected: string | undefined; + for (const [how, doc] of loaded(src)) { + const undo = new UndoManager(doc, { mergeInterval: 0 }); + doc.getMap("m").set("local", 1); + doc.commit(); + doc.import(update); + expect(undo.undo(), how).toBe(true); + expect(undo.redo(), how).toBe(true); + expect(undo.undo(), how).toBe(true); + const meta = doc.getTree("tree").getNodeByID(child)!.data; + expect(meta.get("x"), how).toBe(1); + expect(doc.getMap("m").get("local"), how).toBeUndefined(); + const got = JSON.stringify(doc.toJSON()); + if (expected === undefined) expected = got; + else expect(got, how).toBe(expected); + } + }); + + it("the review harness dump", () => { + const read = (name: string) => + new Uint8Array( + readFileSync(new URL(`../../loro/tests/${name}`, import.meta.url)), + ); + const node = "7@40881" as TreeID; + const fromSnapshot = new LoroDoc(); + fromSnapshot.import(read("unregistered_meta_parent.snapshot.bin")); + const fromUpdates = new LoroDoc(); + fromUpdates.import(read("unregistered_meta_parent.updates.bin")); + for (const [how, doc] of [ + ["updates", fromUpdates], + ["fork", fromSnapshot.fork()], + ["snapshot", fromSnapshot], + ] as [string, LoroDoc][]) { + const meta = doc.getTree("tree").getNodeByID(node)!.data; + expect(meta.isDeleted(), how).toBe(true); + expect(() => { + meta.delete("c0"); + doc.commit(); + }, how).toThrow(/deleted/); + } + }); +}); diff --git a/crates/loro/tests/multi_thread_test.rs b/crates/loro/tests/multi_thread_test.rs index efce0065f..75d8a5661 100644 --- a/crates/loro/tests/multi_thread_test.rs +++ b/crates/loro/tests/multi_thread_test.rs @@ -338,4 +338,138 @@ mod loom_test { assert_eq!(doc.get_text("text").len_utf8(), 2); }); } + + /// A snapshot where peer 2 created node `0@2` under a parent that peer 1 deleted + /// concurrently, then more history. A document loaded from it finds the parent of the + /// node's meta only by parsing the change that created the node (loro-dev/loro#1158). + fn deleted_parent_snapshot() -> (Vec, loro::TreeID) { + let a = LoroDoc::new(); + a.set_peer_id(1).unwrap(); + let parent = a.get_tree("tree").create(loro::TreeParentId::Root).unwrap(); + a.commit(); + let b = LoroDoc::new(); + b.set_peer_id(2).unwrap(); + b.import(&a.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + let child = b.get_tree("tree").create(parent).unwrap(); + b.commit(); + a.get_tree("tree").delete(parent).unwrap(); + a.commit(); + a.import(&b.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + for i in 0..3 { + a.get_map("m").insert("k", i).unwrap(); + a.commit(); + } + (a.export(ExportMode::Snapshot).unwrap(), child) + } + + /// The creator resolver reaches the change store without the op log lock (here under the + /// state lock, from `is_deleted` and `has_container`), while another thread reads the + /// store under the op log lock. Both must take the store's locks in the same order. + #[test] + fn resolving_a_meta_parent_while_another_thread_reads_the_history() { + for reader in 0..3 { + loom::model(move || { + let (snapshot, child) = deleted_parent_snapshot(); + let doc = LoroDoc::new(); + doc.import(&snapshot).unwrap(); + let meta = doc.get_tree("tree").get_meta(child).unwrap(); + let meta_id = loro::ContainerTrait::id(&meta); + let doc2 = doc.clone(); + let h0 = loom::thread::spawn(move || { + if reader == 2 { + assert!(doc.has_container(&meta_id)); + } else { + assert!(loro::ContainerTrait::is_deleted(&meta)); + } + }); + let h1 = loom::thread::spawn(move || { + if reader == 0 { + assert!(doc2.len_changes() > 0); + } else { + doc2.export(ExportMode::all_updates()).unwrap(); + } + }); + h0.join().unwrap(); + h1.join().unwrap(); + }); + } + } + + /// An import that fails before its rollback scope begins (here while decoding JSON) rolls + /// the arena back to where it started. A block that the resolver parsed on another thread + /// in the meantime must not keep indices that the rollback drops. + #[test] + fn resolving_a_meta_parent_while_an_import_fails() { + struct Fixture { + snapshot: Vec, + meta: loro::ContainerID, + bad_json: String, + vv: loro::VersionVector, + history: String, + } + // Built once, in a single-threaded model of its own, so the racing model does not + // explore it. + static FIXTURE: std::sync::Mutex>> = + std::sync::Mutex::new(None); + let mut builder = loom::model::Builder::new(); + builder.max_branches = 100_000; + builder.check(|| { + let (snapshot, child) = deleted_parent_snapshot(); + // A separate document, so that exporting parses nothing in the tested one. + let reference = LoroDoc::new(); + reference.import(&snapshot).unwrap(); + let vv = reference.oplog_vv(); + let history = + serde_json::to_string(&reference.export_json_updates(&Default::default(), &vv)) + .unwrap(); + // Another peer creates a container, so decoding its change registers one. The + // empty change after it fails the decode. + let e = LoroDoc::new(); + e.set_peer_id(9).unwrap(); + e.import(&snapshot).unwrap(); + e.get_map("x") + .insert_container("c", loro::LoroMap::new()) + .unwrap() + .insert("q", 1) + .unwrap(); + e.commit(); + let mut json = serde_json::to_value(e.export_json_updates(&vv, &e.oplog_vv())).unwrap(); + let changes = json["changes"].as_array_mut().unwrap(); + let mut empty = changes.last().unwrap().clone(); + empty["ops"] = serde_json::json!([]); + changes.push(empty); + *FIXTURE.lock().unwrap() = Some(std::sync::Arc::new(Fixture { + snapshot, + meta: child.associated_meta_container(), + bad_json: serde_json::to_string(&json).unwrap(), + vv, + history, + })); + }); + let f = FIXTURE.lock().unwrap().clone().unwrap(); + + let mut builder = loom::model::Builder::new(); + builder.max_branches = 100_000; + builder.check(move || { + let doc = LoroDoc::new(); + doc.import(&f.snapshot).unwrap(); + let (doc1, doc2) = (doc.clone(), doc.clone()); + let (f1, f2) = (f.clone(), f.clone()); + let h0 = loom::thread::spawn(move || { + doc1.has_container(&f1.meta); + }); + let h1 = loom::thread::spawn(move || { + assert!(doc2.import_json_updates(f2.bad_json.as_str()).is_err()); + }); + h0.join().unwrap(); + h1.join().unwrap(); + assert!(doc.has_container(&f.meta)); + let history = + serde_json::to_string(&doc.export_json_updates(&Default::default(), &f.vv)) + .unwrap(); + assert_eq!(history, f.history); + }); + } } diff --git a/crates/loro/tests/unregistered_container_parent.rs b/crates/loro/tests/unregistered_container_parent.rs new file mode 100644 index 000000000..a11c96b31 --- /dev/null +++ b/crates/loro/tests/unregistered_container_parent.rs @@ -0,0 +1,952 @@ +//! A document loaded from a snapshot parses its change blocks lazily, so the +//! parent of a container can be unknown until the change that created it is +//! loaded. The metadata of a tree node that was created inside an already +//! deleted subtree is never materialized, so it has no state entry either: +//! querying or editing it panicked with "Parent is not registered" and aborted +//! the process (loro-dev/loro#1158). See `context/arena-parent-links.md`. +//! +//! Every test compares against a document that imported the same history as +//! updates, where every change is parsed on import. + +use loro::{ + ContainerID, ContainerTrait, ExportMode, LoroDoc, LoroError, ToJson, TreeID, TreeParentId, + UndoManager, ID, +}; +use std::sync::{Arc, Mutex}; + +/// Peer 2 creates `child` under `parent` while peer 1 deletes `parent`; +/// `filler` later commits push the child's `Create` into an older block. +fn concurrent_child_of_deleted_parent(filler: usize) -> (LoroDoc, TreeID) { + let a = LoroDoc::new(); + a.set_peer_id(1).unwrap(); + let tree = a.get_tree("tree"); + let parent = tree.create(TreeParentId::Root).unwrap(); + a.commit(); + let b = LoroDoc::new(); + b.set_peer_id(2).unwrap(); + b.import(&a.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + let child = b.get_tree("tree").create(parent).unwrap(); + b.commit(); + tree.delete(parent).unwrap(); + a.commit(); + a.import(&b.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + for i in 0..filler { + a.get_map("m") + .insert(&format!("k{}", i % 50), i as i64) + .unwrap(); + a.commit(); + } + (a, child) +} + +fn import(bytes: &[u8]) -> LoroDoc { + let doc = LoroDoc::new(); + doc.import(bytes).unwrap(); + doc.set_peer_id(9).unwrap(); + doc +} + +/// The ways to load `src`: all updates (the reference), a snapshot, a fork, +/// and a shallow snapshot whose root is before `child`'s creation. +fn loaded(src: &LoroDoc) -> Vec<(&'static str, LoroDoc)> { + let first_change = LoroDoc::new(); + first_change + .import( + &src.export(ExportMode::updates_in_range(vec![loro::IdSpan::new( + 1, 0, 1, + )])) + .unwrap(), + ) + .unwrap(); + let fork = src.fork(); + fork.set_peer_id(9).unwrap(); + vec![ + ( + "updates", + import(&src.export(ExportMode::all_updates()).unwrap()), + ), + ( + "snapshot", + import(&src.export(ExportMode::Snapshot).unwrap()), + ), + ("fork", fork), + ( + "shallow", + import( + &src.export(ExportMode::shallow_snapshot( + &first_change.oplog_frontiers(), + )) + .unwrap(), + ), + ), + ] +} + +#[test] +fn deleted_meta_queries_and_edits_after_loading() { + let (src, child) = concurrent_child_of_deleted_parent(5000); + let meta_id = child.associated_meta_container(); + for (how, doc) in loaded(&src) { + let tree = doc.get_tree("tree"); + assert!(tree.is_node_deleted(&child).unwrap(), "{how}"); + let meta = tree.get_meta(child).unwrap(); + assert!(meta.is_deleted(), "{how}"); + assert!( + matches!(meta.insert("k", 1), Err(LoroError::ContainerDeleted { .. })), + "{how}" + ); + assert_eq!(doc.get_path_to_container(&meta_id), None, "{how}"); + assert!(doc.has_container(&meta_id), "{how}"); + assert!( + doc.get_container(meta_id.clone()).unwrap().is_deleted(), + "{how}" + ); + } +} + +#[test] +fn container_by_id_in_a_fresh_snapshot_doc() { + // No handler registered the meta before the lookup. + let (src, child) = concurrent_child_of_deleted_parent(5000); + let meta_id = child.associated_meta_container(); + for (how, doc) in loaded(&src) { + let meta = doc.get_container(meta_id.clone()).unwrap(); + assert!(meta.is_deleted(), "{how}"); + assert_eq!(doc.get_path_to_container(&meta_id), None, "{how}"); + } +} + +#[test] +fn reviving_the_node_after_loading() { + let (src, child) = concurrent_child_of_deleted_parent(5000); + let meta_id = child.associated_meta_container(); + for (how, doc) in loaded(&src) { + let tree = doc.get_tree("tree"); + let meta = tree.get_meta(child).unwrap(); + assert!(meta.is_deleted(), "{how}"); + tree.mov(child, TreeParentId::Root).unwrap(); + doc.commit(); + assert!(!meta.is_deleted(), "{how}"); + meta.insert("k", 1).unwrap(); + doc.commit(); + assert_eq!( + doc.get_path_to_container(&meta_id) + .unwrap() + .last() + .unwrap() + .0, + meta_id, + "{how}" + ); + } +} + +/// An update from another peer that revives `child` and sets `x` in its meta. +fn revival_and_meta_edit_by_another_peer(src: &LoroDoc, child: TreeID) -> Vec { + let editor = LoroDoc::new(); + editor.set_peer_id(4).unwrap(); + editor + .import(&src.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + let before = editor.oplog_vv(); + editor + .get_tree("tree") + .mov(child, TreeParentId::Root) + .unwrap(); + editor + .get_tree("tree") + .get_meta(child) + .unwrap() + .insert("x", 1) + .unwrap(); + editor.commit(); + editor.export(ExportMode::updates(&before)).unwrap() +} + +#[test] +fn importing_an_edit_of_the_meta_with_a_subscriber() { + // Another peer revives the node and edits its meta. The loaded doc emits + // events for the meta, which walks its ancestors and its path. + let (src, child) = concurrent_child_of_deleted_parent(5000); + let update = revival_and_meta_edit_by_another_peer(&src, child); + + let mut expected = None; + for (how, doc) in loaded(&src) { + let tree_id = doc.get_tree("tree").id(); + let events = Arc::new(Mutex::new(vec![])); + let e2 = events.clone(); + let _root = doc.subscribe_root(Arc::new(move |e| { + for c in e.events.iter() { + e2.lock() + .unwrap() + .push(format!("{:?} {:?}", c.target, c.path)); + } + })); + let tree_events = Arc::new(Mutex::new(0)); + let t2 = tree_events.clone(); + // A subscriber on the tree is notified through the meta's ancestors. + let _tree = doc.subscribe( + &tree_id, + Arc::new(move |e| { + *t2.lock().unwrap() += e.events.len(); + }), + ); + doc.import(&update).unwrap(); + let meta = doc.get_tree("tree").get_meta(child).unwrap(); + assert_eq!( + meta.get_value().to_json_value(), + serde_json::json!({"x": 1}), + "{how}" + ); + assert!(*tree_events.lock().unwrap() > 0, "{how}"); + let events = events.lock().unwrap().clone(); + match &expected { + None => expected = Some(events), + Some(expected) => assert_eq!(&events, expected, "{how}"), + } + } +} + +/// Undo and revert after the loaded doc revives the node itself. The local +/// move registers the meta's parent, so this also passes on `main`; it checks +/// that the result matches the reference. +#[test] +fn undo_and_revert_touching_the_meta() { + let (src, child) = concurrent_child_of_deleted_parent(5000); + let mut expected = None; + for (how, doc) in loaded(&src) { + let mut undo = UndoManager::new(&doc); + let tree = doc.get_tree("tree"); + let before_revival = doc.oplog_frontiers(); + tree.mov(child, TreeParentId::Root).unwrap(); + doc.commit(); + tree.get_meta(child).unwrap().insert("k", 1).unwrap(); + doc.commit(); + assert!(undo.undo().unwrap(), "{how}"); + assert!(undo.undo().unwrap(), "{how}"); + assert!(tree.get_meta(child).unwrap().is_deleted(), "{how}"); + assert!(undo.redo().unwrap(), "{how}"); + doc.revert_to(&before_revival).unwrap(); + doc.commit(); + let value = doc.get_deep_value().to_json_value(); + match &expected { + None => expected = Some(value), + Some(expected) => assert_eq!(&value, expected, "{how}"), + } + } +} + +/// Undoing a local edit transforms it against the changes imported since, +/// here another peer's revival and edit of the meta. +#[test] +fn undo_after_importing_an_edit_of_the_meta() { + let (src, child) = concurrent_child_of_deleted_parent(5000); + let update = revival_and_meta_edit_by_another_peer(&src, child); + let mut expected = None; + for (how, doc) in loaded(&src) { + let mut undo = UndoManager::new(&doc); + doc.get_map("m").insert("local", 1).unwrap(); + doc.commit(); + doc.import(&update).unwrap(); + assert!(undo.undo().unwrap(), "{how}"); + assert!(undo.redo().unwrap(), "{how}"); + assert!(undo.undo().unwrap(), "{how}"); + let meta = doc.get_tree("tree").get_meta(child).unwrap(); + assert_eq!( + meta.get_value().to_json_value(), + serde_json::json!({"x": 1}), + "{how}" + ); + assert!(doc.get_map("m").get("local").is_none(), "{how}"); + let value = doc.get_deep_value().to_json_value(); + match &expected { + None => expected = Some(value), + Some(expected) => assert_eq!(&value, expected, "{how}"), + } + } +} + +#[test] +fn ids_that_no_op_creates_are_not_containers() { + let (src, _) = concurrent_child_of_deleted_parent(5000); + // 10@1 is a map insert, and 900000@1 is not in the history. + for id in [ID::new(1, 10), ID::new(1, 900_000)] { + let cid = ContainerID::new_normal(id, loro::ContainerType::Map); + for (how, doc) in loaded(&src) { + assert!(!doc.has_container(&cid), "{how} {id}"); + assert!(doc.get_container(cid.clone()).is_none(), "{how} {id}"); + assert_eq!(doc.get_path_to_container(&cid), None, "{how} {id}"); + } + } +} + +/// The state found by `loro.js`'s review harness (seed 5, step 103): node +/// `7@40881` sits under the deleted `0@2`, and its meta has no state entry. +#[test] +fn review_harness_dump() { + let snapshot = include_bytes!("./unregistered_meta_parent.snapshot.bin"); + let updates = include_bytes!("./unregistered_meta_parent.updates.bin"); + let node = TreeID::new(40881, 7); + let from_snapshot = import(snapshot); + for (how, doc) in [ + ("updates", import(updates)), + ("fork", from_snapshot.fork()), + ("snapshot", from_snapshot), + ] { + let tree = doc.get_tree("tree"); + assert!(tree.is_node_deleted(&node).unwrap(), "{how}"); + let meta = tree.get_meta(node).unwrap(); + assert!(meta.is_deleted(), "{how}"); + assert!( + matches!(meta.delete("c0"), Err(LoroError::ContainerDeleted { .. })), + "{how}" + ); + assert!( + doc.get_container(node.associated_meta_container()) + .unwrap() + .is_deleted(), + "{how}" + ); + } +} + +/// A failed import may load old changes while computing its diff, e.g. to find the parent of +/// a container that only the history knows. Loading a change registers the containers its +/// ops use, and the rollback of the failed import drops registrations made during it, so the +/// loaded change must not keep indices into them. +#[test] +fn failed_import_that_loads_an_old_change() { + let a = LoroDoc::new(); + a.set_peer_id(1).unwrap(); + let tree = a.get_tree("tree"); + let parent = tree.create(TreeParentId::Root).unwrap(); + a.commit(); + let b = LoroDoc::new(); + b.set_peer_id(2).unwrap(); + b.import(&a.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + let child = b.get_tree("tree").create(parent).unwrap(); + // A container only this change mentions: it has no state entry and is only + // registered once the change is loaded. + b.get_map("b") + .insert_container("inner", loro::LoroText::new()) + .unwrap() + .insert(0, "inner") + .unwrap(); + b.get_map("b").delete("inner").unwrap(); + b.commit(); + let b_updates = b.export(ExportMode::all_updates()).unwrap(); + let b_vv = b.oplog_vv(); + tree.delete(parent).unwrap(); + a.commit(); + a.import(&b_updates).unwrap(); + for i in 0..5000 { + a.get_map("m") + .insert(&format!("k{}", i % 50), i as i64) + .unwrap(); + a.commit(); + } + let snapshot = a.export(ExportMode::Snapshot).unwrap(); + + // Peer 4 (who has not seen the delete) edits `child`'s meta, then inserts a list + // element far out of bounds, which the state rejects. + let e = LoroDoc::new(); + e.set_peer_id(4).unwrap(); + e.import(&b_updates).unwrap(); + e.get_tree("tree") + .get_meta(child) + .unwrap() + .insert("x", 1) + .unwrap(); + let list = e.get_list("list"); + list.insert(0, "seed").unwrap(); + list.insert(1, "tail").unwrap(); + e.commit(); + let mut json = serde_json::to_value(e.export_json_updates(&b_vv, &e.oplog_vv())).unwrap(); + let last_op = json["changes"].as_array_mut().unwrap().last_mut().unwrap()["ops"] + .as_array_mut() + .unwrap() + .last_mut() + .unwrap(); + assert_eq!(last_op["content"]["type"], "insert"); + last_op["content"]["pos"] = 1000.into(); + let carrier = LoroDoc::new(); + carrier.import(&b_updates).unwrap(); + carrier.detach(); + carrier + .import_json_updates(serde_json::to_string(&json).unwrap()) + .unwrap(); + let bad = carrier.export(ExportMode::updates(&b_vv)).unwrap(); + + let dst = import(&snapshot); + dst.import(&bad) + .expect_err("the out-of-bounds list insert must fail the import"); + + // New registrations may reuse the indices the rollback dropped. + for i in 0..8 { + dst.get_map("fresh") + .insert_container(&format!("c{i}"), loro::LoroMap::new()) + .unwrap(); + } + dst.commit(); + // The history, including the change the failed import loaded, must still name the + // right containers. + let history = |doc: &LoroDoc| { + serde_json::to_string(&doc.export_json_updates(&Default::default(), &a.oplog_vv())).unwrap() + }; + assert_eq!(history(&dst), history(&a)); + assert!(dst.get_tree("tree").get_meta(child).unwrap().is_deleted()); +} + +/// loro-dev/loro#1161: parsing a change also allocates its ops' values in the arena, and +/// the rollback of a failed import truncates them. A change the failed import loaded kept +/// value slices past the end, and a later checkout that read them panicked. +#[test] +fn failed_import_that_loads_old_movable_list_values() { + let a = LoroDoc::new(); + a.set_peer_id(1).unwrap(); + let list = a.get_movable_list("ml"); + for i in 0..20 { + list.insert(i, format!("v{i}")).unwrap(); + a.commit(); + } + let inserted = a.oplog_frontiers(); + for i in 0..20 { + list.set(i, format!("s{i}")).unwrap(); + a.commit(); + } + for i in 0..3000 { + a.get_map("m").insert("k", i).unwrap(); + a.commit(); + } + let snapshot = a.export(ExportMode::Snapshot).unwrap(); + let vv = a.oplog_vv(); + + for edit in ["set", "move"] { + // Peer 4 edits the movable list, whose history the import's diff loads, then + // inserts a list element far out of bounds, which the state rejects. + let e = import(&snapshot); + e.set_peer_id(4).unwrap(); + match edit { + "set" => e.get_movable_list("ml").set(3, "x").unwrap(), + _ => e.get_movable_list("ml").mov(2, 7).unwrap(), + } + e.get_list("list").insert(0, 1).unwrap(); + e.get_list("list").insert(1, 2).unwrap(); + e.commit(); + let mut json = serde_json::to_value(e.export_json_updates(&vv, &e.oplog_vv())).unwrap(); + let last_op = json["changes"].as_array_mut().unwrap().last_mut().unwrap()["ops"] + .as_array_mut() + .unwrap() + .last_mut() + .unwrap(); + last_op["content"]["pos"] = 1000.into(); + let carrier = import(&snapshot); + carrier.detach(); + carrier + .import_json_updates(serde_json::to_string(&json).unwrap()) + .unwrap(); + let bad = carrier.export(ExportMode::updates(&vv)).unwrap(); + + let dst = import(&snapshot); + let reference = import(&snapshot); + for doc in [&dst, &reference] { + // Registered before the import, so only the values are rolled back. + doc.get_movable_list("ml"); + doc.get_map("m"); + } + dst.import(&bad) + .expect_err("the out-of-bounds list insert must fail the import"); + for doc in [&dst, &reference] { + doc.checkout(&inserted).unwrap(); + } + let value = |doc: &LoroDoc| doc.get_deep_value().to_json_value(); + assert_eq!(value(&dst), value(&reference), "{edit}"); + for doc in [&dst, &reference] { + doc.checkout_to_latest(); + } + assert_eq!(value(&dst), value(&reference), "{edit}"); + } +} + +/// Querying these metas resolves their parents through the change store without the op +/// log lock, while other threads read the store under it. A lock-order inversion inside +/// the store deadlocked (about every other run of 3000 trials hit it); +/// `multi_thread_test.rs` has the loom model. Trials: `UNREGISTERED_PARENT_THREAD_TRIALS` +/// (default 500). +#[test] +fn queries_race_with_history_readers() { + use std::sync::{mpsc, Barrier}; + use std::time::Duration; + + // Peer 2 creates 32 children, spread over change blocks, under parents that peer 1 + // deletes concurrently. + let a = LoroDoc::new(); + a.set_peer_id(1).unwrap(); + let tree = a.get_tree("tree"); + let parents: Vec<_> = (0..4) + .map(|_| tree.create(TreeParentId::Root).unwrap()) + .collect(); + a.commit(); + let b = LoroDoc::new(); + b.set_peer_id(2).unwrap(); + b.import(&a.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + let mut children = vec![]; + for p in parents.iter() { + for _ in 0..8 { + children.push(b.get_tree("tree").create(*p).unwrap()); + b.commit(); + for k in 0..30 { + b.get_map("filler").insert("k", k).unwrap(); + b.commit(); + } + } + } + for p in parents.iter() { + tree.delete(*p).unwrap(); + } + a.commit(); + a.import(&b.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + for i in 0..1000 { + a.get_map("m").insert("k", i).unwrap(); + a.commit(); + } + let snapshot = a.export(ExportMode::Snapshot).unwrap(); + + let trials: usize = std::env::var("UNREGISTERED_PARENT_THREAD_TRIALS") + .map(|s| s.parse().unwrap()) + .unwrap_or(500); + let (progress, rx) = mpsc::channel(); + let runner = std::thread::spawn(move || { + for _ in 0..trials { + let doc = import(&snapshot); + let barrier = Arc::new(Barrier::new(3)); + let spawn = |f: Box| { + let (doc, barrier) = (doc.clone(), barrier.clone()); + std::thread::spawn(move || { + barrier.wait(); + f(&doc) + }) + }; + let metas = children.clone(); + let ids = children.clone(); + let threads = [ + spawn(Box::new(move |doc| { + let tree = doc.get_tree("tree"); + for c in metas { + assert!(tree.get_meta(c).unwrap().is_deleted()); + } + })), + spawn(Box::new(move |doc| { + for c in ids { + let id = c.associated_meta_container(); + assert!(doc.has_container(&id)); + assert_eq!(doc.get_path_to_container(&id), None); + } + })), + spawn(Box::new(|doc| { + for _ in 0..3 { + assert!(doc.len_changes() > 0); + doc.export(ExportMode::all_updates()).unwrap(); + doc.frontiers_to_vv(&doc.oplog_frontiers()).unwrap(); + } + })), + ]; + for t in threads { + t.join().unwrap(); + } + progress.send(()).unwrap(); + } + }); + loop { + match rx.recv_timeout(Duration::from_secs(30)) { + Ok(()) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => break, + Err(mpsc::RecvTimeoutError::Timeout) => panic!("no progress for 30 s: deadlock"), + } + } + runner.join().unwrap(); +} + +/// An import that fails before its rollback scope begins (decoding, or depending on history +/// before a shallow root) rolls the arena back to where it started, while another thread +/// resolves parents through the change store. Neither the parsed blocks nor the resolver's +/// caller may keep indices the rollback frees (`multi_thread_test.rs` has the loom model). +/// Before the fix, a failed JSON import hit it within the default trials in most runs. +/// Trials: `UNREGISTERED_PARENT_THREAD_TRIALS` (default 120). +#[test] +fn queries_race_with_failing_imports() { + use loro::{LoroMap, VersionVector}; + use std::sync::Barrier; + use std::time::Instant; + + // Peer 2 creates children under parents that peer 1 deletes concurrently, and map + // children that it deletes again: none of them is in the state, except that a shallow + // snapshot keeps the metas of deleted nodes. + let a = LoroDoc::new(); + a.set_peer_id(1).unwrap(); + let tree = a.get_tree("tree"); + let parents: Vec<_> = (0..4) + .map(|_| tree.create(TreeParentId::Root).unwrap()) + .collect(); + a.commit(); + let before_root = a.export(ExportMode::all_updates()).unwrap(); + a.get_map("m").insert("k", -1).unwrap(); + a.commit(); + let root = a.oplog_frontiers(); + let root_vv = a.oplog_vv(); + let shallow_at_root = a.export(ExportMode::shallow_snapshot(&root)).unwrap(); + let b = LoroDoc::new(); + b.set_peer_id(2).unwrap(); + b.import(&a.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + let mut ids = vec![]; + for (i, p) in parents.iter().enumerate() { + for j in 0..8 { + let child = b.get_tree("tree").create(*p).unwrap(); + ids.push(child.associated_meta_container()); + let key = format!("c{i}_{j}"); + let gone = b + .get_map("gone") + .insert_container(&key, LoroMap::new()) + .unwrap(); + gone.insert("v", 1).unwrap(); + ids.push(gone.id()); + b.get_map("gone").delete(&key).unwrap(); + b.commit(); + for k in 0..30 { + b.get_map("filler").insert("k", k).unwrap(); + b.commit(); + } + } + } + for p in parents.iter() { + tree.delete(*p).unwrap(); + } + a.commit(); + a.import(&b.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + for i in 0..1000 { + a.get_map("m").insert("k", i).unwrap(); + a.commit(); + } + let vv = a.oplog_vv(); + let snapshot = a.export(ExportMode::Snapshot).unwrap(); + // A snapshot of a shallow document whose history starts at `root`. It received the + // rest as updates, so its state lacks the same containers, and it loads lazily. + let shallow_doc = import(&shallow_at_root); + shallow_doc + .import(&a.export(ExportMode::updates(&root_vv)).unwrap()) + .unwrap(); + let shallow = shallow_doc.export(ExportMode::Snapshot).unwrap(); + + // Many changes that create containers, so the failing import registers many and takes a + // while. Decoding the JSON fails at the empty change at its end. + let e = import(&snapshot); + e.set_peer_id(8).unwrap(); + for i in 0..1000 { + e.get_map("new") + .insert_container(&format!("n{i}"), LoroMap::new()) + .unwrap(); + e.commit(); + } + let mut json = serde_json::to_value(e.export_json_updates(&vv, &e.oplog_vv())).unwrap(); + let changes = json["changes"].as_array_mut().unwrap(); + let mut empty = changes.last().unwrap().clone(); + empty["ops"] = serde_json::json!([]); + changes.push(empty); + let bad_json = serde_json::to_string(&json).unwrap(); + // These depend on history before the shallow root. + let o = LoroDoc::new(); + o.set_peer_id(10).unwrap(); + o.import(&before_root).unwrap(); + for i in 0..1000 { + o.get_map("old") + .insert_container(&format!("n{i}"), LoroMap::new()) + .unwrap(); + o.commit(); + } + let old_deps = o.export(ExportMode::all_updates()).unwrap(); + + let history = |doc: &LoroDoc, from: &VersionVector| { + serde_json::to_string(&doc.export_json_updates(from, &vv)).unwrap() + }; + let reference = import(&a.export(ExportMode::all_updates()).unwrap()); + let empty_vv = VersionVector::new(); + let expected = [ + history(&reference, &empty_vv), + history(&reference, &root_vv), + ]; + // Also IDs of peer 2's other ops, which create no container. In the shallow document the + // containers above are in the state, and only these make the resolver load a block. + for counter in (0..vv.get(&2).copied().unwrap()).step_by(13) { + ids.push(ContainerID::new_normal( + ID::new(2, counter), + loro::ContainerType::Map, + )); + } + let exists: Vec = ids.iter().map(|id| reference.has_container(id)).collect(); + + let fail_import = |doc: &LoroDoc, is_shallow: bool| { + if is_shallow { + assert!(matches!( + doc.import(&old_deps), + Err(LoroError::ImportUpdatesThatDependsOnOutdatedVersion) + )); + } else { + assert!(doc.import_json_updates(bad_json.as_str()).is_err()); + } + }; + // How long each failing import takes here, to sweep the queries across it. + let window = [false, true].map(|is_shallow| { + let doc = import(if is_shallow { &shallow } else { &snapshot }); + let start = Instant::now(); + fail_import(&doc, is_shallow); + start.elapsed() + }); + + let trials: usize = std::env::var("UNREGISTERED_PARENT_THREAD_TRIALS") + .map(|s| s.parse().unwrap()) + .unwrap_or(120); + for trial in 0..trials { + let is_shallow = trial % 2 == 1; + let doc = import(if is_shallow { &shallow } else { &snapshot }); + let barrier = Arc::new(Barrier::new(2)); + let query = { + let (doc, barrier, ids) = (doc.clone(), barrier.clone(), ids.clone()); + let delay = window[is_shallow as usize] * (trial as u32 / 2 % 20) / 20; + std::thread::spawn(move || { + barrier.wait(); + let start = Instant::now(); + while start.elapsed() < delay { + std::hint::spin_loop(); + } + for id in ids.iter() { + doc.has_container(id); + } + }) + }; + std::thread::scope(|scope| { + scope.spawn(|| { + barrier.wait(); + fail_import(&doc, is_shallow); + }); + }); + query.join().unwrap(); + for (id, exists) in ids.iter().zip(exists.iter()) { + assert_eq!(doc.has_container(id), *exists, "trial {trial}: {id}"); + } + let from = if is_shallow { &root_vv } else { &empty_vv }; + assert_eq!( + history(&doc, from), + expected[is_shallow as usize], + "trial {trial}" + ); + } +} + +mod random { + use super::*; + use loro::{LoroText, TreeID}; + use rand::{rngs::StdRng, seq::SliceRandom, Rng, SeedableRng}; + + fn random_history(seed: u64) -> LoroDoc { + let mut rng = StdRng::seed_from_u64(seed); + let peers: Vec = (1..=3) + .map(|p| { + let d = LoroDoc::new(); + d.set_peer_id(p).unwrap(); + d + }) + .collect(); + for _ in 0..60 { + let i = rng.gen_range(0..peers.len()); + let doc = &peers[i]; + let tree = doc.get_tree("tree"); + let nodes = tree.nodes(); + let pick = |rng: &mut StdRng| nodes.choose(rng).copied(); + let _ = match rng.gen_range(0..10) { + 0..=2 => { + let parent = match pick(&mut rng) { + Some(n) if rng.gen_bool(0.7) => TreeParentId::Node(n), + _ => TreeParentId::Root, + }; + tree.create(parent).map(|_| ()) + } + 3 => match (pick(&mut rng), pick(&mut rng)) { + (Some(t), Some(p)) => tree.mov(t, p), + _ => Ok(()), + }, + 4 => match pick(&mut rng) { + Some(t) => tree.delete(t), + None => Ok(()), + }, + 5 => match pick(&mut rng) { + Some(t) => tree + .get_meta(t) + .and_then(|m| m.insert("k", rng.gen::())), + None => Ok(()), + }, + 6 => match pick(&mut rng) { + Some(t) => tree + .get_meta(t) + .and_then(|m| m.insert_container("text", LoroText::new())?.insert(0, "t")), + None => Ok(()), + }, + 7 => { + // Later history, so earlier changes end up in older blocks. + for k in 0..rng.gen_range(50..300) { + doc.get_map("filler") + .insert(&format!("k{}", k % 20), k) + .unwrap(); + doc.commit(); + } + Ok(()) + } + _ => { + let j = rng.gen_range(0..peers.len()); + if i != j { + let u = peers[j] + .export(ExportMode::updates(&doc.oplog_vv())) + .unwrap(); + doc.import(&u).unwrap(); + } + Ok(()) + } + }; + doc.commit(); + } + let all = LoroDoc::new(); + for p in peers.iter() { + all.import(&p.export(ExportMode::all_updates()).unwrap()) + .unwrap(); + } + all + } + + /// What a document says about the metadata of `node`. + fn observe(doc: &LoroDoc, node: TreeID) -> String { + let tree = doc.get_tree("tree"); + let meta_id = node.associated_meta_container(); + let meta = tree.get_meta(node).unwrap(); + format!( + "node_deleted={:?} meta_deleted={} has={} by_id={:?} path={:?} value={:?}", + tree.is_node_deleted(&node), + meta.is_deleted(), + doc.has_container(&meta_id), + doc.get_container(meta_id.clone()).map(|c| c.is_deleted()), + doc.get_path_to_container(&meta_id), + meta.get_deep_value(), + ) + } + + fn check(seed: u64) { + let src = random_history(seed); + let mut rng = StdRng::seed_from_u64(seed ^ 0x5eed); + let reference = import(&src.export(ExportMode::all_updates()).unwrap()); + let nodes = reference.get_tree("tree").nodes(); + if nodes.is_empty() { + return; + } + let vv = src.oplog_vv(); + let peers: Vec<_> = vv.iter().filter(|(_, n)| **n > 0).collect(); + let (peer, end) = peers.choose(&mut rng).unwrap(); + let root = src.vv_to_frontiers( + &src.frontiers_to_vv(&ID::new(**peer, rng.gen_range(0..**end)).into()) + .unwrap(), + ); + let fork = src.fork(); + fork.set_peer_id(9).unwrap(); + let loaded = [ + ( + "snapshot", + import(&src.export(ExportMode::Snapshot).unwrap()), + ), + ("fork", fork), + ( + "shallow", + import(&src.export(ExportMode::shallow_snapshot(&root)).unwrap()), + ), + ( + "state_only", + import(&src.export(ExportMode::state_only(None)).unwrap()), + ), + ]; + // Visit the nodes in a random order: resolving one meta must not change the + // answer for another. + let mut order = nodes.clone(); + order.shuffle(&mut rng); + for (how, doc) in loaded.iter() { + for &node in order.iter() { + // Every source has the latest version, where every node exists. + assert!( + doc.get_tree("tree").contains(node), + "seed {seed} {how} {node}" + ); + assert_eq!( + observe(doc, node), + observe(&reference, node), + "seed {seed} {how} node {node}" + ); + } + } + + // Edit every meta, then revive some nodes and edit again. + let mut revive = nodes.clone(); + revive.shuffle(&mut rng); + revive.truncate(3); + for doc in std::iter::once(&reference).chain(loaded.iter().map(|(_, d)| d)) { + for &node in order.iter() { + let r = doc + .get_tree("tree") + .get_meta(node) + .unwrap() + .insert("probe", 1); + assert!( + r.is_ok() || matches!(r, Err(LoroError::ContainerDeleted { .. })), + "seed {seed}: {r:?}" + ); + } + doc.commit(); + for &node in revive.iter() { + let _ = doc.get_tree("tree").mov(node, TreeParentId::Root); + } + doc.commit(); + for &node in order.iter() { + let _ = doc + .get_tree("tree") + .get_meta(node) + .unwrap() + .insert("after", 2); + } + doc.commit(); + } + for (how, doc) in loaded.iter() { + for &node in order.iter() { + assert_eq!( + observe(doc, node), + observe(&reference, node), + "seed {seed} {how} node {node} after edits" + ); + } + } + } + + #[test] + fn metas_after_loading_match_a_full_history_import() { + let seeds: Vec = match std::env::var("UNREGISTERED_PARENT_SEEDS") { + Ok(r) => { + let (a, b) = r.split_once("..").unwrap(); + (a.parse().unwrap()..b.parse().unwrap()).collect() + } + Err(_) => (0..20).collect(), + }; + for seed in seeds { + check(seed); + } + } +} diff --git a/crates/loro/tests/unregistered_meta_parent.snapshot.bin b/crates/loro/tests/unregistered_meta_parent.snapshot.bin new file mode 100644 index 000000000..50f46fec1 Binary files /dev/null and b/crates/loro/tests/unregistered_meta_parent.snapshot.bin differ diff --git a/crates/loro/tests/unregistered_meta_parent.updates.bin b/crates/loro/tests/unregistered_meta_parent.updates.bin new file mode 100644 index 000000000..f02c35a18 Binary files /dev/null and b/crates/loro/tests/unregistered_meta_parent.updates.bin differ