From 6ae05765d2a4cef210378235a1f4c6a947c8329e Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Wed, 23 Sep 2026 18:21:44 +0200 Subject: [PATCH 01/11] Make retired state a function of history alone --- document/format/src/export.rs | 1 + document/format/src/lib.rs | 10 +- document/format/src/persist.rs | 1 + document/format/src/session_state.rs | 4 + document/format/tests/open_create.rs | 5 + document/graph-storage/src/document.rs | 129 ++++++++-- document/graph-storage/src/history.rs | 65 +++-- document/graph-storage/src/model.rs | 21 ++ document/graph-storage/src/session.rs | 299 ++++++++++++++++++++--- document/graph-storage/src/tests/crdt.rs | 216 +++++++++++++++- 10 files changed, 670 insertions(+), 81 deletions(-) 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..14d451bf1ee 100644 --- a/document/format/src/lib.rs +++ b/document/format/src/lib.rs @@ -144,7 +144,15 @@ impl Gdd { (true, true) => { let registry: Registry = io::read_single(&working, layout.registry_basename(), codecs.registry).await?; let history = load_history(&working, &layout, codecs.history).await?; - Session::load(peer, registry, history, session_state.head_rev, session_state.redo_stack, session_state.next_node_counter) + Session::load( + peer, + registry, + history, + session_state.head_rev, + session_state.redo_stack, + session_state.next_node_counter, + session_state.next_hot_sequence, + ) } (true, false) => { // Registry-only export: synthesize a history that reproduces this state. 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..f6375b5af9c 100644 --- a/document/format/src/session_state.rs +++ b/document/format/src/session_state.rs @@ -31,6 +31,10 @@ pub struct SessionState { /// collide on minted IDs. #[serde(default)] pub next_node_counter: u64, + /// How many hot ops this peer has authored, feeding `HotOp::sequence`. Persisted so a reopen + /// continues the run: a reused sequence would look like one already retired and be dropped. + #[serde(default)] + pub next_hot_sequence: u64, /// Per-peer view settings (PTZ, rulers, overlays, snapping, panel collapse). Local to the viewer, /// so kept out of the CRDT/history. Editor owns the keys/values (opaque `ui::doc::*` blobs). #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] diff --git a/document/format/tests/open_create.rs b/document/format/tests/open_create.rs index 4bdd43d7bd7..270db58b2e3 100644 --- a/document/format/tests/open_create.rs +++ b/document/format/tests/open_create.rs @@ -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: 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: 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: 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: 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: 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..7e10b845f9a 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, LamportClock, MAX_EXPORT_SLOTS, NetworkId, NodeId, NodeInput, PeerId, Registry, RegistryDelta, ResourceEntry, RetiredMarks, 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 that arrives after its own retirement is recognized + /// and dropped rather than re-entering the hot log for good. Retirement discards the link from a + /// delta back to the hot op it came from, so this is the only record of it. See [`RetiredMarks`]. + pub(crate) retired: RetiredMarks, /// 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,8 @@ 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: u64, } impl Document { @@ -66,24 +72,14 @@ 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 delta in canonical history order matching `predicate`, among those reachable from `head` or + /// any history tip (following all parents, including a merge's `extra_parents`). Tips cover deltas + /// absorbed mid-merge that `head` doesn't reach yet. Canonical order rather than a walk from `head`, + /// because resurrection reads the removed entity out of the match and peers sit on different merge revs. 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 @@ -100,7 +96,7 @@ impl Document { } /// 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. + /// Doesn't touch history or `head`, since hot ops are transient. pub 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); @@ -111,11 +107,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> { + // Its effect is already in retired history, and re-adding it would leave an entry no retirement + // list will ever name. + if self.is_retired(hot_op.id()) { + return Ok(()); + } + + // A hot op is identified by its timestamp, so a re-announcement of one already held is a no-op + // rather than a second copy that replays and re-broadcasts as though it were new work. + if self.hot_log.iter().any(|held| held.timestamp == hot_op.timestamp) { + return Ok(()); + } + + // Our own ops replayed after a reload are what carry the sequence counter back. + 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 another peer's retirement marks as well as this peer's. + pub(crate) fn absorb_retired(&mut self, remote: &RetiredMarks) { + self.retired.absorb(remote); + + self.drop_retired_hot_ops(); + } + + /// Drop hot ops history already covers, so a retired op never sits in the hot log whatever path + /// put it there. + fn drop_retired_hot_ops(&mut self) { + let retired = std::mem::take(&mut self.retired); + self.hot_log.retain(|hot_op| !retired.covers(hot_op.id())); + self.retired = retired; + } + /// 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 +355,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 +394,52 @@ 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(()); + } + 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), } - Ok(()) + } + + /// The most recent hot removal of a network, for reviving it in the working zone. Returns `None` + /// for the snapshot, which must stay 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 this peer's own, so reviving the snapshot zone from it would make the retired + // state depend on which hot ops happened to be here. Only history is shared. + 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..e7a6807a175 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}; @@ -64,22 +64,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, so two peers that absorb the same delta set end up with byte-identical history + /// rather than 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 +109,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..e31e39763d8 100644 --- a/document/graph-storage/src/model.rs +++ b/document/graph-storage/src/model.rs @@ -47,6 +47,27 @@ 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 { + let slot = InputSlot { + input: NodeInput::Value { + value: serde_json::Value::Null, + exposed: false, + }, + timestamp: TimeStamp::ORIGIN, + attributes: Attributes::new(), + }; + + Self { + implementation, + inputs: vec![slot; inputs], + attributes: Attributes::new(), + network, + } + } + #[cfg(test)] pub(crate) fn dummy() -> Self { Self { diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index cea711596ad..5d74467a99b 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,17 @@ impl Session { retired_snapshot: Registry::default(), history: History::new(), hot_log: Vec::new(), + retired: RetiredMarks::default(), head: None, redo_stack: Vec::new(), clock: LamportClock::new(peer), peer, last_broadcast_rev: None, next_node_counter: 0, + next_hot_sequence: 0, }, remote_tips: HashMap::new(), + runtime_base: None, } } @@ -77,11 +83,18 @@ 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); + self.runtime_base = Some(conversion.registry.clone()); let hot_ops = self.stage_ops(ops)?; 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 +153,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,9 +167,11 @@ impl Session { let mut staged = Vec::with_capacity(pending.len()); for op in pending { + self.document.next_hot_sequence += 1; let hot_op = HotOp { op, timestamp: self.document.clock.tick(), + sequence: self.document.next_hot_sequence, }; self.document.apply_hot_op(hot_op.clone())?; staged.push(hot_op); @@ -186,6 +201,8 @@ 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 parent = self.document.head; @@ -213,7 +230,7 @@ impl Session { /// Wrap an already-materialized snapshot. Trusts `registry` to match `history`; advances the /// clock past every observed timestamp but does not re-apply ops. `history` is taken in on-disk /// (topological) order. - pub fn load(peer: PeerId, registry: Registry, history: Vec, head: Option, redo_stack: Vec, next_node_counter: u64) -> Self { + pub fn load(peer: PeerId, registry: Registry, history: Vec, head: Option, redo_stack: Vec, next_node_counter: u64, next_hot_sequence: u64) -> Self { let mut clock = LamportClock::new(peer); for delta in &history { clock.observe(delta.timestamp); @@ -227,14 +244,17 @@ impl Session { working_registry: registry, history: History::from_ordered(history), hot_log: Vec::new(), + retired: RetiredMarks::default(), head, redo_stack, clock, peer, last_broadcast_rev: None, next_node_counter, + next_hot_sequence, }, remote_tips: HashMap::new(), + runtime_base: None, } } @@ -256,47 +276,133 @@ impl Session { Ok(session) } - /// Apply a hot op without going through the broadcast stream. + /// Apply a hot op received from another peer. pub fn apply_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> { self.document.apply_hot_op(hot_op) } + /// The hot ops a `retire(up_to)` call would drain. Broadcast alongside the retired deltas so guests + /// drop exactly these; a cutoff alone doesn't transfer, since a lagging peer's op can carry a + /// timestamp below the cutoff yet reach the host only after that retirement. + 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. + pub fn discard_hot_ops(&mut self, retired: &[HotOpId]) { + 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()); + } + + /// Which hot ops history already covers, for a peer catching up. See [`RetiredMarks`]. + pub fn retired_marks(&self) -> &RetiredMarks { + &self.document.retired + } + + /// Take on another peer's retirement marks as well as this peer's. + pub fn absorb_retired_marks(&mut self, remote: &RetiredMarks) { + self.document.absorb_retired(remote); + } + /// Replay a persisted hot op. Idempotent on structural ops, suitable for crash recovery /// where the registry may already reflect the op's effect from a prior retired snapshot. pub fn replay_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> { 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. Every peer sorts history the same + /// way, so this is independent of the order deltas arrived in and of which hot ops a peer holds. + /// Folded on a throwaway clone, so it observes rather than changes this session. + pub fn snapshot_from_history(&self) -> Result { + let mut folded = self.clone(); + folded.document.retired_snapshot = Registry::default(); + + let replay: Vec<(RegistryDelta, TimeStamp)> = folded.document.history.iter().map(|delta| (delta.kind.clone(), delta.timestamp)).collect(); + for (kind, timestamp) in replay { + folded.document.apply_op_with(RegistryTarget::Snapshot, kind, timestamp, ApplyMode::Idempotent)?; + } + + Ok(folded.document.retired_snapshot) + } + + /// Rebuild both registries from canonical history, then re-layer the hot tail. Costs a full replay, + /// so callers check that it is needed first. + fn refold_registries(&mut self) -> Result<(), CrdtError> { + self.document.retired_snapshot = self.snapshot_from_history()?; + + // A hot op the refold left unapplicable is kept rather than dropped, since a later delta or hot + // op can still supply the referent it names. + 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)); + } + } + + match failure { + Some(error) => Err(error), + None => 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); + // The working registry is the snapshot plus the hot tail, so a retired delta lands on both + // rather than cloning one over the other, which would promote unretired 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_ids.is_empty() { + return Ok(MergeOutcome::NoOp); } - if absorbed.is_empty() { - return Ok(None); + self.document.history.canonical_sort(); + + // The loop above 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. Refold unless the sort left the + // batch in the order it was applied. + let folded_in_canonical_order = self.document.history.iter().skip(length_before).map(|delta| delta.id).eq(arrival_order.iter().copied()); + if !folded_in_canonical_order { + self.refold_registries()?; } - 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); + 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(); - // 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 @@ -304,6 +410,8 @@ impl Session { /// /// Today: one retired delta per hot op. Coarsening is a future step. pub fn retire(&mut self, up_to: TimeStamp) -> Result, CrdtError> { + // Drained in hot-log order, which 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()); for hot_op in self.document.hot_log.drain(..) { @@ -314,6 +422,7 @@ impl Session { } } self.document.hot_log = remaining; + self.document.mark_retired(drained.iter().map(HotOp::id)); self.commit_ops(drained.into_iter().map(|hot_op| hot_op.op), true) } @@ -369,16 +478,18 @@ 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 + /// 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 + /// rather than a single delta, matching the legacy per-interaction granularity. The undone interaction's /// `head` rev is pushed onto the redo stack. Reflog semantics: the DAG is never 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 +498,11 @@ 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())?; + // The working registry is the snapshot plus the hot tail, so rewind both rather than copying + // one over the other, which would promote unretired hot ops into retired state. + self.document.revert_delta(RegistryTarget::Snapshot, delta)?; + self.document.head = parent; match parent { @@ -397,9 +512,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 +521,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 +536,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 +560,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 +677,12 @@ impl Session { pub fn next_node_counter(&self) -> u64 { self.document.next_node_counter } + + /// How many hot ops this peer has authored. Persisted and carried across a reload so a fresh op + /// never reuses a sequence an earlier one already spent. See [`HotOp::sequence`]. + pub fn next_hot_sequence(&self) -> u64 { + self.document.next_hot_sequence + } } /// Errors from `Session::commit_from_runtime`. @@ -572,12 +702,109 @@ 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 own run of hot ops, counting from 1 with no gaps, so a watermark over it + /// stands for a contiguous prefix. The Lamport counter cannot: it skips on observing a higher remote + /// timestamp, making a missing op indistinguishable from a skipped tick. + pub sequence: u64, +} + +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, + } + } +} + +/// One hot op's author and position in that author's run. Retirement names the ops it promoted with +/// these rather than with timestamps, so a receiver can tell which prefix history now covers. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +pub struct HotOpId { + pub peer: PeerId, + pub sequence: u64, +} + +/// Which hot ops history already covers. `through` is how far each author's run has retired without a +/// gap, and `above` names the retired ops past that point, which is where an op lands when it retires +/// before an earlier one from the same author has arrived. Replicated so a peer catching up can tell +/// an op already in the history it was handed from one still owed to it. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RetiredMarks { + pub through: HashMap, + pub above: HashSet, +} + +impl RetiredMarks { + /// Whether history already holds this hot op. + pub fn covers(&self, id: HotOpId) -> bool { + self.through.get(&id.peer).is_some_and(|&through| id.sequence <= through) || self.above.contains(&id) + } + + /// Take on `remote`'s coverage as well as this one's. + pub fn absorb(&mut self, remote: &Self) { + for (&peer, &remote_through) in &remote.through { + let through = self.through.entry(peer).or_default(); + *through = (*through).max(remote_through); + } + self.above.extend(remote.above.iter().copied()); + + self.compact(); + } + + /// Record newly retired ops. + pub fn extend(&mut self, retired: impl IntoIterator) { + self.above.extend(retired); + + self.compact(); + } + + /// Fold every exception that continues its author's prefix into `through`, so the set only ever + /// holds ops still separated from the prefix by a gap. + fn compact(&mut self) { + let mut by_author: HashMap> = HashMap::new(); + for id in &self.above { + by_author.entry(id.peer).or_default().push(id.sequence); + } + + for (peer, mut sequences) in by_author { + sequences.sort_unstable(); + let through = self.through.entry(peer).or_default(); + for sequence in sequences { + if sequence == *through + 1 { + *through = sequence; + } + } + } + + let through = std::mem::take(&mut self.through); + self.above.retain(|id| through.get(&id.peer).is_none_or(|&covered| id.sequence > covered)); + self.through = through; + } } #[derive(Debug, thiserror::Error)] diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index 78fd8f082ae..fae8baa2d05 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: 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,190 @@ 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, which sits at +/// `retired_snapshot + hot tail`. `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 and report whether node 7 survived, +/// with the stamp on its `tint` attribute. +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)) +} + +/// Retired deltas reach the registries in arrival order while only `history` is canonically sorted, +/// so concurrent deltas must commute. A remove concurrent with an attribute change does not: taken +/// change-first the node is gone, taken remove-first it is resurrected from the removal's snapshot. +#[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"); +} + +/// A hot op can retire before an earlier one from the same author has arrived, so the marks cover it +/// out of order. The prefix absorbs it once the gap fills, which is what bounds the exception set. +#[test] +fn retired_marks_cover_a_gap_and_compact_once_it_fills() { + let author = PeerId(1); + let id = |sequence| crate::HotOpId { peer: author, sequence }; + + let mut marks = crate::RetiredMarks::default(); + marks.extend([id(1), id(2)]); + assert_eq!(marks.through.get(&author), Some(&2), "a contiguous run folds straight into the prefix"); + assert!(marks.above.is_empty()); + + // Sequence 3 never arrived, so 4 retires above the prefix rather than extending it. + marks.extend([id(4)]); + assert_eq!(marks.through.get(&author), Some(&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.through.get(&author), Some(&4)); + assert!(marks.above.is_empty(), "the exception set empties once the prefix reaches it"); +} + +/// Marks merge by union, and a peer adopting another's wholesale must not lose its own coverage. +#[test] +fn absorbing_marks_keeps_both_sides_coverage() { + let author = PeerId(1); + let id = |sequence| crate::HotOpId { peer: author, sequence }; + + let mut local = crate::RetiredMarks::default(); + local.extend([id(1), id(4)]); + + let mut remote = crate::RetiredMarks::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.through.get(&author), Some(&5)); + assert!(local.above.is_empty()); + for sequence in 1..=5 { + assert!(local.covers(id(sequence)), "sequence {sequence} must stay covered"); + } +} + +/// Undo rewinds retired history, so it must leave the hot tail alone: the retired snapshot is the +/// state history alone produces, and copying the working registry onto it promotes live work nobody +/// has retired. Hot ops exist between staging and retirement even with no peers. +#[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 stays `snapshot + hot tail` rather than +/// drifting apart in the other direction. +#[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 rewrites the local registries without emitting anything, so it is only legal while a +/// commit is still 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"); +} From 1bbce952e27f0cee036b211799c544888a3bb4b0 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Wed, 23 Sep 2026 18:45:44 +0200 Subject: [PATCH 02/11] Shorten the comments on the retired-state changes --- document/graph-storage/src/document.rs | 32 +++++------- document/graph-storage/src/session.rs | 64 ++++++++++-------------- document/graph-storage/src/tests/crdt.rs | 30 +++++------ 3 files changed, 52 insertions(+), 74 deletions(-) diff --git a/document/graph-storage/src/document.rs b/document/graph-storage/src/document.rs index 7e10b845f9a..e20024b0c27 100644 --- a/document/graph-storage/src/document.rs +++ b/document/graph-storage/src/document.rs @@ -11,9 +11,9 @@ 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 that arrives after its own retirement is recognized - /// and dropped rather than re-entering the hot log for good. Retirement discards the link from a - /// delta back to the hot op it came from, so this is the only record of it. See [`RetiredMarks`]. + /// Which hot ops history already covers, so one arriving after its own retirement is dropped rather + /// than re-entering the hot log. Retirement drops the delta's link back to its hot op, so this is the + /// only record. See [`RetiredMarks`]. pub(crate) retired: RetiredMarks, /// 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 @@ -72,10 +72,8 @@ impl Document { self.revert_delta(target, delta) } - /// The last delta in canonical history order matching `predicate`, among those reachable from `head` or - /// any history tip (following all parents, including a merge's `extra_parents`). Tips cover deltas - /// absorbed mid-merge that `head` doesn't reach yet. Canonical order rather than a walk from `head`, - /// because resurrection reads the removed entity out of the match and peers sit on different merge revs. + /// 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 reachable = self.history.ancestors(self.head.into_iter().chain(self.history.tips())); @@ -107,19 +105,17 @@ 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> { - // Its effect is already in retired history, and re-adding it would leave an entry no retirement - // list will ever name. + // 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(()); } - // A hot op is identified by its timestamp, so a re-announcement of one already held is a no-op - // rather than a second copy that replays and re-broadcasts as though it were new work. + // 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(()); } - // Our own ops replayed after a reload are what carry the sequence counter back. + // 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); } @@ -141,15 +137,14 @@ impl Document { self.drop_retired_hot_ops(); } - /// Take on another peer's retirement marks as well as this peer's. + /// Take on a peer's retirement marks as well as this peer's. pub(crate) fn absorb_retired(&mut self, remote: &RetiredMarks) { self.retired.absorb(remote); self.drop_retired_hot_ops(); } - /// Drop hot ops history already covers, so a retired op never sits in the hot log whatever path - /// put it there. + /// Drop hot ops history already covers, whatever path put them in the log. fn drop_retired_hot_ops(&mut self) { let retired = std::mem::take(&mut self.retired); self.hot_log.retain(|hot_op| !retired.covers(hot_op.id())); @@ -404,8 +399,8 @@ impl Document { } } - /// The most recent hot removal of a network, for reviving it in the working zone. Returns `None` - /// for the snapshot, which must stay a function of history alone. + /// 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; @@ -433,8 +428,7 @@ impl Document { if self.registry_ref(target).networks.contains_key(&network_id) { return Ok(()); } - // The hot log is this peer's own, so reviving the snapshot zone from it would make the retired - // state depend on which hot ops happened to be here. Only history is shared. + // 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), diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index 5d74467a99b..9c0a612d33d 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -281,9 +281,8 @@ impl Session { self.document.apply_hot_op(hot_op) } - /// The hot ops a `retire(up_to)` call would drain. Broadcast alongside the retired deltas so guests - /// drop exactly these; a cutoff alone doesn't transfer, since a lagging peer's op can carry a - /// timestamp below the cutoff yet reach the host only after that retirement. + /// The hot ops a `retire(up_to)` call would drain. Sent with the deltas so peers drop exactly these; + /// the cutoff alone doesn't transfer, since 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() } @@ -300,7 +299,7 @@ impl Session { &self.document.retired } - /// Take on another peer's retirement marks as well as this peer's. + /// See [`RetiredMarks::absorb`]. pub fn absorb_retired_marks(&mut self, remote: &RetiredMarks) { self.document.absorb_retired(remote); } @@ -311,9 +310,8 @@ impl Session { self.document.replay_hot_op(hot_op) } - /// The retired snapshot as canonical history alone produces it. Every peer sorts history the same - /// way, so this is independent of the order deltas arrived in and of which hot ops a peer holds. - /// Folded on a throwaway clone, so it observes rather than changes this session. + /// The retired snapshot as canonical history alone produces it, independent of arrival order and the + /// hot log. Folded on a clone, so it observes rather than changes this session. pub fn snapshot_from_history(&self) -> Result { let mut folded = self.clone(); folded.document.retired_snapshot = Registry::default(); @@ -326,13 +324,12 @@ impl Session { Ok(folded.document.retired_snapshot) } - /// Rebuild both registries from canonical history, then re-layer the hot tail. Costs a full replay, - /// so callers check that it is needed first. + /// Rebuild both registries from canonical history, then re-layer the hot tail. A full replay, so + /// callers check it is needed first. fn refold_registries(&mut self) -> Result<(), CrdtError> { self.document.retired_snapshot = self.snapshot_from_history()?; - // A hot op the refold left unapplicable is kept rather than dropped, since a later delta or hot - // op can still supply the referent it names. + // 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) { @@ -364,8 +361,7 @@ impl Session { } arrival_order.push(delta.id); self.document.apply_op_idempotent(delta.kind.clone(), delta.timestamp)?; - // The working registry is the snapshot plus the hot tail, so a retired delta lands on both - // rather than cloning one over the other, which would promote unretired hot ops. + // 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); @@ -375,9 +371,8 @@ impl Session { } self.document.history.canonical_sort(); - // The loop above 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. Refold unless the sort left the - // batch in the order it was applied. + // 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()); if !folded_in_canonical_order { self.refold_registries()?; @@ -410,8 +405,7 @@ impl Session { /// /// Today: one retired delta per hot op. Coarsening is a future step. pub fn retire(&mut self, up_to: TimeStamp) -> Result, CrdtError> { - // Drained in hot-log order, which is causal, so the deltas commit in an order their references - // survive. + // 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()); for hot_op in self.document.hot_log.drain(..) { @@ -478,11 +472,10 @@ 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 - /// rather than a single delta, matching the legacy per-interaction 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 spans a whole + /// `commit_from_runtime` batch, so the run is reverted rather than one delta. Its `head` rev goes on + /// the redo stack; the DAG itself is never rewritten. pub fn undo(&mut self) -> Result { if !self.can_undo() { return Err(CrdtError::NothingToUndo); @@ -499,8 +492,7 @@ impl Session { let parent = delta.parent; self.document.revert_delta(RegistryTarget::Working, delta.clone())?; - // The working registry is the snapshot plus the hot tail, so rewind both rather than copying - // one over the other, which would promote unretired hot ops into retired state. + // 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; @@ -678,8 +670,7 @@ impl Session { self.document.next_node_counter } - /// How many hot ops this peer has authored. Persisted and carried across a reload so a fresh op - /// never reuses a sequence an earlier one already spent. See [`HotOp::sequence`]. + /// How many hot ops this peer has authored, carried across a reload so no sequence is spent twice. pub fn next_hot_sequence(&self) -> u64 { self.document.next_hot_sequence } @@ -725,9 +716,8 @@ impl MergeOutcome { pub struct HotOp { pub op: RegistryDelta, pub timestamp: TimeStamp, - /// Position in its author's own run of hot ops, counting from 1 with no gaps, so a watermark over it - /// stands for a contiguous prefix. The Lamport counter cannot: it skips on observing a higher remote - /// timestamp, making a missing op indistinguishable from a skipped tick. + /// Position in its author's run, from 1 with no gaps, so a watermark over it means a contiguous prefix. + /// The Lamport counter skips on observing a higher remote timestamp, so it cannot. pub sequence: u64, } @@ -741,18 +731,17 @@ impl HotOp { } } -/// One hot op's author and position in that author's run. Retirement names the ops it promoted with -/// these rather than with timestamps, so a receiver can tell which prefix history now covers. +/// One hot op's author and position in its run. Retirement names promoted ops by these, so a receiver +/// can tell which prefix history covers. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] pub struct HotOpId { pub peer: PeerId, pub sequence: u64, } -/// Which hot ops history already covers. `through` is how far each author's run has retired without a -/// gap, and `above` names the retired ops past that point, which is where an op lands when it retires -/// before an earlier one from the same author has arrived. Replicated so a peer catching up can tell -/// an op already in the history it was handed from one still owed to it. +/// Which hot ops history already covers: `through` is each author's gap-free retired prefix, `above` the +/// retired ops past it. Replicated, so a peer catching up can tell an op already in history from one +/// still owed to it. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RetiredMarks { pub through: HashMap, @@ -783,8 +772,7 @@ impl RetiredMarks { self.compact(); } - /// Fold every exception that continues its author's prefix into `through`, so the set only ever - /// holds ops still separated from the prefix by a gap. + /// Fold exceptions that continue their author's prefix into `through`, leaving only those past a gap. fn compact(&mut self) { let mut by_author: HashMap> = HashMap::new(); for id in &self.above { diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index fae8baa2d05..90f0bf3fcda 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -906,8 +906,8 @@ 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, which sits at -/// `retired_snapshot + hot tail`. `commit_op_for_test` alone only advances the snapshot zone. +/// 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"); @@ -917,8 +917,7 @@ fn commit_retired(session: &mut Session, op: RegistryDelta) { } } -/// Merge `deltas` into a copy of `base` in the given order and report whether node 7 survived, -/// with the stamp on its `tint` attribute. +/// 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(); @@ -932,9 +931,8 @@ fn merge_order_outcome(base: &Session, deltas: &[Delta], order: [usize; 2]) -> O .map(|value| (value.value.clone(), value.timestamp)) } -/// Retired deltas reach the registries in arrival order while only `history` is canonically sorted, -/// so concurrent deltas must commute. A remove concurrent with an attribute change does not: taken -/// change-first the node is gone, taken remove-first it is resurrected from the removal's snapshot. +/// The registries are folded in arrival order while only `history` is sorted, so concurrent deltas must +/// commute. Change-first leaves the node gone; remove-first resurrects it from the removal's snapshot. #[test] fn a_concurrent_remove_and_attribute_change_commute() { let network_id = NetworkId(5); @@ -978,8 +976,8 @@ fn a_concurrent_remove_and_attribute_change_commute() { assert_eq!(removal_first, change_first, "the retired registry must not depend on delta arrival order"); } -/// A hot op can retire before an earlier one from the same author has arrived, so the marks cover it -/// out of order. The prefix absorbs it once the gap fills, which is what bounds the exception set. +/// 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); @@ -1002,7 +1000,7 @@ fn retired_marks_cover_a_gap_and_compact_once_it_fills() { assert!(marks.above.is_empty(), "the exception set empties once the prefix reaches it"); } -/// Marks merge by union, and a peer adopting another's wholesale must not lose its own coverage. +/// 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); @@ -1024,9 +1022,8 @@ fn absorbing_marks_keeps_both_sides_coverage() { } } -/// Undo rewinds retired history, so it must leave the hot tail alone: the retired snapshot is the -/// state history alone produces, and copying the working registry onto it promotes live work nobody -/// has retired. Hot ops exist between staging and retirement even with no peers. +/// 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)); @@ -1051,8 +1048,7 @@ fn undo_does_not_promote_hot_ops_into_the_retired_snapshot() { ); } -/// Redo puts the interaction back on both zones, so the pair stays `snapshot + hot tail` rather than -/// drifting apart in the other direction. +/// 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)); @@ -1074,8 +1070,8 @@ fn redo_restores_the_retired_snapshot_without_the_hot_tail() { assert!(session.registry().attributes.contains_key("hot"), "the hot op must survive an undo/redo round trip"); } -/// Silent undo rewrites the local registries without emitting anything, so it is only legal while a -/// commit is still unpublished. Once peers hold it, a rewind would diverge from them for good. +/// 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)); From 1d9653d40fa86fe5852479f39db807114227d4a0 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Wed, 23 Sep 2026 18:57:04 +0200 Subject: [PATCH 03/11] Address review findings --- document/format/src/lib.rs | 12 ++---- document/format/src/session_state.rs | 9 +++-- document/graph-storage/src/model.rs | 17 ++++++++- document/graph-storage/src/session.rs | 48 ++++++++++++++++++------ document/graph-storage/src/tests/crdt.rs | 24 +++++++++++- 5 files changed, 83 insertions(+), 27 deletions(-) diff --git a/document/format/src/lib.rs b/document/format/src/lib.rs index 14d451bf1ee..295354bf772 100644 --- a/document/format/src/lib.rs +++ b/document/format/src/lib.rs @@ -144,15 +144,7 @@ impl Gdd { (true, true) => { let registry: Registry = io::read_single(&working, layout.registry_basename(), codecs.registry).await?; let history = load_history(&working, &layout, codecs.history).await?; - Session::load( - peer, - registry, - history, - session_state.head_rev, - session_state.redo_stack, - session_state.next_node_counter, - session_state.next_hot_sequence, - ) + Session::load(peer, registry, history, session_state.head_rev, session_state.redo_stack, session_state.next_node_counter) } (true, false) => { // Registry-only export: synthesize a history that reproduces this state. @@ -166,6 +158,8 @@ impl Gdd { if let Some(rev) = session_state.last_broadcast_rev { session.publish_up_to(rev); } + // Likewise the authored-op count, so a reopen through any arm continues this peer's run. + 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/session_state.rs b/document/format/src/session_state.rs index f6375b5af9c..79db156e4c1 100644 --- a/document/format/src/session_state.rs +++ b/document/format/src/session_state.rs @@ -31,10 +31,6 @@ pub struct SessionState { /// collide on minted IDs. #[serde(default)] pub next_node_counter: u64, - /// How many hot ops this peer has authored, feeding `HotOp::sequence`. Persisted so a reopen - /// continues the run: a reused sequence would look like one already retired and be dropped. - #[serde(default)] - pub next_hot_sequence: u64, /// Per-peer view settings (PTZ, rulers, overlays, snapping, panel collapse). Local to the viewer, /// so kept out of the CRDT/history. Editor owns the keys/values (opaque `ui::doc::*` blobs). #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] @@ -43,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`]. + /// Persisted so a reopen continues the run rather than reusing a spent sequence. Appended last: + /// a positional codec decodes these fields in declaration order. + #[serde(default)] + pub next_hot_sequence: u64, } diff --git a/document/graph-storage/src/model.rs b/document/graph-storage/src/model.rs index e31e39763d8..b696dff3b38 100644 --- a/document/graph-storage/src/model.rs +++ b/document/graph-storage/src/model.rs @@ -51,9 +51,11 @@ impl Node { /// `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` deserializes this as a `TaggedValue`, whose unit `None` variant encodes as this + // string. `unset_input_slot_deserializes_as_tagged_value_none` pins the pairing. let slot = InputSlot { input: NodeInput::Value { - value: serde_json::Value::Null, + value: serde_json::Value::String("None".to_string()), exposed: false, }, timestamp: TimeStamp::ORIGIN, @@ -180,6 +182,19 @@ mod tests { } } + /// `Node::new` writes an unset slot value that `to_runtime` has to be able to read back, so the + /// literal it stores 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 9c0a612d33d..dd275bff957 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -85,8 +85,11 @@ impl Session { let conversion = Registry::convert_from_runtime(network, metadata, resources, self.document.peer)?; let base = self.runtime_base.as_ref().unwrap_or(&self.document.working_registry); let ops = crate::delta::compute_deltas(base, &conversion.registry); - self.runtime_base = Some(conversion.registry.clone()); + + // 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)) } @@ -167,13 +170,16 @@ impl Session { let mut staged = Vec::with_capacity(pending.len()); for op in pending { - self.document.next_hot_sequence += 1; + // 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 + 1; let hot_op = HotOp { op, timestamp: self.document.clock.tick(), - sequence: self.document.next_hot_sequence, + sequence, }; self.document.apply_hot_op(hot_op.clone())?; + + self.document.next_hot_sequence = sequence; staged.push(hot_op); } Ok(staged) @@ -230,7 +236,7 @@ impl Session { /// Wrap an already-materialized snapshot. Trusts `registry` to match `history`; advances the /// clock past every observed timestamp but does not re-apply ops. `history` is taken in on-disk /// (topological) order. - pub fn load(peer: PeerId, registry: Registry, history: Vec, head: Option, redo_stack: Vec, next_node_counter: u64, next_hot_sequence: u64) -> Self { + pub fn load(peer: PeerId, registry: Registry, history: Vec, head: Option, redo_stack: Vec, next_node_counter: u64) -> Self { let mut clock = LamportClock::new(peer); for delta in &history { clock.observe(delta.timestamp); @@ -251,7 +257,7 @@ impl Session { peer, last_broadcast_rev: None, next_node_counter, - next_hot_sequence, + next_hot_sequence: 0, }, remote_tips: HashMap::new(), runtime_base: None, @@ -276,9 +282,10 @@ impl Session { Ok(session) } - /// Apply a hot op received from another peer. + /// Apply a hot op received from another peer. Idempotent, and a no-op for one history already + /// covers, since 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 so peers drop exactly these; @@ -312,11 +319,22 @@ impl Session { /// The retired snapshot as canonical history alone produces it, independent of arrival order and the /// hot log. Folded on a clone, so it observes rather than changes this session. + /// + /// Only `head`'s ancestry: an undone delta stays in the DAG for redo to find, so folding all of + /// history would restore work the user undid. 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(); - let replay: Vec<(RegistryDelta, TimeStamp)> = folded.document.history.iter().map(|delta| (delta.kind.clone(), delta.timestamp)).collect(); + let replay: Vec<(RegistryDelta, TimeStamp)> = folded + .document + .history + .iter() + .filter(|delta| reachable.contains(&delta.id)) + .map(|delta| (delta.kind.clone(), delta.timestamp)) + .collect(); for (kind, timestamp) in replay { folded.document.apply_op_with(RegistryTarget::Snapshot, kind, timestamp, ApplyMode::Idempotent)?; } @@ -374,9 +392,6 @@ impl Session { // 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()); - if !folded_in_canonical_order { - self.refold_registries()?; - } let history = &self.document.history; let mut candidates: Vec = history.tips().into_iter().filter(|tip| absorbed_ids.contains(tip)).collect(); @@ -397,6 +412,11 @@ impl Session { }; self.document.head = outcome.head(); + // After `head`, since the fold is over its ancestry. + if !folded_in_canonical_order { + self.refold_registries()?; + } + Ok(outcome) } @@ -674,6 +694,12 @@ impl Session { pub fn next_hot_sequence(&self) -> u64 { self.document.next_hot_sequence } + + /// Restore the authored-op count after a load, so a fresh op cannot reuse a spent sequence. Raises + /// only, so replaying a persisted hot log afterwards cannot lower it. + pub fn restore_hot_sequence(&mut self, sequence: u64) { + self.document.next_hot_sequence = self.document.next_hot_sequence.max(sequence); + } } /// Errors from `Session::commit_from_runtime`. diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index 90f0bf3fcda..bf31dfcb0f4 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -931,8 +931,8 @@ fn merge_order_outcome(base: &Session, deltas: &[Delta], order: [usize; 2]) -> O .map(|value| (value.value.clone(), value.timestamp)) } -/// The registries are folded in arrival order while only `history` is sorted, so concurrent deltas must -/// commute. Change-first leaves the node gone; remove-first resurrects it from the removal's snapshot. +/// 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); @@ -1088,3 +1088,23 @@ fn publishing_a_commit_disables_silent_undo() { assert!(!session.can_undo(), "a published interaction must not be silently rewound"); } + +/// An undone delta stays in the DAG so redo can find it, but it is no longer in `head`'s ancestry. +/// Folding history must respect that or a refold 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"); +} From ec057cea3b614d6f557d77590efdc39bb339ee2e Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Wed, 23 Sep 2026 19:17:16 +0200 Subject: [PATCH 04/11] Retry an owed refold and keep the unguarded hot-op path crate-private --- document/graph-storage/src/document.rs | 11 ++++++++--- document/graph-storage/src/session.rs | 21 +++++++++++++++++---- document/graph-storage/src/tests/crdt.rs | 22 ++++++++++++++++++++++ 3 files changed, 47 insertions(+), 7 deletions(-) diff --git a/document/graph-storage/src/document.rs b/document/graph-storage/src/document.rs index e20024b0c27..91cf04e06f9 100644 --- a/document/graph-storage/src/document.rs +++ b/document/graph-storage/src/document.rs @@ -42,6 +42,9 @@ pub struct Document { 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: u64, + /// Set while the registries are still folded from an older history, so a failed refold is retried + /// rather than leaving derived state behind the history it comes from. + pub(crate) refold_owed: bool, } impl Document { @@ -93,9 +96,11 @@ 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`, since hot ops are transient. - pub fn apply_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> { + /// Apply a locally staged op. Updates the registry via LWW and appends to the hot log; doesn't touch + /// history or `head`, since hot ops are transient. Crate-private: it skips the retirement check that + /// an op off the wire needs, so outside callers go through [`Session::apply_hot_op`], while local + /// staging must not have a covered sequence silently drop its own work. + 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(()) diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index dd275bff957..11ab40a9f4c 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -47,6 +47,7 @@ impl Session { last_broadcast_rev: None, next_node_counter: 0, next_hot_sequence: 0, + refold_owed: false, }, remote_tips: HashMap::new(), runtime_base: None, @@ -258,6 +259,7 @@ impl Session { last_broadcast_rev: None, next_node_counter, next_hot_sequence: 0, + refold_owed: false, }, remote_tips: HashMap::new(), runtime_base: None, @@ -345,6 +347,9 @@ impl Session { /// Rebuild both registries from canonical history, then re-layer the hot tail. A full replay, so /// 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, so a failure here has to be + // retried rather than leaving the registries derived from an older history for good. + 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. @@ -357,10 +362,12 @@ impl Session { } } - match failure { - Some(error) => Err(error), - None => Ok(()), + 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 @@ -385,6 +392,12 @@ impl Session { self.document.history.push(delta); } 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(); @@ -413,7 +426,7 @@ impl Session { self.document.head = outcome.head(); // After `head`, since the fold is over its ancestry. - if !folded_in_canonical_order { + if !folded_in_canonical_order || self.document.refold_owed { self.refold_registries()?; } diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index bf31dfcb0f4..e82686609fd 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -1108,3 +1108,25 @@ fn snapshot_from_history_ignores_undone_deltas() { assert!(folded.attributes.contains_key("first"), "history still holds the kept interaction"); assert!(!folded.attributes.contains_key("second"), "a fold restored an undone delta"); } + +/// A refold that fails leaves the registries derived from an older history, and the deltas are already +/// in history so re-merging them absorbs nothing. The owed refold has to be retried on the next merge +/// or 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" + ); +} From a8d6bb73e797adb2423d45b088612535fa4cbd35 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Wed, 23 Sep 2026 19:44:21 +0200 Subject: [PATCH 05/11] Vary comment phrasing and cut the repeated constructions --- document/format/src/session_state.rs | 6 ++-- document/graph-storage/src/document.rs | 15 ++++----- document/graph-storage/src/history.rs | 4 +-- document/graph-storage/src/model.rs | 8 ++--- document/graph-storage/src/session.rs | 43 +++++++++++------------- document/graph-storage/src/tests/crdt.rs | 9 +++-- 6 files changed, 40 insertions(+), 45 deletions(-) diff --git a/document/format/src/session_state.rs b/document/format/src/session_state.rs index 79db156e4c1..ee271e0bd30 100644 --- a/document/format/src/session_state.rs +++ b/document/format/src/session_state.rs @@ -39,9 +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`]. - /// Persisted so a reopen continues the run rather than reusing a spent sequence. Appended last: - /// a positional codec decodes these fields in declaration order. + /// 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: u64, } diff --git a/document/graph-storage/src/document.rs b/document/graph-storage/src/document.rs index 91cf04e06f9..3f3de709e48 100644 --- a/document/graph-storage/src/document.rs +++ b/document/graph-storage/src/document.rs @@ -11,8 +11,8 @@ 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 rather - /// than re-entering the hot log. Retirement drops the delta's link back to its hot op, so this is the + /// 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 [`RetiredMarks`]. pub(crate) retired: RetiredMarks, /// The registry as of the last retirement, with no un-retired hot ops applied. Retirement computes @@ -42,8 +42,8 @@ pub struct Document { 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: u64, - /// Set while the registries are still folded from an older history, so a failed refold is retried - /// rather than leaving derived state behind the history it comes from. + /// 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, } @@ -96,10 +96,9 @@ impl Document { self.apply_op_with(target, delta.kind, delta.timestamp, ApplyMode::Force) } - /// Apply a locally staged op. Updates the registry via LWW and appends to the hot log; doesn't touch - /// history or `head`, since hot ops are transient. Crate-private: it skips the retirement check that - /// an op off the wire needs, so outside callers go through [`Session::apply_hot_op`], while local - /// staging must not have a covered sequence silently drop its own work. + /// 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); diff --git a/document/graph-storage/src/history.rs b/document/graph-storage/src/history.rs index e7a6807a175..fafa4a9e9c6 100644 --- a/document/graph-storage/src/history.rs +++ b/document/graph-storage/src/history.rs @@ -67,8 +67,8 @@ impl History { /// 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). /// - /// Deterministic, so two peers that absorb the same delta set end up with byte-identical history - /// rather than two different valid orderings. Arrival order is erased. + /// 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()); diff --git a/document/graph-storage/src/model.rs b/document/graph-storage/src/model.rs index b696dff3b38..94762ca6cff 100644 --- a/document/graph-storage/src/model.rs +++ b/document/graph-storage/src/model.rs @@ -51,8 +51,8 @@ impl Node { /// `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` deserializes this as a `TaggedValue`, whose unit `None` variant encodes as this - // string. `unset_input_slot_deserializes_as_tagged_value_none` pins the pairing. + // `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()), @@ -182,8 +182,8 @@ mod tests { } } - /// `Node::new` writes an unset slot value that `to_runtime` has to be able to read back, so the - /// literal it stores must stay in step with how `TaggedValue::None` serializes. + /// `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); diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index 11ab40a9f4c..8b49d34930f 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -284,14 +284,14 @@ impl Session { Ok(session) } - /// Apply a hot op received from another peer. Idempotent, and a no-op for one history already - /// covers, since a late delivery must not put a retired op back in the hot log. + /// 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.replay_hot_op(hot_op) } - /// The hot ops a `retire(up_to)` call would drain. Sent with the deltas so peers drop exactly these; - /// the cutoff alone doesn't transfer, since a lagging op can arrive below it afterwards. + /// 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() } @@ -320,10 +320,8 @@ impl Session { } /// The retired snapshot as canonical history alone produces it, independent of arrival order and the - /// hot log. Folded on a clone, so it observes rather than changes this session. - /// - /// Only `head`'s ancestry: an undone delta stays in the DAG for redo to find, so folding all of - /// history would restore work the user undid. + /// 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. pub fn snapshot_from_history(&self) -> Result { let reachable = self.document.history.ancestors(self.document.head); @@ -344,11 +342,11 @@ impl Session { Ok(folded.document.retired_snapshot) } - /// Rebuild both registries from canonical history, then re-layer the hot tail. A full replay, so - /// callers check it is needed first. + /// 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, so a failure here has to be - // retried rather than leaving the registries derived from an older history for good. + // 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()?; @@ -506,9 +504,8 @@ impl Session { } /// 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 spans a whole - /// `commit_from_runtime` batch, so the run is reverted rather than one delta. Its `head` rev goes on - /// the redo stack; the DAG itself is never rewritten. + /// 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); @@ -703,13 +700,13 @@ impl Session { self.document.next_node_counter } - /// How many hot ops this peer has authored, carried across a reload so no sequence is spent twice. + /// How many hot ops this peer has authored. Carried across a reload; no sequence is spent twice. pub fn next_hot_sequence(&self) -> u64 { self.document.next_hot_sequence } - /// Restore the authored-op count after a load, so a fresh op cannot reuse a spent sequence. Raises - /// only, so replaying a persisted hot log afterwards cannot lower it. + /// 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: u64) { self.document.next_hot_sequence = self.document.next_hot_sequence.max(sequence); } @@ -755,8 +752,8 @@ impl MergeOutcome { pub struct HotOp { pub op: RegistryDelta, pub timestamp: TimeStamp, - /// Position in its author's run, from 1 with no gaps, so a watermark over it means a contiguous prefix. - /// The Lamport counter skips on observing a higher remote timestamp, so it cannot. + /// 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: u64, } @@ -770,8 +767,8 @@ impl HotOp { } } -/// One hot op's author and position in its run. Retirement names promoted ops by these, so a receiver -/// can tell which prefix history covers. +/// 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, @@ -779,7 +776,7 @@ pub struct HotOpId { } /// Which hot ops history already covers: `through` is each author's gap-free retired prefix, `above` the -/// retired ops past it. Replicated, so a peer catching up can tell an op already in history from one +/// 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 RetiredMarks { diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index e82686609fd..977351b3a4f 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -1089,8 +1089,8 @@ fn publishing_a_commit_disables_silent_undo() { assert!(!session.can_undo(), "a published interaction must not be silently rewound"); } -/// An undone delta stays in the DAG so redo can find it, but it is no longer in `head`'s ancestry. -/// Folding history must respect that or a refold restores work the user undid. +/// 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)); @@ -1109,9 +1109,8 @@ fn snapshot_from_history_ignores_undone_deltas() { assert!(!folded.attributes.contains_key("second"), "a fold restored an undone delta"); } -/// A refold that fails leaves the registries derived from an older history, and the deltas are already -/// in history so re-merging them absorbs nothing. The owed refold has to be retried on the next merge -/// or the registries never catch up. +/// 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)); From e1508acac55c25d7f8f84e22c13a5e08240da32d Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Wed, 23 Sep 2026 19:58:36 +0200 Subject: [PATCH 06/11] Give hot-op sequence numbers their own type --- document/format/src/lib.rs | 2 +- document/format/src/session_state.rs | 4 +-- document/format/tests/open_create.rs | 12 ++++---- document/graph-storage/src/document.rs | 6 ++-- document/graph-storage/src/session.rs | 36 +++++++++++++++++------- document/graph-storage/src/tests/crdt.rs | 30 +++++++++++++++----- 6 files changed, 61 insertions(+), 29 deletions(-) diff --git a/document/format/src/lib.rs b/document/format/src/lib.rs index 295354bf772..eee3178987c 100644 --- a/document/format/src/lib.rs +++ b/document/format/src/lib.rs @@ -158,7 +158,7 @@ impl Gdd { if let Some(rev) = session_state.last_broadcast_rev { session.publish_up_to(rev); } - // Likewise the authored-op count, so a reopen through any arm continues this peer's run. + // 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/session_state.rs b/document/format/src/session_state.rs index ee271e0bd30..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; @@ -43,5 +43,5 @@ pub struct SessionState { /// 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: u64, + pub next_hot_sequence: HotSequence, } diff --git a/document/format/tests/open_create.rs b/document/format/tests/open_create.rs index 270db58b2e3..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,7 +128,7 @@ fn apply_hot_op_persists_to_hot_log_and_survives_reopen() { network: Network::default(), }, timestamp: TimeStamp { counter: 1, peer: PeerId(5) }, - sequence: 1, + sequence: HotSequence(1), }; gdd.apply_hot_op(hot_op).unwrap_or_else(|error| panic!("apply_hot_op failed: {error:?}")); @@ -155,7 +155,7 @@ fn retire_moves_eligible_hot_ops_to_history_and_keeps_rest() { network: Network::default(), }, timestamp: TimeStamp { counter: 1, peer: PeerId(5) }, - sequence: 1, + sequence: HotSequence(1), }; let late = HotOp { op: RegistryDelta::AddNetwork { @@ -163,7 +163,7 @@ fn retire_moves_eligible_hot_ops_to_history_and_keeps_rest() { network: Network::default(), }, timestamp: TimeStamp { counter: 10, peer: PeerId(5) }, - sequence: 2, + sequence: HotSequence(2), }; gdd.apply_hot_op(early).unwrap(); gdd.apply_hot_op(late).unwrap(); @@ -203,7 +203,7 @@ fn last_broadcast_rev_persists_across_reopen() { network: Network::default(), }, timestamp: TimeStamp { counter: 1, peer: PeerId(5) }, - sequence: 1, + 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:?}")); @@ -659,7 +659,7 @@ fn persist_path_writes_at_manifest_declared_codec_paths() { network: Network::default(), }, timestamp: TimeStamp { counter: 1, peer: PeerId(5) }, - sequence: 1, + 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 3f3de709e48..969294437fa 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, HotOpId, LamportClock, MAX_EXPORT_SLOTS, NetworkId, NodeId, NodeInput, PeerId, Registry, RegistryDelta, ResourceEntry, RetiredMarks, 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, RetiredMarks, + Rev, SourceValue, TimeStamp, apply_attribute_delta, reverse_attribute_delta, }; #[derive(Clone, Debug)] @@ -41,7 +41,7 @@ pub struct Document { /// 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: u64, + 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, diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index 8b49d34930f..aba8c4ab5a8 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -46,7 +46,7 @@ impl Session { peer, last_broadcast_rev: None, next_node_counter: 0, - next_hot_sequence: 0, + next_hot_sequence: HotSequence::NONE, refold_owed: false, }, remote_tips: HashMap::new(), @@ -172,7 +172,7 @@ 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 + 1; + let sequence = self.document.next_hot_sequence.next(); let hot_op = HotOp { op, timestamp: self.document.clock.tick(), @@ -258,7 +258,7 @@ impl Session { peer, last_broadcast_rev: None, next_node_counter, - next_hot_sequence: 0, + next_hot_sequence: HotSequence::NONE, refold_owed: false, }, remote_tips: HashMap::new(), @@ -701,13 +701,13 @@ impl Session { } /// How many hot ops this peer has authored. Carried across a reload; no sequence is spent twice. - pub fn next_hot_sequence(&self) -> u64 { + 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: u64) { + pub fn restore_hot_sequence(&mut self, sequence: HotSequence) { self.document.next_hot_sequence = self.document.next_hot_sequence.max(sequence); } } @@ -754,7 +754,7 @@ pub struct HotOp { 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: u64, + pub sequence: HotSequence, } impl HotOp { @@ -767,12 +767,28 @@ impl HotOp { } } +/// 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: u64, + pub sequence: HotSequence, } /// Which hot ops history already covers: `through` is each author's gap-free retired prefix, `above` the @@ -780,7 +796,7 @@ pub struct HotOpId { /// still owed to it. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RetiredMarks { - pub through: HashMap, + pub through: HashMap, pub above: HashSet, } @@ -810,7 +826,7 @@ impl RetiredMarks { /// Fold exceptions that continue their author's prefix into `through`, leaving only those past a gap. fn compact(&mut self) { - let mut by_author: HashMap> = HashMap::new(); + let mut by_author: HashMap> = HashMap::new(); for id in &self.above { by_author.entry(id.peer).or_default().push(id.sequence); } @@ -819,7 +835,7 @@ impl RetiredMarks { sequences.sort_unstable(); let through = self.through.entry(peer).or_default(); for sequence in sequences { - if sequence == *through + 1 { + if sequence == through.next() { *through = sequence; } } diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index 977351b3a4f..e8f21bd0ece 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -40,7 +40,7 @@ fn apply_hot_op_advances_clock_past_observed_timestamp() { let hot_op = HotOp { op: remove_node_op(NodeId(99)), timestamp: observed, - sequence: 1, + sequence: crate::HotSequence(1), }; document.apply_hot_op(hot_op).expect("RemoveNode on absent node is a no-op, not an error"); @@ -981,22 +981,25 @@ fn a_concurrent_remove_and_attribute_change_commute() { #[test] fn retired_marks_cover_a_gap_and_compact_once_it_fills() { let author = PeerId(1); - let id = |sequence| crate::HotOpId { peer: author, sequence }; + let id = |sequence: u64| crate::HotOpId { + peer: author, + sequence: crate::HotSequence(sequence), + }; let mut marks = crate::RetiredMarks::default(); marks.extend([id(1), id(2)]); - assert_eq!(marks.through.get(&author), Some(&2), "a contiguous run folds straight into the prefix"); + assert_eq!(marks.through.get(&author), Some(&crate::HotSequence(2)), "a contiguous run folds straight into the prefix"); assert!(marks.above.is_empty()); // Sequence 3 never arrived, so 4 retires above the prefix rather than extending it. marks.extend([id(4)]); - assert_eq!(marks.through.get(&author), Some(&2)); + assert_eq!(marks.through.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.through.get(&author), Some(&4)); + assert_eq!(marks.through.get(&author), Some(&crate::HotSequence(4))); assert!(marks.above.is_empty(), "the exception set empties once the prefix reaches it"); } @@ -1004,7 +1007,10 @@ fn retired_marks_cover_a_gap_and_compact_once_it_fills() { #[test] fn absorbing_marks_keeps_both_sides_coverage() { let author = PeerId(1); - let id = |sequence| crate::HotOpId { peer: author, sequence }; + let id = |sequence: u64| crate::HotOpId { + peer: author, + sequence: crate::HotSequence(sequence), + }; let mut local = crate::RetiredMarks::default(); local.extend([id(1), id(4)]); @@ -1015,7 +1021,7 @@ fn absorbing_marks_keeps_both_sides_coverage() { local.absorb(&remote); // The union is 1..=5 contiguous, so it all collapses into the prefix. - assert_eq!(local.through.get(&author), Some(&5)); + assert_eq!(local.through.get(&author), Some(&crate::HotSequence(5))); assert!(local.above.is_empty()); for sequence in 1..=5 { assert!(local.covers(id(sequence)), "sequence {sequence} must stay covered"); @@ -1129,3 +1135,13 @@ fn an_owed_refold_is_retried_by_a_merge_that_absorbs_nothing() { "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)); +} From c408e6ed69340d70547c917fd19387290a93bc08 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Wed, 23 Sep 2026 20:57:17 +0200 Subject: [PATCH 07/11] Record what the retired-beyond set actually bounds --- document/graph-storage/src/document.rs | 8 +-- document/graph-storage/src/session.rs | 52 ++++++++++--------- document/graph-storage/src/tests/crdt.rs | 66 ++++++++++++++++++++---- 3 files changed, 88 insertions(+), 38 deletions(-) diff --git a/document/graph-storage/src/document.rs b/document/graph-storage/src/document.rs index 969294437fa..b37b5330f3d 100644 --- a/document/graph-storage/src/document.rs +++ b/document/graph-storage/src/document.rs @@ -1,5 +1,5 @@ use crate::{ - CrdtError, Delta, ExportSlot, History, HotOp, HotOpId, HotSequence, LamportClock, MAX_EXPORT_SLOTS, NetworkId, NodeId, NodeInput, PeerId, Registry, RegistryDelta, ResourceEntry, RetiredMarks, + 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, }; @@ -13,8 +13,8 @@ pub struct Document { 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 [`RetiredMarks`]. - pub(crate) retired: RetiredMarks, + /// 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 @@ -142,7 +142,7 @@ impl Document { } /// Take on a peer's retirement marks as well as this peer's. - pub(crate) fn absorb_retired(&mut self, remote: &RetiredMarks) { + pub(crate) fn absorb_retired(&mut self, remote: &RetiredHotOps) { self.retired.absorb(remote); self.drop_retired_hot_ops(); diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index aba8c4ab5a8..96712c60673 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -39,7 +39,7 @@ impl Session { retired_snapshot: Registry::default(), history: History::new(), hot_log: Vec::new(), - retired: RetiredMarks::default(), + retired: RetiredHotOps::default(), head: None, redo_stack: Vec::new(), clock: LamportClock::new(peer), @@ -251,7 +251,7 @@ impl Session { working_registry: registry, history: History::from_ordered(history), hot_log: Vec::new(), - retired: RetiredMarks::default(), + retired: RetiredHotOps::default(), head, redo_stack, clock, @@ -303,13 +303,13 @@ impl Session { self.document.mark_retired(retired.iter().copied()); } - /// Which hot ops history already covers, for a peer catching up. See [`RetiredMarks`]. - pub fn retired_marks(&self) -> &RetiredMarks { + /// Which hot ops history already covers, for a peer catching up. See [`RetiredHotOps`]. + pub fn retired_marks(&self) -> &RetiredHotOps { &self.document.retired } - /// See [`RetiredMarks::absorb`]. - pub fn absorb_retired_marks(&mut self, remote: &RetiredMarks) { + /// See [`RetiredHotOps::absorb`]. + pub fn absorb_retired_marks(&mut self, remote: &RetiredHotOps) { self.document.absorb_retired(remote); } @@ -791,49 +791,53 @@ pub struct HotOpId { pub sequence: HotSequence, } -/// Which hot ops history already covers: `through` is each author's gap-free retired prefix, `above` the -/// retired ops past it. Replicated, letting a peer catching up separate an op already in history from one -/// still owed to it. +/// 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 RetiredMarks { - pub through: HashMap, - pub above: HashSet, +pub struct RetiredHotOps { + pub retired_up_to: HashMap, + pub unretired: HashSet, } -impl RetiredMarks { +impl RetiredHotOps { /// Whether history already holds this hot op. pub fn covers(&self, id: HotOpId) -> bool { - self.through.get(&id.peer).is_some_and(|&through| id.sequence <= through) || self.above.contains(&id) + self.retired_up_to.get(&id.peer).is_some_and(|&through| id.sequence <= through) || self.unretired.contains(&id) } /// Take on `remote`'s coverage as well as this one's. pub fn absorb(&mut self, remote: &Self) { - for (&peer, &remote_through) in &remote.through { - let through = self.through.entry(peer).or_default(); + for (&peer, &remote_through) in &remote.retired_up_to { + let through = self.retired_up_to.entry(peer).or_default(); *through = (*through).max(remote_through); } - self.above.extend(remote.above.iter().copied()); + self.unretired.extend(remote.unretired.iter().copied()); self.compact(); } /// Record newly retired ops. pub fn extend(&mut self, retired: impl IntoIterator) { - self.above.extend(retired); + self.unretired.extend(retired); self.compact(); } - /// Fold exceptions that continue their author's prefix into `through`, leaving only those past a gap. + /// Fold ops that continue their author's prefix into `retired_up_to`, leaving only those past a gap. + /// + /// A gap that never fills holds the prefix still, so every later op from that author stays here for + /// the document's life. Advancing past a gap whose author can no longer supply it is what would + /// bound this, and needs connectivity the storage layer does not have. fn compact(&mut self) { let mut by_author: HashMap> = HashMap::new(); - for id in &self.above { + for id in &self.unretired { by_author.entry(id.peer).or_default().push(id.sequence); } for (peer, mut sequences) in by_author { sequences.sort_unstable(); - let through = self.through.entry(peer).or_default(); + let through = self.retired_up_to.entry(peer).or_default(); for sequence in sequences { if sequence == through.next() { *through = sequence; @@ -841,9 +845,9 @@ impl RetiredMarks { } } - let through = std::mem::take(&mut self.through); - self.above.retain(|id| through.get(&id.peer).is_none_or(|&covered| id.sequence > covered)); - self.through = through; + let through = std::mem::take(&mut self.retired_up_to); + self.unretired.retain(|id| through.get(&id.peer).is_none_or(|&covered| id.sequence > covered)); + self.retired_up_to = through; } } diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index e8f21bd0ece..527f64b96ca 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -986,21 +986,21 @@ fn retired_marks_cover_a_gap_and_compact_once_it_fills() { sequence: crate::HotSequence(sequence), }; - let mut marks = crate::RetiredMarks::default(); + let mut marks = crate::RetiredHotOps::default(); marks.extend([id(1), id(2)]); - assert_eq!(marks.through.get(&author), Some(&crate::HotSequence(2)), "a contiguous run folds straight into the prefix"); - assert!(marks.above.is_empty()); + assert_eq!(marks.retired_up_to.get(&author), Some(&crate::HotSequence(2)), "a contiguous run folds straight into the prefix"); + assert!(marks.unretired.is_empty()); // Sequence 3 never arrived, so 4 retires above the prefix rather than extending it. marks.extend([id(4)]); - assert_eq!(marks.through.get(&author), Some(&crate::HotSequence(2))); + 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.through.get(&author), Some(&crate::HotSequence(4))); - assert!(marks.above.is_empty(), "the exception set empties once the prefix reaches it"); + assert_eq!(marks.retired_up_to.get(&author), Some(&crate::HotSequence(4))); + assert!(marks.unretired.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. @@ -1012,17 +1012,17 @@ fn absorbing_marks_keeps_both_sides_coverage() { sequence: crate::HotSequence(sequence), }; - let mut local = crate::RetiredMarks::default(); + let mut local = crate::RetiredHotOps::default(); local.extend([id(1), id(4)]); - let mut remote = crate::RetiredMarks::default(); + 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.through.get(&author), Some(&crate::HotSequence(5))); - assert!(local.above.is_empty()); + assert_eq!(local.retired_up_to.get(&author), Some(&crate::HotSequence(5))); + assert!(local.unretired.is_empty()); for sequence in 1..=5 { assert!(local.covers(id(sequence)), "sequence {sequence} must stay covered"); } @@ -1145,3 +1145,49 @@ fn hot_sequence_serializes_as_a_bare_number() { 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.unretired.contains(&second.id()), "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 stays in the set for the +/// document's life. Documents the current limit; bounding it needs retirement to advance past a gap whose +/// author can no longer supply the missing op. +#[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)]); + } + + eprintln!("prefix {:?}, exception set holds {}", marks.retired_up_to.get(&author), marks.unretired.len()); + assert_eq!(marks.unretired.len(), 19, "every op after the gap stayed in the set"); +} From 7b10164b726bb06dc2169bf793e79ace3f52f437 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Thu, 24 Sep 2026 00:34:28 +0200 Subject: [PATCH 08/11] Store retired ops past a prefix as runs so a permanent gap cannot grow them --- document/graph-storage/src/session.rs | 67 ++++++++++++++++-------- document/graph-storage/src/tests/crdt.rs | 21 ++++---- 2 files changed, 56 insertions(+), 32 deletions(-) diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index 96712c60673..04765bfd900 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -797,13 +797,37 @@ pub struct HotOpId { #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RetiredHotOps { pub retired_up_to: HashMap, - pub unretired: HashSet, + /// 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 { - self.retired_up_to.get(&id.peer).is_some_and(|&through| id.sequence <= through) || self.unretired.contains(&id) + 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. @@ -812,42 +836,39 @@ impl RetiredHotOps { let through = self.retired_up_to.entry(peer).or_default(); *through = (*through).max(remote_through); } - self.unretired.extend(remote.unretired.iter().copied()); + 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) { - self.unretired.extend(retired); + for id in retired { + self.retired_beyond.entry(id.peer).or_default().push((id.sequence, id.sequence)); + } self.compact(); } - /// Fold ops that continue their author's prefix into `retired_up_to`, leaving only those past a gap. - /// - /// A gap that never fills holds the prefix still, so every later op from that author stays here for - /// the document's life. Advancing past a gap whose author can no longer supply it is what would - /// bound this, and needs connectivity the storage layer does not have. + /// Fold runs that continue their author's prefix into `retired_up_to`, leaving only those past a gap. fn compact(&mut self) { - let mut by_author: HashMap> = HashMap::new(); - for id in &self.unretired { - by_author.entry(id.peer).or_default().push(id.sequence); - } + for (&peer, runs) in &mut self.retired_beyond { + coalesce(runs); - for (peer, mut sequences) in by_author { - sequences.sort_unstable(); - let through = self.retired_up_to.entry(peer).or_default(); - for sequence in sequences { - if sequence == through.next() { - *through = sequence; - } + 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); } } - let through = std::mem::take(&mut self.retired_up_to); - self.unretired.retain(|id| through.get(&id.peer).is_none_or(|&covered| id.sequence > covered)); - self.retired_up_to = through; + self.retired_beyond.retain(|_, runs| !runs.is_empty()); } } diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index 527f64b96ca..e025cd1d9b0 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -989,7 +989,7 @@ fn retired_marks_cover_a_gap_and_compact_once_it_fills() { 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.unretired.is_empty()); + assert!(marks.retired_beyond.is_empty()); // Sequence 3 never arrived, so 4 retires above the prefix rather than extending it. marks.extend([id(4)]); @@ -1000,7 +1000,7 @@ fn retired_marks_cover_a_gap_and_compact_once_it_fills() { // 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.unretired.is_empty(), "the exception set empties once the prefix reaches it"); + 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. @@ -1022,7 +1022,7 @@ fn absorbing_marks_keeps_both_sides_coverage() { // 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.unretired.is_empty()); + assert!(local.retired_beyond.is_empty()); for sequence in 1..=5 { assert!(local.covers(id(sequence)), "sequence {sequence} must stay covered"); } @@ -1166,13 +1166,12 @@ fn retiring_over_a_gap_lands_beyond_the_prefix() { 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.unretired.contains(&second.id()), "the retired op has to be recorded past the gap"); + 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 stays in the set for the -/// document's life. Documents the current limit; bounding it needs retirement to advance past a gap whose -/// author can no longer supply the missing op. +/// 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); @@ -1188,6 +1187,10 @@ fn a_permanent_gap_accumulates_every_later_op() { marks.extend([id(sequence)]); } - eprintln!("prefix {:?}, exception set holds {}", marks.retired_up_to.get(&author), marks.unretired.len()); - assert_eq!(marks.unretired.len(), 19, "every op after the gap stayed in the set"); + 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"); } From dc4fa940122db1d08eb13a03fdfecb4758600036 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Thu, 24 Sep 2026 00:36:01 +0200 Subject: [PATCH 09/11] Make a peer's working registry agree with its own retired snapshot --- document/graph-storage/src/document.rs | 19 +++++--- document/graph-storage/src/history.rs | 5 ++ document/graph-storage/src/session.rs | 56 ++++++++++++++++----- document/graph-storage/src/tests/crdt.rs | 62 ++++++++++++++++++++++++ 4 files changed, 123 insertions(+), 19 deletions(-) diff --git a/document/graph-storage/src/document.rs b/document/graph-storage/src/document.rs index b37b5330f3d..dfda39e1847 100644 --- a/document/graph-storage/src/document.rs +++ b/document/graph-storage/src/document.rs @@ -141,18 +141,21 @@ impl Document { self.drop_retired_hot_ops(); } - /// Take on a peer's retirement marks as well as this peer's. - pub(crate) fn absorb_retired(&mut self, remote: &RetiredHotOps) { + /// 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(); + self.drop_retired_hot_ops() } - /// Drop hot ops history already covers, whatever path put them in the log. - fn drop_retired_hot_ops(&mut self) { - let retired = std::mem::take(&mut self.retired); - self.hot_log.retain(|hot_op| !retired.covers(hot_op.id())); - self.retired = retired; + /// 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 diff --git a/document/graph-storage/src/history.rs b/document/graph-storage/src/history.rs index fafa4a9e9c6..941aef16b46 100644 --- a/document/graph-storage/src/history.rs +++ b/document/graph-storage/src/history.rs @@ -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() } diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index 04765bfd900..eef3bb943c0 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -196,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. @@ -211,9 +218,9 @@ impl Session { // 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; @@ -296,11 +303,19 @@ impl Session { 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. - pub fn discard_hot_ops(&mut self, retired: &[HotOpId]) { + /// 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`]. @@ -308,9 +323,14 @@ impl Session { &self.document.retired } - /// See [`RetiredHotOps::absorb`]. - pub fn absorb_retired_marks(&mut self, remote: &RetiredHotOps) { - self.document.absorb_retired(remote); + /// 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 @@ -431,25 +451,39 @@ impl Session { 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)); - self.commit_ops(drained.into_iter().map(|hot_op| hot_op.op), true) + 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()?; + } + + Ok(revs) } /// Mark a retired delta as the end of a user interaction, so the undo cursor treats it as a checkpoint. diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index e025cd1d9b0..cb04ed8ab27 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -1194,3 +1194,65 @@ fn a_permanent_gap_accumulates_every_later_op() { } 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"); +} From 3b9474e3467b571ef866494dadb188ceae7ea251 Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Thu, 24 Sep 2026 00:47:19 +0200 Subject: [PATCH 10/11] Stamp a freshly created node's implementation at the origin timestamp --- document/graph-storage/src/model.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/document/graph-storage/src/model.rs b/document/graph-storage/src/model.rs index 94762ca6cff..bfd65b57482 100644 --- a/document/graph-storage/src/model.rs +++ b/document/graph-storage/src/model.rs @@ -64,6 +64,7 @@ impl Node { Self { implementation, + implementation_timestamp: TimeStamp::ORIGIN, inputs: vec![slot; inputs], attributes: Attributes::new(), network, From 4b2a63d0f33149f0f9bc02cfc0647c7c08bf7c1b Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Thu, 24 Sep 2026 15:57:57 +0000 Subject: [PATCH 11/11] Resurrect from the replay position when folding a snapshot from history --- document/graph-storage/src/session.rs | 17 +++---- document/graph-storage/src/tests/crdt.rs | 63 ++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 9 deletions(-) diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index eef3bb943c0..99421ac569f 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -342,21 +342,20 @@ impl Session { /// 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(); - let replay: Vec<(RegistryDelta, TimeStamp)> = folded - .document - .history - .iter() - .filter(|delta| reachable.contains(&delta.id)) - .map(|delta| (delta.kind.clone(), delta.timestamp)) - .collect(); - for (kind, timestamp) in replay { - folded.document.apply_op_with(RegistryTarget::Snapshot, kind, timestamp, ApplyMode::Idempotent)?; + 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) diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index cb04ed8ab27..d45b31edeb3 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -1115,6 +1115,69 @@ fn snapshot_from_history_ignores_undone_deltas() { 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]