diff --git a/document/format/src/export.rs b/document/format/src/export.rs index 14ce23f2be6..96d0c4c9935 100644 --- a/document/format/src/export.rs +++ b/document/format/src/export.rs @@ -150,6 +150,7 @@ impl Gdd { last_broadcast_rev: self.session.last_broadcast_rev(), redo_stack: self.session.redo_stack().to_vec(), next_node_counter: self.session.next_node_counter(), + next_hot_sequence: self.session.next_hot_sequence(), view_settings: self.view_settings.clone(), network_view_settings: self.network_view_settings.clone(), }; diff --git a/document/format/src/lib.rs b/document/format/src/lib.rs index 63f49b74ccf..eee3178987c 100644 --- a/document/format/src/lib.rs +++ b/document/format/src/lib.rs @@ -158,6 +158,8 @@ impl Gdd { if let Some(rev) = session_state.last_broadcast_rev { session.publish_up_to(rev); } + // Every arm must continue this peer's authored-op count too. + session.restore_hot_sequence(session_state.next_hot_sequence); replay_hot_log(&working, &layout, codecs.hot_log, &mut session).await?; diff --git a/document/format/src/persist.rs b/document/format/src/persist.rs index 19bc7476dfc..1ab4042005b 100644 --- a/document/format/src/persist.rs +++ b/document/format/src/persist.rs @@ -198,6 +198,7 @@ impl Gdd { last_broadcast_rev: self.session.last_broadcast_rev(), redo_stack: self.session.redo_stack().to_vec(), next_node_counter: self.session.next_node_counter(), + next_hot_sequence: self.session.next_hot_sequence(), view_settings: self.view_settings.clone(), network_view_settings: self.network_view_settings.clone(), }; diff --git a/document/format/src/session_state.rs b/document/format/src/session_state.rs index 8b1135cf8f1..61fb5174ede 100644 --- a/document/format/src/session_state.rs +++ b/document/format/src/session_state.rs @@ -4,7 +4,7 @@ //! //! Lives in `session.json`. Rewritten on retirement. -use document_graph_storage::{NetworkId, PeerId, Rev}; +use document_graph_storage::{HotSequence, NetworkId, PeerId, Rev}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; @@ -39,4 +39,9 @@ pub struct SessionState { /// Per-peer like [`view_settings`](Self::view_settings); opaque `ui::nav::*` / `ui::previewing` blobs. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub network_view_settings: BTreeMap>, + /// How many hot ops this peer has authored, feeding [`document_graph_storage::HotOp::sequence`]. A + /// reopen continues the run instead of reusing a spent sequence. Appended last: a positional codec + /// decodes these fields in declaration order. + #[serde(default)] + pub next_hot_sequence: HotSequence, } diff --git a/document/format/tests/open_create.rs b/document/format/tests/open_create.rs index 4bdd43d7bd7..d11649baab5 100644 --- a/document/format/tests/open_create.rs +++ b/document/format/tests/open_create.rs @@ -5,7 +5,7 @@ use document_container::AnyContainer; use document_container::backends::memory::MemoryBackend; use document_format::{Codec, Error, GddV1, GddV1Layout, Layout, Manifest, io, manifest}; -use document_graph_storage::{HotOp, Network, NetworkId, PeerId, ROOT_NETWORK, RegistryDelta, TimeStamp}; +use document_graph_storage::{HotOp, HotSequence, Network, NetworkId, PeerId, ROOT_NETWORK, RegistryDelta, TimeStamp}; fn empty_container() -> AnyContainer { AnyContainer::Memory(MemoryBackend::new()) @@ -128,6 +128,7 @@ fn apply_hot_op_persists_to_hot_log_and_survives_reopen() { network: Network::default(), }, timestamp: TimeStamp { counter: 1, peer: PeerId(5) }, + sequence: HotSequence(1), }; gdd.apply_hot_op(hot_op).unwrap_or_else(|error| panic!("apply_hot_op failed: {error:?}")); @@ -154,6 +155,7 @@ fn retire_moves_eligible_hot_ops_to_history_and_keeps_rest() { network: Network::default(), }, timestamp: TimeStamp { counter: 1, peer: PeerId(5) }, + sequence: HotSequence(1), }; let late = HotOp { op: RegistryDelta::AddNetwork { @@ -161,6 +163,7 @@ fn retire_moves_eligible_hot_ops_to_history_and_keeps_rest() { network: Network::default(), }, timestamp: TimeStamp { counter: 10, peer: PeerId(5) }, + sequence: HotSequence(2), }; gdd.apply_hot_op(early).unwrap(); gdd.apply_hot_op(late).unwrap(); @@ -200,6 +203,7 @@ fn last_broadcast_rev_persists_across_reopen() { network: Network::default(), }, timestamp: TimeStamp { counter: 1, peer: PeerId(5) }, + sequence: HotSequence(1), }; gdd.apply_hot_op(op).unwrap(); let retired = gdd.retire(TimeStamp { counter: 1, peer: PeerId(5) }).unwrap_or_else(|error| panic!("retire failed: {error:?}")); @@ -655,6 +659,7 @@ fn persist_path_writes_at_manifest_declared_codec_paths() { network: Network::default(), }, timestamp: TimeStamp { counter: 1, peer: PeerId(5) }, + sequence: HotSequence(1), }; gdd.apply_hot_op(hot_op).unwrap_or_else(|error| panic!("apply_hot_op failed: {error:?}")); diff --git a/document/graph-storage/src/document.rs b/document/graph-storage/src/document.rs index bd298d665df..dfda39e1847 100644 --- a/document/graph-storage/src/document.rs +++ b/document/graph-storage/src/document.rs @@ -1,6 +1,6 @@ use crate::{ - CrdtError, Delta, ExportSlot, History, HotOp, LamportClock, MAX_EXPORT_SLOTS, NetworkId, NodeId, NodeInput, PeerId, Registry, RegistryDelta, ResourceEntry, Rev, SourceValue, TimeStamp, - apply_attribute_delta, reverse_attribute_delta, + CrdtError, Delta, ExportSlot, History, HotOp, HotOpId, HotSequence, LamportClock, MAX_EXPORT_SLOTS, NetworkId, NodeId, NodeInput, PeerId, Registry, RegistryDelta, ResourceEntry, RetiredHotOps, + Rev, SourceValue, TimeStamp, apply_attribute_delta, reverse_attribute_delta, }; #[derive(Clone, Debug)] @@ -11,6 +11,10 @@ pub struct Document { /// Live broadcast stream, applied to the `working_registry` on receive, GC'd at retirement. /// Persisted for crash recovery so in-flight unretired work survives editor restarts. pub(crate) hot_log: Vec, + /// Which hot ops history already covers, so one arriving after its own retirement is dropped instead + /// of re-entering the log. Retirement discards the delta's link back to its hot op, leaving this the + /// only record. See [`RetiredHotOps`]. + pub(crate) retired: RetiredHotOps, /// The registry as of the last retirement, with no un-retired hot ops applied. Retirement computes /// each delta's `reverse` against this (so LWW reverses capture the true pre-op value, not the /// hot-polluted working state) and advances it, stamping fields at the fresh `T_retire`. Kept equal @@ -36,6 +40,11 @@ pub struct Document { /// peer is calling; collision avoidance comes from hashing `(self.peer, counter)`, so two peers /// reading the same counter still produce distinct IDs. pub(crate) next_node_counter: u64, + /// Counts this peer's own hot ops, so each carries its position in a gap-free run. See [`HotOp::sequence`]. + pub(crate) next_hot_sequence: HotSequence, + /// Set while the registries are still folded from an older history. A failed refold is retried; the + /// derived state does not stay behind the history it comes from. + pub(crate) refold_owed: bool, } impl Document { @@ -66,24 +75,12 @@ impl Document { self.revert_delta(target, delta) } - /// Search every delta reachable from `head` (following all parents, including a merge's - /// `extra_parents`) for the first matching `predicate`, breadth-first. Resurrection needs full - /// ancestry reachability, so a node added only on a merged-in branch is still found. + /// The last match in canonical history order, over everything reachable from `head` or any tip. Not a + /// walk from `head`: peers sit on different merge revs, and resurrection must pick the same delta. fn find_in_ancestry(&self, predicate: impl Fn(&Delta) -> bool) -> Option { - let mut queue: std::collections::VecDeque = self.head.into_iter().collect(); - let mut seen: std::collections::HashSet = self.head.into_iter().collect(); - while let Some(rev) = queue.pop_front() { - let Some(delta) = self.history.get(rev) else { continue }; - if predicate(delta) { - return Some(delta.clone()); - } - for parent in delta.all_parents() { - if seen.insert(parent) { - queue.push_back(parent); - } - } - } - None + let reachable = self.history.ancestors(self.head.into_iter().chain(self.history.tips())); + + self.history.iter().filter(|delta| reachable.contains(&delta.id) && predicate(delta)).last().cloned() } /// Apply a delta's `reverse` as the new forward op (silent-zone undo). Force-applied: structural @@ -99,9 +96,10 @@ impl Document { self.apply_op_with(target, delta.kind, delta.timestamp, ApplyMode::Force) } - /// Apply a live broadcast op. Updates the registry via LWW and appends to the hot log. - /// Doesn't touch history or `head` — hot ops are transient. - pub fn apply_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> { + /// Apply a locally staged op: LWW into the registry, append to the hot log, leave history and `head` + /// alone. Crate-private because it skips the retirement check an op off the wire needs; staging must + /// not have a covered sequence silently drop local work. Outside callers use [`Session::apply_hot_op`]. + pub(crate) fn apply_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> { self.apply_op(hot_op.op.clone(), hot_op.timestamp)?; self.hot_log.push(hot_op); Ok(()) @@ -111,11 +109,55 @@ impl Document { /// re-applying an op whose effect is already reflected in the registry is a no-op rather /// than an error. pub fn replay_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> { + // Already in retired history; re-adding it would leave an entry no retirement will ever name. + if self.is_retired(hot_op.id()) { + return Ok(()); + } + + // Identified by timestamp, so a re-announcement of one already held must not become a second copy. + if self.hot_log.iter().any(|held| held.timestamp == hot_op.timestamp) { + return Ok(()); + } + + // Replaying our own ops is what carries the sequence counter across a reload. + if hot_op.timestamp.peer == self.peer { + self.next_hot_sequence = self.next_hot_sequence.max(hot_op.sequence); + } + self.apply_op_idempotent(hot_op.op.clone(), hot_op.timestamp)?; self.hot_log.push(hot_op); Ok(()) } + /// Whether this hot op has already been promoted into history. + pub(crate) fn is_retired(&self, id: HotOpId) -> bool { + self.retired.covers(id) + } + + /// Record newly retired hot ops, dropping any the hot log still holds. + pub(crate) fn mark_retired(&mut self, retired: impl IntoIterator) { + self.retired.extend(retired); + + self.drop_retired_hot_ops(); + } + + /// Take on a peer's retirement marks as well as this peer's. Returns whether a hot op was dropped, + /// which owes the caller a refold. + pub(crate) fn absorb_retired(&mut self, remote: &RetiredHotOps) -> bool { + self.retired.absorb(remote); + + self.drop_retired_hot_ops() + } + + /// Drop hot ops the retired snapshot already accounts for. A mark alone will not do: it can arrive + /// ahead of the delta carrying the op into history, stranding its effect in the working registry. + fn drop_retired_hot_ops(&mut self) -> bool { + let before = self.hot_log.len(); + self.hot_log.retain(|hot_op| !self.history.contains_timestamp(hot_op.timestamp)); + + before != self.hot_log.len() + } + /// Apply a retired commit. Idempotent on structural ops (AddNode/AddNetwork on existing /// targets, Remove on missing ones) since hot ops already produced the structural state. /// The point is to bump field timestamps to T_retire via the LWW arms. @@ -315,7 +357,7 @@ impl Document { /// Resurrect (from history) any nodes/networks an op references that were concurrently removed, so /// the op applies against a consistent registry. Cascading: a node's owning network is restored /// before the node. No-op for ops that reference nothing absent. - fn ensure_referenced_exist(&mut self, target: RegistryTarget, op: &RegistryDelta) -> Result<(), CrdtError> { + pub(crate) fn ensure_referenced_exist(&mut self, target: RegistryTarget, op: &RegistryDelta) -> Result<(), CrdtError> { match op { RegistryDelta::AddNode { node, .. } => self.ensure_network_exists(target, node.network())?, RegistryDelta::ChangeNodeInput { id, new_input, .. } => { @@ -354,17 +396,51 @@ impl Document { } fn ensure_node_exists(&mut self, target: RegistryTarget, node_id: NodeId) -> Result<(), CrdtError> { - if !self.registry_ref(target).node_instances.contains_key(&node_id) { - self.restore_node_from_history(target, node_id)?; + if self.registry_ref(target).node_instances.contains_key(&node_id) { + return Ok(()); } - Ok(()) + let removal = self.hot_log_node_removal(target, node_id); + match removal { + Some((revive, timestamp)) => self.apply_op_with(target, revive, timestamp, ApplyMode::Force), + None => self.restore_node_from_history(target, node_id), + } + } + + /// The most recent hot removal of a network, for reviving it in the working zone. `None` for the + /// snapshot, which stays a function of history alone. + fn hot_log_removal(&self, target: RegistryTarget, network_id: NetworkId) -> Option<(RegistryDelta, TimeStamp)> { + if target != RegistryTarget::Working { + return None; + } + + self.hot_log.iter().rev().find_map(|hot_op| match &hot_op.op { + RegistryDelta::RemoveNetwork { id, snapshot } if *id == network_id => Some((RegistryDelta::AddNetwork { id: *id, network: snapshot.clone() }, hot_op.timestamp)), + _ => None, + }) + } + + /// The node equivalent of [`hot_log_removal`](Self::hot_log_removal). + fn hot_log_node_removal(&self, target: RegistryTarget, node_id: NodeId) -> Option<(RegistryDelta, TimeStamp)> { + if target != RegistryTarget::Working { + return None; + } + + self.hot_log.iter().rev().find_map(|hot_op| match &hot_op.op { + RegistryDelta::RemoveNode { id, snapshot } if *id == node_id => Some((RegistryDelta::AddNode { id: *id, node: snapshot.clone() }, hot_op.timestamp)), + _ => None, + }) } fn ensure_network_exists(&mut self, target: RegistryTarget, network_id: NetworkId) -> Result<(), CrdtError> { - if !self.registry_ref(target).networks.contains_key(&network_id) { - self.restore_network_from_history(target, network_id)?; + if self.registry_ref(target).networks.contains_key(&network_id) { + return Ok(()); + } + // The hot log is peer-local, so reviving retired state from it would make it peer-dependent. + let removal = self.hot_log_removal(target, network_id); + match removal { + Some((revive, timestamp)) => self.apply_op_with(target, revive, timestamp, ApplyMode::Force), + None => self.restore_network_from_history(target, network_id), } - Ok(()) } /// Compute the inverse of `delta` against the registry named by `target`. Retirement passes diff --git a/document/graph-storage/src/history.rs b/document/graph-storage/src/history.rs index f091334e2c3..941aef16b46 100644 --- a/document/graph-storage/src/history.rs +++ b/document/graph-storage/src/history.rs @@ -8,7 +8,7 @@ //! construction. The only operation that introduces out-of-order deltas is [`merge`](History::merge), //! which re-sorts the combined set into the canonical order to restore the invariant. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use crate::{AttributesWrite, CrdtError, Delta, Rev, TimeStamp}; @@ -39,6 +39,11 @@ impl History { self.index.contains_key(&rev) } + /// Whether history holds a delta authored at `timestamp`. + pub fn contains_timestamp(&self, timestamp: TimeStamp) -> bool { + self.deltas.iter().any(|delta| delta.timestamp == timestamp) + } + pub fn len(&self) -> usize { self.deltas.len() } @@ -64,22 +69,12 @@ impl History { self.deltas.iter() } - /// Absorb `incoming` (dedup by `Rev`) and canonically re-sort the whole combined history. - /// - /// The sort is deterministic (topological, ties broken by `Rev`), so two peers that absorb the same - /// delta set produce byte-identical history, not merely two different valid orderings. This is the - /// history-convergence mechanism: arrival order is erased. Callers update the registry separately - /// (LWW apply is commutative, so the registry converges regardless of order). - pub fn merge(&mut self, incoming: impl IntoIterator) { - for delta in incoming { - self.push(delta); - } - self.canonical_sort(); - } - /// Re-order `deltas` into the canonical topological order and rebuild the index: parents precede /// children, and among deltas whose parents are all emitted the lowest `Rev` goes first. O(V + E). - fn canonical_sort(&mut self) { + /// + /// Deterministic: two peers that absorb the same delta set end up with byte-identical history, not two + /// different valid orderings. Arrival order is erased. + pub fn canonical_sort(&mut self) { // Unsatisfied in-history parent count per delta, plus reverse edges to decrement as parents emit. let mut pending_parents: HashMap = HashMap::with_capacity(self.deltas.len()); let mut children: HashMap> = HashMap::new(); @@ -119,6 +114,51 @@ impl History { } } + /// `roots` and everything reachable from them through all parent links. Unknown roots are skipped. + pub fn ancestors(&self, roots: impl IntoIterator) -> HashSet { + let mut seen = HashSet::new(); + let mut stack: Vec = roots.into_iter().filter(|rev| self.contains(*rev)).collect(); + while let Some(rev) = stack.pop() { + if !seen.insert(rev) { + continue; + } + if let Some(delta) = self.get(rev) { + stack.extend(delta.all_parents()); + } + } + seen + } + + pub fn is_ancestor(&self, ancestor: Rev, descendant: Rev) -> bool { + self.ancestors([descendant]).contains(&ancestor) + } + + /// `tip` plus the revs at first-parent distance 1, 2, 4, 8, ... behind it. Sent to a remote peer + /// so it can locate the divergence point within a factor of two of the true distance. + pub fn sample_chain(&self, tip: Rev) -> Vec { + let mut samples = vec![tip]; + let mut current = tip; + let mut distance = 0; + let mut next_sample_distance = 1; + + while let Some(parent) = self.get(current).and_then(|delta| delta.parent) { + current = parent; + distance += 1; + if distance == next_sample_distance { + samples.push(current); + next_sample_distance *= 2; + } + } + + samples + } + + /// Every delta not reachable from the `known` revs the remote peer reported, in topological order. + pub fn deltas_unknown_to(&self, known: impl IntoIterator) -> Vec<&Delta> { + let known = self.ancestors(known); + self.deltas.iter().filter(|delta| !known.contains(&delta.id)).collect() + } + /// The current tips: revs that no other delta lists as a parent (the divergent heads). A linear /// history has exactly one tip; concurrent branches have several. Sorted ascending for determinism. pub fn tips(&self) -> Vec { diff --git a/document/graph-storage/src/model.rs b/document/graph-storage/src/model.rs index 521eacc9532..bfd65b57482 100644 --- a/document/graph-storage/src/model.rs +++ b/document/graph-storage/src/model.rs @@ -47,6 +47,30 @@ impl Node { attributes_value_equal(&self.attributes, &other.attributes) } + /// A node in `network` whose `inputs` slots are unset, for a caller that fills them in with + /// `ChangeNodeInput`. The slot count is fixed at creation, since changing an input addresses a + /// slot by position. + pub fn new(network: NetworkId, implementation: Implementation, inputs: usize) -> Self { + // `to_runtime` reads this back as a `TaggedValue`, whose unit `None` variant encodes as this string. + // Pinned by `unset_input_slot_deserializes_as_tagged_value_none`. + let slot = InputSlot { + input: NodeInput::Value { + value: serde_json::Value::String("None".to_string()), + exposed: false, + }, + timestamp: TimeStamp::ORIGIN, + attributes: Attributes::new(), + }; + + Self { + implementation, + implementation_timestamp: TimeStamp::ORIGIN, + inputs: vec![slot; inputs], + attributes: Attributes::new(), + network, + } + } + #[cfg(test)] pub(crate) fn dummy() -> Self { Self { @@ -159,6 +183,19 @@ mod tests { } } + /// `to_runtime` has to read back what `Node::new` writes into an unset slot, so its literal must stay + /// in step with how `TaggedValue::None` serializes. + #[test] + fn unset_input_slot_deserializes_as_tagged_value_none() { + let node = Node::new(crate::ROOT_NETWORK, Implementation::Network(crate::ROOT_NETWORK), 1); + let NodeInput::Value { value, .. } = &node.inputs()[0].input else { + panic!("a fresh slot holds a value input"); + }; + + let tagged: graph_craft::document::value::TaggedValue = serde_json::from_value(value.clone()).expect("the unset slot value must deserialize"); + assert_eq!(tagged, graph_craft::document::value::TaggedValue::None); + } + /// A `SetExport(None)` truncation leaves a trailing empty slot. Such a network is value-equal to /// the same network without that slot, so the soak oracle doesn't false-report drift. #[test] diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index cea711596ad..99421ac569f 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -16,6 +16,9 @@ pub struct Session { /// leader-eligibility computation (lowest PeerId among peers whose tip matches the session max). #[expect(dead_code, reason = "Populated once heartbeat/leader-election transport lands; held now so the field and constructors are in place.")] remote_tips: HashMap, + /// The registry the runtime was last built from or staged to. Local diffs are taken against it, + /// so edits peers applied to the working registry in the meantime are not diffed away. + runtime_base: Option, } impl Session { @@ -36,14 +39,18 @@ impl Session { retired_snapshot: Registry::default(), history: History::new(), hot_log: Vec::new(), + retired: RetiredHotOps::default(), head: None, redo_stack: Vec::new(), clock: LamportClock::new(peer), peer, last_broadcast_rev: None, next_node_counter: 0, + next_hot_sequence: HotSequence::NONE, + refold_owed: false, }, remote_tips: HashMap::new(), + runtime_base: None, } } @@ -77,11 +84,21 @@ impl Session { resources: &graphene_resource::ResourceRegistry, ) -> Result<(Vec, from_runtime::RuntimeConversion), CommitError> { let conversion = Registry::convert_from_runtime(network, metadata, resources, self.document.peer)?; - let ops = crate::delta::compute_deltas(&self.document.working_registry, &conversion.registry); + let base = self.runtime_base.as_ref().unwrap_or(&self.document.working_registry); + let ops = crate::delta::compute_deltas(base, &conversion.registry); + + // The base moves only once the ops are staged, so a failure leaves the next diff covering them. let hot_ops = self.stage_ops(ops)?; + self.runtime_base = Some(conversion.registry.clone()); + Ok((hot_ops, conversion)) } + /// Record that the runtime now reflects the working registry, after the caller rebuilt it from there. + pub fn mark_runtime_current(&mut self) { + self.runtime_base = Some(self.document.working_registry.clone()); + } + /// Resolve each runtime `network_path` to its stable [`NetworkId`] for this document's peer, so the /// caller can key per-network, per-peer view state (`session.json`) by a stable id. Derived from the /// network structure alone; resources/declarations are irrelevant to the ids. @@ -140,8 +157,8 @@ impl Session { /// /// The peer's first contribution is preceded by a `RegisterPeer` op, so the device's /// `PeerId → UserId` mapping is established (and, under causal delivery, observed by other peers) - /// before any of its edits. A no-op batch doesn't register — registration rides a real edit. - fn stage_ops(&mut self, ops: impl IntoIterator) -> Result, CrdtError> { + /// before any of its edits. A no-op batch doesn't register, since registration rides a real edit. + pub fn stage_ops(&mut self, ops: impl IntoIterator) -> Result, CrdtError> { let mut pending: Vec = ops.into_iter().collect(); if pending.is_empty() { return Ok(Vec::new()); @@ -154,11 +171,16 @@ impl Session { let mut staged = Vec::with_capacity(pending.len()); for op in pending { + // The counter advances only once the op is in the log, so a failure leaves no gap in the run. + let sequence = self.document.next_hot_sequence.next(); let hot_op = HotOp { op, timestamp: self.document.clock.tick(), + sequence, }; self.document.apply_hot_op(hot_op.clone())?; + + self.document.next_hot_sequence = sequence; staged.push(hot_op); } Ok(staged) @@ -174,11 +196,18 @@ impl Session { /// `idempotent`: pass `true` when the snapshot already reflects the op (retirement of an already- /// applied hot op) so duplicate structural inserts no-op rather than error. fn commit_ops(&mut self, ops: impl IntoIterator, idempotent: bool) -> Result, CrdtError> { + self.commit_ops_authored_at(ops.into_iter().map(|op| (op, None)), idempotent) + } + + /// [`commit_ops`](Self::commit_ops) for ops authored elsewhere, paired with the timestamp they were + /// authored at (`None` to mint one). The delta's timestamp is what the registry resolves LWW on, so + /// retirement passes the op's own; a coarsened delta passes the newest it fuses. + fn commit_ops_authored_at(&mut self, ops: impl IntoIterator)>, idempotent: bool) -> Result, CrdtError> { let target = RegistryTarget::Snapshot; let ops = ops.into_iter(); let mut produced = Vec::with_capacity(ops.size_hint().0); - for op in ops { + for (op, authored_at) in ops { // A new edit abandons any undone-forward branch: those revs stay in the DAG but are no // longer reachable via redo. (Mirrors the legacy editor clearing its redo history on // commit.) Done on the first real op so a no-op commit doesn't silently disable redo. @@ -186,10 +215,12 @@ impl Session { self.document.redo_stack.clear(); } + // The reverse must read the pre-op value of a target that may have been concurrently removed. + self.document.ensure_referenced_exist(target, &op)?; let reverse = self.document.compute_reverse_delta(target, &op)?; - let timestamp = self.document.clock.tick(); + let timestamp = authored_at.unwrap_or_else(|| self.document.clock.tick()); let parent = self.document.head; - let author = self.document.peer; + let author = timestamp.peer; let delta = Delta::new(parent, author, timestamp, op, reverse); let rev = delta.id; @@ -227,14 +258,18 @@ impl Session { working_registry: registry, history: History::from_ordered(history), hot_log: Vec::new(), + retired: RetiredHotOps::default(), head, redo_stack, clock, peer, last_broadcast_rev: None, next_node_counter, + next_hot_sequence: HotSequence::NONE, + refold_owed: false, }, remote_tips: HashMap::new(), + runtime_base: None, } } @@ -256,9 +291,46 @@ impl Session { Ok(session) } - /// Apply a hot op without going through the broadcast stream. + /// Apply a hot op from another peer. Idempotent, and a no-op for one history already covers: a late + /// delivery must not put a retired op back in the hot log. pub fn apply_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> { - self.document.apply_hot_op(hot_op) + self.document.replay_hot_op(hot_op) + } + + /// The hot ops a `retire(up_to)` call would drain. Sent with the deltas for peers to drop exactly + /// these; the cutoff alone doesn't transfer, a lagging op can arrive below it afterwards. + pub fn hot_ops_up_to(&self, up_to: TimeStamp) -> Vec { + self.document.hot_log.iter().filter(|hot_op| hot_op.timestamp <= up_to).map(HotOp::id).collect() + } + + /// Drop hot ops another peer has retired without retiring them locally. Refolds like + /// [`absorb_retired_marks`](Self::absorb_retired_marks). + pub fn discard_hot_ops(&mut self, retired: &[HotOpId]) -> Result<(), CrdtError> { + let before = self.document.hot_log.len(); + self.document.hot_log.retain(|hot_op| !retired.contains(&hot_op.id())); + // How a peer that did not retire these learns they are in history now. + self.document.mark_retired(retired.iter().copied()); + + if before != self.document.hot_log.len() { + self.refold_registries()?; + } + + Ok(()) + } + + /// Which hot ops history already covers, for a peer catching up. See [`RetiredHotOps`]. + pub fn retired_marks(&self) -> &RetiredHotOps { + &self.document.retired + } + + /// See [`RetiredHotOps::absorb`]. A drop owes a refold, since the working registry holds the op's + /// effect in hot-log order where the snapshot holds it in canonical order. + pub fn absorb_retired_marks(&mut self, remote: &RetiredHotOps) -> Result<(), CrdtError> { + if self.document.absorb_retired(remote) { + self.refold_registries()?; + } + + Ok(()) } /// Replay a persisted hot op. Idempotent on structural ops, suitable for crash recovery @@ -267,55 +339,150 @@ impl Session { self.document.replay_hot_op(hot_op) } - /// Integrate `incoming` retired deltas from another branch and emit a [`RegistryDelta::Merge`] - /// joining the resulting tips, returning the new merge `Rev` (or `None` if `incoming` adds nothing). - /// Applies each incoming op to the registry, then hands the set to [`History::merge`]. Incoming - /// deltas must arrive in causal order. - pub fn merge(&mut self, incoming: impl IntoIterator) -> Result, CrdtError> { - let mut absorbed: Vec = Vec::new(); + /// The retired snapshot as canonical history alone produces it, independent of arrival order and the + /// hot log. Folded on a clone, over `head`'s ancestry only: an undone delta stays in the DAG for redo + /// to find, and folding all of history would restore work the user undid. + /// + /// The clone's history grows with the replay, as it did when each delta first landed. Resurrection + /// takes the last removal in canonical order, and with the whole history in view that can be one past + /// the replay position, whose snapshot places the entity in a network the replay has not created yet. + pub fn snapshot_from_history(&self) -> Result { + let reachable = self.document.history.ancestors(self.document.head); + + let mut folded = self.clone(); + folded.document.retired_snapshot = Registry::default(); + folded.document.history = History::new(); + + for delta in self.document.history.iter().filter(|delta| reachable.contains(&delta.id)) { + folded.document.apply_op_with(RegistryTarget::Snapshot, delta.kind.clone(), delta.timestamp, ApplyMode::Idempotent)?; + folded.document.history.push(delta.clone()); + } + + Ok(folded.document.retired_snapshot) + } + + /// Rebuild both registries from canonical history, then re-layer the hot tail. A full replay; callers + /// check it is needed first. + fn refold_registries(&mut self) -> Result<(), CrdtError> { + // Cleared only on the way out. The deltas are in history either way, and a failure here leaves the + // registries derived from an older one until something retries. + self.document.refold_owed = true; + self.document.retired_snapshot = self.snapshot_from_history()?; + + // One left unapplicable is kept, not dropped: a later delta or hot op can still supply its referent. + self.document.working_registry = self.document.retired_snapshot.clone(); + let mut failure = None; + for hot_op in std::mem::take(&mut self.document.hot_log) { + if let Err(error) = self.document.replay_hot_op(hot_op.clone()) { + self.document.hot_log.push(hot_op); + failure = failure.or(Some(error)); + } + } + + if let Some(error) = failure { + return Err(error); + } + + self.document.refold_owed = false; + Ok(()) + } + + /// Integrate `incoming` retired deltas (in causal order) from another peer. Moves `head` forward + /// without a new delta when the incoming history extends it, otherwise joins `head` and the + /// incoming tips with a [`RegistryDelta::Merge`]. + pub fn merge(&mut self, incoming: impl IntoIterator) -> Result { + let length_before = self.document.history.len(); + + // Each delta enters history before the next is applied, so a later delta in the batch that + // targets something an earlier one removed can resurrect it. + let mut absorbed_ids = HashSet::new(); + let mut arrival_order = Vec::new(); for delta in incoming { if self.document.history.contains(delta.id) { continue; } + arrival_order.push(delta.id); self.document.apply_op_idempotent(delta.kind.clone(), delta.timestamp)?; - absorbed.push(delta); + // Working is snapshot plus hot tail, so a delta lands on both; cloning would promote hot ops. + self.document.apply_op_with(RegistryTarget::Snapshot, delta.kind.clone(), delta.timestamp, ApplyMode::Idempotent)?; + absorbed_ids.insert(delta.id); + self.document.history.push(delta); } - if absorbed.is_empty() { - return Ok(None); + if absorbed_ids.is_empty() { + // Nothing new, but a refold left owed by an earlier failure is still the only way the registries + // catch up with the history they are derived from. + if self.document.refold_owed { + self.refold_registries()?; + } + + return Ok(MergeOutcome::NoOp); } + self.document.history.canonical_sort(); + + // The loop folds in arrival order, and concurrent ops do not all commute: a remove and a change to + // the same target resolve by whichever lands second. + let folded_in_canonical_order = self.document.history.iter().skip(length_before).map(|delta| delta.id).eq(arrival_order.iter().copied()); + + let history = &self.document.history; + let mut candidates: Vec = history.tips().into_iter().filter(|tip| absorbed_ids.contains(tip)).collect(); + candidates.extend(self.document.head); + let is_dominated = |candidate: Rev| candidates.iter().any(|&other| other != candidate && history.is_ancestor(candidate, other)); + let parents: Vec = candidates.iter().copied().filter(|&candidate| !is_dominated(candidate)).collect(); + + let outcome = match parents.as_slice() { + [tip] => MergeOutcome::FastForward(*tip), + _ => { + let timestamp = self.document.clock.tick(); + let merge = Delta::merge(parents, self.document.peer, timestamp); + let merge_rev = merge.id; + // Its parents are tips, so `push` keeps the canonical order without a re-sort. + self.document.history.push(merge); + MergeOutcome::Merged(merge_rev) + } + }; + self.document.head = outcome.head(); - self.document.history.merge(absorbed); - let tips = self.document.history.tips(); - let timestamp = self.document.clock.tick(); - let merge = Delta::merge(tips, self.document.peer, timestamp); - let merge_rev = merge.id; - // The merge's parents are the current tips, so it sorts last: `push` preserves the canonical - // order without re-sorting the whole history. - self.document.history.push(merge); - self.document.head = Some(merge_rev); + // After `head`, since the fold is over its ancestry. + if !folded_in_canonical_order || self.document.refold_owed { + self.refold_registries()?; + } - // Merge runs with an empty hot log; keep the retired snapshot in step with the working registry. - self.document.retired_snapshot = self.document.working_registry.clone(); - Ok(Some(merge_rev)) + Ok(outcome) } - /// Promote hot ops with timestamp `≤ up_to` into retired deltas, re-applied with fresh - /// retirement timestamps so LWW arms bump field timestamps to `T_retire`. + /// Promote hot ops with timestamp `≤ up_to` into retired deltas, each keeping the timestamp it was + /// authored at. /// /// Today: one retired delta per hot op. Coarsening is a future step. pub fn retire(&mut self, up_to: TimeStamp) -> Result, CrdtError> { + // Hot-log order is causal, so the deltas commit in an order their references survive. let mut drained = Vec::new(); let mut remaining = Vec::with_capacity(self.document.hot_log.len()); + + // Retiring by timestamp reorders against the hot-log order the working registry applied: past a + // skipped op, everything drained was applied here before it. + let mut skipped = false; + let mut reordered = false; for hot_op in self.document.hot_log.drain(..) { if hot_op.timestamp <= up_to { + reordered |= skipped; drained.push(hot_op); } else { + skipped = true; remaining.push(hot_op); } } self.document.hot_log = remaining; + self.document.mark_retired(drained.iter().map(HotOp::id)); + + let revs = self.commit_ops_authored_at(drained.into_iter().map(|hot_op| (hot_op.op, Some(hot_op.timestamp))), true)?; + + // Structural ops have no timestamp to arbitrate, so the reorder changes the outcome. + if reordered { + self.refold_registries()?; + } - self.commit_ops(drained.into_iter().map(|hot_op| hot_op.op), true) + Ok(revs) } /// Mark a retired delta as the end of a user interaction, so the undo cursor treats it as a checkpoint. @@ -369,16 +536,16 @@ impl Session { !self.document.redo_stack.is_empty() } - /// Silent-zone undo of one *interaction*: revert deltas walking `head` back along first-parents until - /// it reaches the previous interaction boundary (a delta marked `interaction_end`) or the empty root. One - /// interaction spans several deltas (one `commit_from_runtime` batch), so undo reverts the whole run, - /// not a single delta — matching the legacy per-interaction undo granularity. The undone interaction's - /// `head` rev is pushed onto the redo stack. Reflog semantics: the DAG is never rewritten. + /// Silent-zone undo of one *interaction*: revert deltas back along first-parents until `head` reaches + /// the previous `interaction_end` boundary or the root. An interaction is a whole `commit_from_runtime` + /// batch, so the run reverts together. Its `head` rev goes on the redo stack; the DAG is not rewritten. pub fn undo(&mut self) -> Result { if !self.can_undo() { return Err(CrdtError::NothingToUndo); } let checkpoint = self.document.head.ok_or(CrdtError::NothingToUndo)?; + // The caller rebuilds the runtime from the rewound registry; until then diff against it directly. + self.runtime_base = None; // Revert this interaction's last delta, then keep going back until `head` rests on the previous // interaction's boundary (its `interaction_end` delta) or the root. @@ -387,7 +554,10 @@ impl Session { let delta = self.document.history.get(rev).ok_or(CrdtError::NotFoundInHistory(rev))?.clone(); let parent = delta.parent; - self.document.revert_delta(RegistryTarget::Working, delta)?; + self.document.revert_delta(RegistryTarget::Working, delta.clone())?; + // Working is snapshot plus hot tail, so rewind both; copying would promote unretired hot ops. + self.document.revert_delta(RegistryTarget::Snapshot, delta)?; + self.document.head = parent; match parent { @@ -397,9 +567,6 @@ impl Session { } } - // Undo runs with an empty hot log, so keep the retired snapshot in lockstep with the rewound - // working registry (the next interaction's reverses are computed against it). - self.document.retired_snapshot = self.document.working_registry.clone(); self.document.redo_stack.push(checkpoint); Ok(checkpoint) } @@ -409,6 +576,7 @@ impl Session { /// parents back from the checkpoint to `head` (the chain is linear in the silent solo zone). pub fn redo(&mut self) -> Result { let checkpoint = self.document.redo_stack.pop().ok_or(CrdtError::NothingToRedo)?; + self.runtime_base = None; let mut forward = Vec::new(); let mut cursor = Some(checkpoint); @@ -423,11 +591,11 @@ impl Session { // at the same timestamp. Symmetric with `revert_delta`. for delta in forward.into_iter().rev() { self.document.force_apply_op(delta.kind.clone(), delta.timestamp)?; + // Both zones move together, for the same reason `undo` rewinds both. + self.document.apply_op_with(RegistryTarget::Snapshot, delta.kind.clone(), delta.timestamp, ApplyMode::Force)?; } self.document.head = Some(checkpoint); - // Redo runs with an empty hot log; keep the retired snapshot in lockstep with the working registry. - self.document.retired_snapshot = self.document.working_registry.clone(); Ok(checkpoint) } @@ -447,6 +615,17 @@ impl Session { self.document.history.iter() } + /// Revs sampled at exponentially growing distances behind `head`, for a remote peer to locate what + /// this session is missing (see [`History::sample_chain`]). + pub fn known_revs(&self) -> Vec { + self.document.head.map(|head| self.document.history.sample_chain(head)).unwrap_or_default() + } + + /// Retired deltas not reachable from `known`, in replay order. + pub fn deltas_unknown_to(&self, known: impl IntoIterator) -> Vec<&Delta> { + self.document.history.deltas_unknown_to(known) + } + /// The retired delta for `rev`, or `None` if it isn't in history. O(1) lookup, for callers that /// already hold the revs they want (e.g. persisting a freshly-retired batch) and don't need a scan. pub fn delta(&self, rev: Rev) -> Option<&Delta> { @@ -553,6 +732,17 @@ impl Session { pub fn next_node_counter(&self) -> u64 { self.document.next_node_counter } + + /// How many hot ops this peer has authored. Carried across a reload; no sequence is spent twice. + pub fn next_hot_sequence(&self) -> HotSequence { + self.document.next_hot_sequence + } + + /// Restore the authored-op count after a load, keeping a fresh op off a spent sequence. Raises only: + /// replaying a persisted hot log afterwards cannot lower it. + pub fn restore_hot_sequence(&mut self, sequence: HotSequence) { + self.document.next_hot_sequence = self.document.next_hot_sequence.max(sequence); + } } /// Errors from `Session::commit_from_runtime`. @@ -572,12 +762,147 @@ impl Default for Session { } } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum MergeOutcome { + NoOp, + FastForward(Rev), + Merged(Rev), +} + +impl MergeOutcome { + /// The head after the merge, or `None` when nothing changed. + pub fn head(self) -> Option { + match self { + Self::NoOp => None, + Self::FastForward(rev) | Self::Merged(rev) => Some(rev), + } + } +} + /// One live op in the hot zone. Carries only enough to drive live LWW; no parents (transient), /// no Rev (not content-addressed in the durable DAG). GC'd at retirement. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct HotOp { pub op: RegistryDelta, pub timestamp: TimeStamp, + /// Position in its author's run, from 1 with no gaps. A watermark over it therefore means a contiguous + /// prefix; one over the Lamport counter cannot, since that skips on observing a higher remote stamp. + pub sequence: HotSequence, +} + +impl HotOp { + /// Identifies the op for retirement, which tracks a contiguous prefix per author. + pub fn id(&self) -> HotOpId { + HotOpId { + peer: self.timestamp.peer, + sequence: self.sequence, + } + } +} + +/// Position in one author's run of hot ops, counting from 1 with no gaps. Kept apart from the Lamport +/// counter, which skips on observing a higher remote timestamp and so cannot bound a contiguous prefix. +#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Default, Serialize, Deserialize)] +#[serde(transparent)] +pub struct HotSequence(pub u64); + +impl HotSequence { + /// Before this author has written anything, covering no op. + pub const NONE: Self = Self(0); + + /// The next position in the run. + pub fn next(self) -> Self { + Self(self.0 + 1) + } +} + +/// One hot op's author and position in its run. Retirement names promoted ops by these, which tells a +/// receiver which prefix history covers. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +pub struct HotOpId { + pub peer: PeerId, + pub sequence: HotSequence, +} + +/// Which hot ops history already covers: `retired_up_to` is each author's gap-free retired prefix, and +/// `unretired` the retired ops past it. Replicated, letting a peer catching up separate an op already in +/// history from one still owed to it. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RetiredHotOps { + pub retired_up_to: HashMap, + /// Retired ops sitting past their author's prefix, as inclusive runs. A gap that never fills pins the + /// prefix and everything later from that author lands here, so runs keep this proportional to the + /// number of gaps instead of the number of ops. + pub retired_beyond: HashMap>, +} + +/// Sort and coalesce runs, joining any that touch or abut. +fn coalesce(runs: &mut Vec<(HotSequence, HotSequence)>) { + runs.sort_unstable(); + + let mut merged: Vec<(HotSequence, HotSequence)> = Vec::with_capacity(runs.len()); + for &(start, end) in runs.iter() { + match merged.last_mut() { + Some((_, last_end)) if start <= last_end.next() => *last_end = (*last_end).max(end), + _ => merged.push((start, end)), + } + } + + *runs = merged; +} + +impl RetiredHotOps { + /// Whether history already holds this hot op. + pub fn covers(&self, id: HotOpId) -> bool { + if self.retired_up_to.get(&id.peer).is_some_and(|&through| id.sequence <= through) { + return true; + } + + self.retired_beyond + .get(&id.peer) + .is_some_and(|runs| runs.iter().any(|&(start, end)| id.sequence >= start && id.sequence <= end)) + } + + /// Take on `remote`'s coverage as well as this one's. + pub fn absorb(&mut self, remote: &Self) { + for (&peer, &remote_through) in &remote.retired_up_to { + let through = self.retired_up_to.entry(peer).or_default(); + *through = (*through).max(remote_through); + } + for (&peer, runs) in &remote.retired_beyond { + self.retired_beyond.entry(peer).or_default().extend(runs.iter().copied()); + } + + self.compact(); + } + + /// Record newly retired ops. + pub fn extend(&mut self, retired: impl IntoIterator) { + for id in retired { + self.retired_beyond.entry(id.peer).or_default().push((id.sequence, id.sequence)); + } + + self.compact(); + } + + /// Fold runs that continue their author's prefix into `retired_up_to`, leaving only those past a gap. + fn compact(&mut self) { + for (&peer, runs) in &mut self.retired_beyond { + coalesce(runs); + + let mut through = self.retired_up_to.get(&peer).copied().unwrap_or(HotSequence::NONE); + while runs.first().is_some_and(|&(start, _)| start <= through.next()) { + let (_, end) = runs.remove(0); + through = through.max(end); + } + + if through != HotSequence::NONE { + self.retired_up_to.insert(peer, through); + } + } + + self.retired_beyond.retain(|_, runs| !runs.is_empty()); + } } #[derive(Debug, thiserror::Error)] diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index 78fd8f082ae..d45b31edeb3 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -40,6 +40,7 @@ fn apply_hot_op_advances_clock_past_observed_timestamp() { let hot_op = HotOp { op: remove_node_op(NodeId(99)), timestamp: observed, + sequence: crate::HotSequence(1), }; document.apply_hot_op(hot_op).expect("RemoveNode on absent node is a no-op, not an error"); @@ -182,9 +183,10 @@ fn merge_converges_to_identical_history() { // Cross-merge: feed each peer the other's full delta set. The shared base dedups by `Rev`. let deltas_a = session_a.cloned_deltas(); let deltas_b = session_b.cloned_deltas(); - let merge_a = session_a.merge(deltas_b).expect("merge into A failed").expect("A produced a merge"); - let merge_b = session_b.merge(deltas_a).expect("merge into B failed").expect("B produced a merge"); + let merge_a = session_a.merge(deltas_b).expect("merge into A failed"); + let merge_b = session_b.merge(deltas_a).expect("merge into B failed"); + assert!(matches!(merge_a, crate::MergeOutcome::Merged(_)), "divergent branches must produce a merge delta"); assert_eq!(merge_a, merge_b, "same tips must mint the identical parent-set-addressed merge commit"); let order_a: Vec = session_a.history().map(|d| d.id).collect(); @@ -193,6 +195,29 @@ fn merge_converges_to_identical_history() { assert_eq!(session_a.head_rev(), session_b.head_rev(), "both peers land on the same merge head"); } +/// A peer whose history is a prefix of the incoming one moves its head forward without minting a +/// merge delta, and the negotiated transfer sends only the deltas past the sampled known revs. +#[test] +fn merge_fast_forwards_a_prefix_history() { + let mut session_a = Session::with_peer(PeerId(1)); + session_a.commit_op_for_test(set_document_attribute("compute::base", 0)).expect("base commit"); + let mut session_b = session_a.clone(); + + for value in 1..=5 { + session_b.commit_op_for_test(set_document_attribute("compute::b", value)).expect("B edit"); + } + + let known = session_a.known_revs(); + let missing: Vec<_> = session_b.deltas_unknown_to(known).into_iter().cloned().collect(); + assert_eq!(missing.len(), 5, "only B's new commits are unknown to A"); + + let outcome = session_a.merge(missing).expect("merge failed"); + assert_eq!(outcome, crate::MergeOutcome::FastForward(session_b.head_rev().unwrap())); + assert_eq!(session_a.history().count(), session_b.history().count(), "no merge delta was added"); + + assert_eq!(session_b.merge(session_a.cloned_deltas()).expect("merge failed"), crate::MergeOutcome::NoOp); +} + /// Resurrection must reach into a merged-in branch: a network added then removed on the other peer's /// branch lives only under the merge's secondary parent, so a `SetNetworkExport` targeting it after /// the merge can only restore it by traversing all ancestors (not the primary-parent chain). @@ -880,3 +905,417 @@ fn add_node_rev_is_independent_of_attribute_insertion_order() { assert_eq!(delta_forward.id, delta_reversed.id, "Rev must not depend on attribute insertion order"); } + +/// Commit a retired delta and mirror it onto the working registry; `commit_op_for_test` alone only +/// advances the snapshot zone. +fn commit_retired(session: &mut Session, op: RegistryDelta) { + let before = session.history().count(); + session.commit_op_for_test(op).expect("commit failed"); + + for delta in session.cloned_deltas().into_iter().skip(before) { + session.document.apply_op_idempotent(delta.kind, delta.timestamp).expect("mirroring onto the working registry"); + } +} + +/// Merge `deltas` into a copy of `base` in the given order, reporting node 7's surviving `tint`. +fn merge_order_outcome(base: &Session, deltas: &[Delta], order: [usize; 2]) -> Option<(serde_json::Value, TimeStamp)> { + let mut session = base.clone(); + let ordered: Vec = order.iter().map(|&index| deltas[index].clone()).collect(); + session.merge(ordered).expect("merge failed"); + + session + .retired_registry() + .node_instances + .get(&NodeId(7)) + .and_then(|node| node.attributes.get("tint")) + .map(|value| (value.value.clone(), value.timestamp)) +} + +/// The raw ops are arrival-order dependent: change-first leaves the node gone, remove-first resurrects +/// it from the removal's snapshot. Merge has to fold both orders onto the same retired registry. +#[test] +fn a_concurrent_remove_and_attribute_change_commute() { + let network_id = NetworkId(5); + let node = Node::new(network_id, crate::Implementation::Network(network_id), 0); + + // Shared base holding node 7, then the two peers diverge. + let mut base = Session::with_peer(PeerId(1)); + commit_retired( + &mut base, + RegistryDelta::AddNetwork { + id: network_id, + network: Network::default(), + }, + ); + commit_retired(&mut base, RegistryDelta::AddNode { id: NodeId(7), node: node.clone() }); + + let mut remover = base.clone(); + commit_retired(&mut remover, RegistryDelta::RemoveNode { id: NodeId(7), snapshot: node }); + + let mut changer = Session::with_peer(PeerId(2)); + changer.merge(base.cloned_deltas()).expect("changer adopts the base"); + commit_retired( + &mut changer, + RegistryDelta::ChangeNodeAttribute { + id: NodeId(7), + delta: crate::AttributeDelta { + key: "tint".to_string(), + value: Some(serde_json::json!(75)), + }, + }, + ); + + // One delta from each branch, neither an ancestor of the other. + let removal = remover.cloned_deltas().into_iter().last().expect("removal delta"); + let change = changer.cloned_deltas().into_iter().last().expect("change delta"); + let deltas = [removal, change]; + + let removal_first = merge_order_outcome(&base, &deltas, [0, 1]); + let change_first = merge_order_outcome(&base, &deltas, [1, 0]); + + assert_eq!(removal_first, change_first, "the retired registry must not depend on delta arrival order"); +} + +/// An op can retire before an earlier one from its author arrives, so the marks cover it out of order. +/// The prefix absorbs it once the gap fills, which bounds the exception set. +#[test] +fn retired_marks_cover_a_gap_and_compact_once_it_fills() { + let author = PeerId(1); + let id = |sequence: u64| crate::HotOpId { + peer: author, + sequence: crate::HotSequence(sequence), + }; + + let mut marks = crate::RetiredHotOps::default(); + marks.extend([id(1), id(2)]); + assert_eq!(marks.retired_up_to.get(&author), Some(&crate::HotSequence(2)), "a contiguous run folds straight into the prefix"); + assert!(marks.retired_beyond.is_empty()); + + // Sequence 3 never arrived, so 4 retires above the prefix rather than extending it. + marks.extend([id(4)]); + assert_eq!(marks.retired_up_to.get(&author), Some(&crate::HotSequence(2))); + assert!(marks.covers(id(4)), "an op past the gap is still recognized as retired"); + assert!(!marks.covers(id(3)), "the missing op is not claimed"); + + // The gap fills, so the prefix swallows both it and the exception behind it. + marks.extend([id(3)]); + assert_eq!(marks.retired_up_to.get(&author), Some(&crate::HotSequence(4))); + assert!(marks.retired_beyond.is_empty(), "the exception set empties once the prefix reaches it"); +} + +/// Marks merge by union, so adopting a peer's wholesale must not lose local coverage. +#[test] +fn absorbing_marks_keeps_both_sides_coverage() { + let author = PeerId(1); + let id = |sequence: u64| crate::HotOpId { + peer: author, + sequence: crate::HotSequence(sequence), + }; + + let mut local = crate::RetiredHotOps::default(); + local.extend([id(1), id(4)]); + + let mut remote = crate::RetiredHotOps::default(); + remote.extend([id(1), id(2), id(3), id(5)]); + + local.absorb(&remote); + + // The union is 1..=5 contiguous, so it all collapses into the prefix. + assert_eq!(local.retired_up_to.get(&author), Some(&crate::HotSequence(5))); + assert!(local.retired_beyond.is_empty()); + for sequence in 1..=5 { + assert!(local.covers(id(sequence)), "sequence {sequence} must stay covered"); + } +} + +/// Undo rewinds retired history and must leave the hot tail alone: copying the working registry onto +/// the snapshot promotes unretired work. Hot ops exist between staging and retirement even solo. +#[test] +fn undo_does_not_promote_hot_ops_into_the_retired_snapshot() { + let mut session = Session::with_peer(PeerId(1)); + + // Two retired interactions, so the second is undoable. + commit_retired(&mut session, set_document_attribute("first", 1)); + let boundary = session.history().last().expect("a committed delta").id; + session.mark_interaction_end(boundary); + commit_retired(&mut session, set_document_attribute("second", 2)); + + // Unretired live work sitting on top, on a key no retired delta touches. + session.stage_ops([set_document_attribute("hot", 3)]).expect("stage"); + assert!(session.registry().attributes.contains_key("hot"), "the hot op must be in the working registry"); + assert!(!session.retired_registry().attributes.contains_key("hot"), "and must not be in the snapshot"); + + session.undo().expect("undo"); + + assert!(!session.retired_registry().attributes.contains_key("second"), "undo must rewind the retired snapshot"); + assert!( + !session.retired_registry().attributes.contains_key("hot"), + "undo promoted an unretired hot op into the retired snapshot" + ); +} + +/// Redo puts the interaction back on both zones, so the pair does not drift the other way. +#[test] +fn redo_restores_the_retired_snapshot_without_the_hot_tail() { + let mut session = Session::with_peer(PeerId(1)); + + commit_retired(&mut session, set_document_attribute("first", 1)); + let boundary = session.history().last().expect("a committed delta").id; + session.mark_interaction_end(boundary); + commit_retired(&mut session, set_document_attribute("second", 2)); + + session.stage_ops([set_document_attribute("hot", 3)]).expect("stage"); + session.undo().expect("undo"); + session.redo().expect("redo"); + + assert!(session.retired_registry().attributes.contains_key("second"), "redo must restore the retired delta"); + assert!( + !session.retired_registry().attributes.contains_key("hot"), + "redo promoted an unretired hot op into the retired snapshot" + ); + assert!(session.registry().attributes.contains_key("hot"), "the hot op must survive an undo/redo round trip"); +} + +/// Silent undo emits nothing, so it is only legal while a commit is unpublished. Once peers hold it, a +/// rewind would diverge from them for good. +#[test] +fn publishing_a_commit_disables_silent_undo() { + let mut session = Session::with_peer(PeerId(1)); + + commit_retired(&mut session, set_document_attribute("first", 1)); + let boundary = session.history().last().expect("a committed delta").id; + session.mark_interaction_end(boundary); + commit_retired(&mut session, set_document_attribute("second", 2)); + + let head = session.head_rev().expect("a head"); + assert!(session.can_undo(), "an unpublished interaction is undoable"); + + session.publish_up_to(head); + + assert!(!session.can_undo(), "a published interaction must not be silently rewound"); +} + +/// An undone delta stays in the DAG for redo, outside `head`'s ancestry. A fold that ignores that +/// restores work the user undid. +#[test] +fn snapshot_from_history_ignores_undone_deltas() { + let mut session = Session::with_peer(PeerId(1)); + + commit_retired(&mut session, set_document_attribute("first", 1)); + let boundary = session.history().last().expect("a committed delta").id; + session.mark_interaction_end(boundary); + commit_retired(&mut session, set_document_attribute("second", 2)); + + session.undo().expect("undo"); + assert!(!session.retired_registry().attributes.contains_key("second"), "undo rewound the snapshot"); + + let folded = session.snapshot_from_history().expect("fold"); + + assert!(folded.attributes.contains_key("first"), "history still holds the kept interaction"); + assert!(!folded.attributes.contains_key("second"), "a fold restored an undone delta"); +} + +/// Resurrection takes the last removal of the entity in canonical order. When the deltas first landed +/// that was the latest one so far, but a fold over finished history also sees the removals after the +/// replay position, and one of those can carry the node in a network the replay has not created yet +/// (simulation seed 3588534). +#[test] +fn snapshot_from_history_resurrects_from_the_replay_position() { + let node_id = NodeId(3); + let first_network = NetworkId(1); + let later_network = NetworkId(3); + let node_in = |network: NetworkId| Node { network, ..Node::dummy() }; + + let mut session = Session::with_peer(PeerId(1)); + commit_retired( + &mut session, + RegistryDelta::AddNetwork { + id: first_network, + network: Network::default(), + }, + ); + commit_retired( + &mut session, + RegistryDelta::AddNode { + id: node_id, + node: node_in(first_network), + }, + ); + commit_retired( + &mut session, + RegistryDelta::RemoveNode { + id: node_id, + snapshot: node_in(first_network), + }, + ); + // Resurrects the node from the removal above, the only one so far. + commit_retired( + &mut session, + RegistryDelta::SetNetworkExport { + id: first_network, + index: 0, + export: Some(crate::NodeInput::Node { id: node_id, index: 0 }), + }, + ); + commit_retired( + &mut session, + RegistryDelta::AddNetwork { + id: later_network, + network: Network::default(), + }, + ); + // A later removal whose snapshot places the node in the network created after the export. + commit_retired( + &mut session, + RegistryDelta::RemoveNode { + id: node_id, + snapshot: node_in(later_network), + }, + ); + + let folded = session.snapshot_from_history().expect("a fold must resurrect from what the replay has seen, not from a later removal"); + + assert_eq!(&folded, session.retired_registry(), "the fold must reproduce the snapshot the deltas built as they landed"); +} + +/// A failed refold leaves the registries on an older history, and the deltas are already absorbed, so +/// re-merging them is a no-op. Without a retry the registries never catch up. +#[test] +fn an_owed_refold_is_retried_by_a_merge_that_absorbs_nothing() { + let mut session = Session::with_peer(PeerId(1)); + commit_retired(&mut session, set_document_attribute("first", 1)); + + // Stand in for a refold that failed partway: history moved on, the snapshot did not. + session.document.retired_snapshot = crate::Registry::default(); + session.document.refold_owed = true; + + let outcome = session.merge(Vec::new()).expect("a merge absorbing nothing still settles the refold"); + + assert_eq!(outcome, crate::MergeOutcome::NoOp); + assert!(!session.document.refold_owed, "the refold must not stay owed"); + assert!( + session.retired_registry().attributes.contains_key("first"), + "the retried refold must bring the snapshot back to its history" + ); +} + +/// The newtype is `#[serde(transparent)]`, so persisted state and the wire carry a bare number and the +/// field's encoding is unchanged. +#[test] +fn hot_sequence_serializes_as_a_bare_number() { + let encoded = serde_json::to_string(&crate::HotSequence(7)).expect("serialize"); + + assert_eq!(encoded, "7"); + assert_eq!(serde_json::from_str::("7").expect("deserialize"), crate::HotSequence(7)); +} + +/// Retirement drains the hot log, so a retirer that never received an author's earlier op cannot +/// retire it. The later op still retires, landing beyond the author's prefix instead of extending it. +#[test] +fn retiring_over_a_gap_lands_beyond_the_prefix() { + let author = PeerId(2); + let mut host = Session::with_peer(PeerId(1)); + + // Only the author's second op reaches this peer; the first was lost with the link that carried it. + let second = HotOp { + op: set_document_attribute("late", 1), + timestamp: TimeStamp { counter: 9, peer: author }, + sequence: crate::HotSequence(2), + }; + host.apply_hot_op(second.clone()).expect("apply"); + + host.retire(second.timestamp).expect("retire"); + + let retired = host.retired_marks(); + let prefix = retired.retired_up_to.get(&author).copied().unwrap_or(crate::HotSequence::NONE); + assert_eq!(prefix, crate::HotSequence::NONE, "sequence 1 never arrived, so the prefix cannot move"); + assert!(retired.retired_beyond.contains_key(&author), "the retired op has to be recorded past the gap"); + assert!(retired.covers(second.id()), "and still count as retired"); +} + +/// A gap that never fills pins the prefix, so every later op from that author lands past it. Stored as +/// runs, that stays one entry however many ops follow. +#[test] +fn a_permanent_gap_accumulates_every_later_op() { + let author = PeerId(2); + let id = |sequence: u64| crate::HotOpId { + peer: author, + sequence: crate::HotSequence(sequence), + }; + + let mut marks = crate::RetiredHotOps::default(); + + // Sequence 1 is lost for good; everything this author writes afterwards still retires. + for sequence in 2..=20 { + marks.extend([id(sequence)]); + } + + let runs = marks.retired_beyond.get(&author).map(Vec::len).unwrap_or_default(); + assert_eq!(runs, 1, "the ops after the gap coalesce into a single run"); + for sequence in 2..=20 { + assert!(marks.covers(id(sequence)), "sequence {sequence} must stay covered"); + } + assert!(!marks.covers(id(1)), "the lost op is not claimed"); +} + +/// A retired delta keeps its op's authoring timestamp, so the snapshot resolves LWW on the same stamps +/// the live view did rather than on retirement order. +#[test] +fn retirement_preserves_the_live_lww_winner() { + let mut host = Session::with_peer(PeerId(1)); + + // Same key from two authors, the newer op arriving first, so LWW keeps it and discards the older. + let newer = HotOp { + op: set_document_attribute("k", 1), + timestamp: TimeStamp { counter: 10, peer: PeerId(2) }, + sequence: crate::HotSequence(1), + }; + let older = HotOp { + op: set_document_attribute("k", 2), + timestamp: TimeStamp { counter: 5, peer: PeerId(3) }, + sequence: crate::HotSequence(1), + }; + host.apply_hot_op(newer.clone()).expect("apply newer"); + host.apply_hot_op(older).expect("apply older"); + + let live = host.registry().attributes.get("k").cloned(); + host.retire(newer.timestamp).expect("retire"); + + assert_eq!(host.registry().attributes.get("k"), live.as_ref(), "retirement must not move the live value"); + assert_eq!( + host.retired_registry().attributes.get("k").map(|attribute| &attribute.value), + live.as_ref().map(|attribute| &attribute.value), + "the snapshot has to record the value the live view settled on" + ); +} + +/// A straggler LWW discarded still retires, and must not return by doing so: stamped at retirement it +/// would outrank the op that beat it, putting the discarded value into history itself. +#[test] +fn retiring_a_straggler_leaves_the_discarded_value_behind() { + let mut host = Session::with_peer(PeerId(1)); + + let winner = HotOp { + op: set_document_attribute("k", 1), + timestamp: TimeStamp { counter: 10, peer: PeerId(2) }, + sequence: crate::HotSequence(1), + }; + host.apply_hot_op(winner.clone()).expect("apply winner"); + host.retire(winner.timestamp).expect("retire winner"); + + // Arriving behind the winner, this op loses in the live view and its value is dropped. + let straggler = HotOp { + op: set_document_attribute("k", 2), + timestamp: TimeStamp { counter: 5, peer: PeerId(3) }, + sequence: crate::HotSequence(1), + }; + host.apply_hot_op(straggler.clone()).expect("apply straggler"); + host.retire(straggler.timestamp).expect("retire straggler"); + + let value = |registry: &crate::Registry| registry.attributes.get("k").map(|attribute| attribute.value.clone()); + assert_eq!(value(host.registry()), Some(serde_json::json!(1)), "the live view keeps the winner"); + assert_eq!(value(host.retired_registry()), Some(serde_json::json!(1)), "and retiring the straggler must not overwrite it"); + + let replayed = host.snapshot_from_history().expect("refold"); + assert_eq!(value(&replayed), Some(serde_json::json!(1)), "nor may history replay to the discarded value"); +}