Make retired state a function of history alone by hardening CRDT logic

document/format/src/export.rs

@@ -150,6 +150,7 @@ impl<L: Layout> Gdd<L> {
 			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(),
 		};

document/format/src/lib.rs

@@ -158,6 +158,8 @@ impl<L: Layout> Gdd<L> {
 		if let Some(rev) = session_state.last_broadcast_rev {
 			session.publish_up_to(rev);
 		}
+		// Every arm must continue this peer's authored-op count too.
+		session.restore_hot_sequence(session_state.next_hot_sequence);
 
 		replay_hot_log(&working, &layout, codecs.hot_log, &mut session).await?;
 

document/format/src/persist.rs

@@ -198,6 +198,7 @@ impl<L: Layout> Gdd<L> {
 			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(),
 		};

document/format/src/session_state.rs

@@ -4,7 +4,7 @@
 //!
 //! Lives in `session.json`. Rewritten on retirement.
 
-use document_graph_storage::{NetworkId, PeerId, Rev};
+use document_graph_storage::{HotSequence, NetworkId, PeerId, Rev};
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
 
@@ -39,4 +39,9 @@ pub struct SessionState {
 	/// Per-peer like [`view_settings`](Self::view_settings); opaque `ui::nav::*` / `ui::previewing` blobs.
 	#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
 	pub network_view_settings: BTreeMap<NetworkId, BTreeMap<String, serde_json::Value>>,
+	/// How many hot ops this peer has authored, feeding [`document_graph_storage::HotOp::sequence`]. A
+	/// reopen continues the run instead of reusing a spent sequence. Appended last: a positional codec
+	/// decodes these fields in declaration order.
+	#[serde(default)]
+	pub next_hot_sequence: HotSequence,
 }

document/format/tests/open_create.rs

@@ -5,7 +5,7 @@
 use document_container::AnyContainer;
 use document_container::backends::memory::MemoryBackend;
 use document_format::{Codec, Error, GddV1, GddV1Layout, Layout, Manifest, io, manifest};
-use document_graph_storage::{HotOp, Network, NetworkId, PeerId, ROOT_NETWORK, RegistryDelta, TimeStamp};
+use document_graph_storage::{HotOp, HotSequence, Network, NetworkId, PeerId, ROOT_NETWORK, RegistryDelta, TimeStamp};
 
 fn empty_container() -> AnyContainer {
 	AnyContainer::Memory(MemoryBackend::new())
@@ -128,6 +128,7 @@ fn apply_hot_op_persists_to_hot_log_and_survives_reopen() {
 				network: Network::default(),
 			},
 			timestamp: TimeStamp { counter: 1, peer: PeerId(5) },
+			sequence: HotSequence(1),
 		};
 		gdd.apply_hot_op(hot_op).unwrap_or_else(|error| panic!("apply_hot_op failed: {error:?}"));
 
@@ -154,13 +155,15 @@ fn retire_moves_eligible_hot_ops_to_history_and_keeps_rest() {
 				network: Network::default(),
 			},
 			timestamp: TimeStamp { counter: 1, peer: PeerId(5) },
+			sequence: HotSequence(1),
 		};
 		let late = HotOp {
 			op: RegistryDelta::AddNetwork {
 				id: NetworkId(42),
 				network: Network::default(),
 			},
 			timestamp: TimeStamp { counter: 10, peer: PeerId(5) },
+			sequence: HotSequence(2),
 		};
 		gdd.apply_hot_op(early).unwrap();
 		gdd.apply_hot_op(late).unwrap();
@@ -200,6 +203,7 @@ fn last_broadcast_rev_persists_across_reopen() {
 				network: Network::default(),
 			},
 			timestamp: TimeStamp { counter: 1, peer: PeerId(5) },
+			sequence: HotSequence(1),
 		};
 		gdd.apply_hot_op(op).unwrap();
 		let retired = gdd.retire(TimeStamp { counter: 1, peer: PeerId(5) }).unwrap_or_else(|error| panic!("retire failed: {error:?}"));
@@ -655,6 +659,7 @@ fn persist_path_writes_at_manifest_declared_codec_paths() {
 				network: Network::default(),
 			},
 			timestamp: TimeStamp { counter: 1, peer: PeerId(5) },
+			sequence: HotSequence(1),
 		};
 		gdd.apply_hot_op(hot_op).unwrap_or_else(|error| panic!("apply_hot_op failed: {error:?}"));
 

document/graph-storage/src/document.rs

@@ -1,6 +1,6 @@
 use crate::{
-	CrdtError, Delta, ExportSlot, History, HotOp, LamportClock, MAX_EXPORT_SLOTS, NetworkId, NodeId, NodeInput, PeerId, Registry, RegistryDelta, ResourceEntry, Rev, SourceValue, TimeStamp,
-	apply_attribute_delta, reverse_attribute_delta,
+	CrdtError, Delta, ExportSlot, History, HotOp, HotOpId, HotSequence, LamportClock, MAX_EXPORT_SLOTS, NetworkId, NodeId, NodeInput, PeerId, Registry, RegistryDelta, ResourceEntry, RetiredHotOps,
+	Rev, SourceValue, TimeStamp, apply_attribute_delta, reverse_attribute_delta,
 };
 
 #[derive(Clone, Debug)]
@@ -11,6 +11,10 @@ pub struct Document {
 	/// Live broadcast stream, applied to the `working_registry` on receive, GC'd at retirement.
 	/// Persisted for crash recovery so in-flight unretired work survives editor restarts.
 	pub(crate) hot_log: Vec<HotOp>,
+	/// Which hot ops history already covers, so one arriving after its own retirement is dropped instead
+	/// of re-entering the log. Retirement discards the delta's link back to its hot op, leaving this the
+	/// only record. See [`RetiredHotOps`].
+	pub(crate) retired: RetiredHotOps,
 	/// The registry as of the last retirement, with no un-retired hot ops applied. Retirement computes
 	/// each delta's `reverse` against this (so LWW reverses capture the true pre-op value, not the
 	/// hot-polluted working state) and advances it, stamping fields at the fresh `T_retire`. Kept equal
@@ -36,6 +40,11 @@ pub struct Document {
 	/// peer is calling; collision avoidance comes from hashing `(self.peer, counter)`, so two peers
 	/// reading the same counter still produce distinct IDs.
 	pub(crate) next_node_counter: u64,
+	/// Counts this peer's own hot ops, so each carries its position in a gap-free run. See [`HotOp::sequence`].
+	pub(crate) next_hot_sequence: HotSequence,
+	/// Set while the registries are still folded from an older history. A failed refold is retried; the
+	/// derived state does not stay behind the history it comes from.
+	pub(crate) refold_owed: bool,
 }
 
 impl Document {
@@ -66,24 +75,12 @@ impl Document {
 		self.revert_delta(target, delta)
 	}
 
-	/// Search every delta reachable from `head` (following all parents, including a merge's
-	/// `extra_parents`) for the first matching `predicate`, breadth-first. Resurrection needs full
-	/// ancestry reachability, so a node added only on a merged-in branch is still found.
+	/// The last match in canonical history order, over everything reachable from `head` or any tip. Not a
+	/// walk from `head`: peers sit on different merge revs, and resurrection must pick the same delta.
 	fn find_in_ancestry(&self, predicate: impl Fn(&Delta) -> bool) -> Option<Delta> {
-		let mut queue: std::collections::VecDeque<Rev> = self.head.into_iter().collect();
-		let mut seen: std::collections::HashSet<Rev> = self.head.into_iter().collect();
-		while let Some(rev) = queue.pop_front() {
-			let Some(delta) = self.history.get(rev) else { continue };
-			if predicate(delta) {
-				return Some(delta.clone());
-			}
-			for parent in delta.all_parents() {
-				if seen.insert(parent) {
-					queue.push_back(parent);
-				}
-			}
-		}
-		None
+		let reachable = self.history.ancestors(self.head.into_iter().chain(self.history.tips()));
+
+		self.history.iter().filter(|delta| reachable.contains(&delta.id) && predicate(delta)).last().cloned()
 	}
 
 	/// Apply a delta's `reverse` as the new forward op (silent-zone undo). Force-applied: structural
@@ -99,9 +96,10 @@ impl Document {
 		self.apply_op_with(target, delta.kind, delta.timestamp, ApplyMode::Force)
 	}
 
-	/// Apply a live broadcast op. Updates the registry via LWW and appends to the hot log.
-	/// Doesn't touch history or `head` — hot ops are transient.
-	pub fn apply_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> {
+	/// Apply a locally staged op: LWW into the registry, append to the hot log, leave history and `head`
+	/// alone. Crate-private because it skips the retirement check an op off the wire needs; staging must
+	/// not have a covered sequence silently drop local work. Outside callers use [`Session::apply_hot_op`].
+	pub(crate) fn apply_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> {
 		self.apply_op(hot_op.op.clone(), hot_op.timestamp)?;
 		self.hot_log.push(hot_op);
 		Ok(())
@@ -111,11 +109,55 @@ impl Document {
 	/// re-applying an op whose effect is already reflected in the registry is a no-op rather
 	/// than an error.
 	pub fn replay_hot_op(&mut self, hot_op: HotOp) -> Result<(), CrdtError> {
+		// Already in retired history; re-adding it would leave an entry no retirement will ever name.
+		if self.is_retired(hot_op.id()) {
+			return Ok(());
+		}
+
+		// Identified by timestamp, so a re-announcement of one already held must not become a second copy.
+		if self.hot_log.iter().any(|held| held.timestamp == hot_op.timestamp) {
+			return Ok(());
+		}
+
+		// Replaying our own ops is what carries the sequence counter across a reload.
+		if hot_op.timestamp.peer == self.peer {
+			self.next_hot_sequence = self.next_hot_sequence.max(hot_op.sequence);
+		}
+
 		self.apply_op_idempotent(hot_op.op.clone(), hot_op.timestamp)?;
 		self.hot_log.push(hot_op);
 		Ok(())
 	}
 
+	/// Whether this hot op has already been promoted into history.
+	pub(crate) fn is_retired(&self, id: HotOpId) -> bool {
+		self.retired.covers(id)
+	}
+
+	/// Record newly retired hot ops, dropping any the hot log still holds.
+	pub(crate) fn mark_retired(&mut self, retired: impl IntoIterator<Item = HotOpId>) {
+		self.retired.extend(retired);
+
+		self.drop_retired_hot_ops();
+	}
+
+	/// Take on a peer's retirement marks as well as this peer's. Returns whether a hot op was dropped,
+	/// which owes the caller a refold.
+	pub(crate) fn absorb_retired(&mut self, remote: &RetiredHotOps) -> bool {
+		self.retired.absorb(remote);
+
+		self.drop_retired_hot_ops()
+	}
+
+	/// Drop hot ops the retired snapshot already accounts for. A mark alone will not do: it can arrive
+	/// ahead of the delta carrying the op into history, stranding its effect in the working registry.
+	fn drop_retired_hot_ops(&mut self) -> bool {
+		let before = self.hot_log.len();
+		self.hot_log.retain(|hot_op| !self.history.contains_timestamp(hot_op.timestamp));
+
+		before != self.hot_log.len()
+	}
+
 	/// Apply a retired commit. Idempotent on structural ops (AddNode/AddNetwork on existing
 	/// targets, Remove on missing ones) since hot ops already produced the structural state.
 	/// The point is to bump field timestamps to T_retire via the LWW arms.
@@ -315,7 +357,7 @@ impl Document {
 	/// Resurrect (from history) any nodes/networks an op references that were concurrently removed, so
 	/// the op applies against a consistent registry. Cascading: a node's owning network is restored
 	/// before the node. No-op for ops that reference nothing absent.
-	fn ensure_referenced_exist(&mut self, target: RegistryTarget, op: &RegistryDelta) -> Result<(), CrdtError> {
+	pub(crate) fn ensure_referenced_exist(&mut self, target: RegistryTarget, op: &RegistryDelta) -> Result<(), CrdtError> {
 		match op {
 			RegistryDelta::AddNode { node, .. } => self.ensure_network_exists(target, node.network())?,
 			RegistryDelta::ChangeNodeInput { id, new_input, .. } => {
@@ -354,17 +396,51 @@ impl Document {
 	}
 
 	fn ensure_node_exists(&mut self, target: RegistryTarget, node_id: NodeId) -> Result<(), CrdtError> {
-		if !self.registry_ref(target).node_instances.contains_key(&node_id) {
-			self.restore_node_from_history(target, node_id)?;
+		if self.registry_ref(target).node_instances.contains_key(&node_id) {
+			return Ok(());
 		}
-		Ok(())
+		let removal = self.hot_log_node_removal(target, node_id);
+		match removal {
+			Some((revive, timestamp)) => self.apply_op_with(target, revive, timestamp, ApplyMode::Force),
+			None => self.restore_node_from_history(target, node_id),
+		}
+	}
+
+	/// The most recent hot removal of a network, for reviving it in the working zone. `None` for the
+	/// snapshot, which stays a function of history alone.
+	fn hot_log_removal(&self, target: RegistryTarget, network_id: NetworkId) -> Option<(RegistryDelta, TimeStamp)> {
+		if target != RegistryTarget::Working {
+			return None;
+		}
+
+		self.hot_log.iter().rev().find_map(|hot_op| match &hot_op.op {
+			RegistryDelta::RemoveNetwork { id, snapshot } if *id == network_id => Some((RegistryDelta::AddNetwork { id: *id, network: snapshot.clone() }, hot_op.timestamp)),
+			_ => None,
+		})
+	}
+
+	/// The node equivalent of [`hot_log_removal`](Self::hot_log_removal).
+	fn hot_log_node_removal(&self, target: RegistryTarget, node_id: NodeId) -> Option<(RegistryDelta, TimeStamp)> {
+		if target != RegistryTarget::Working {
+			return None;
+		}
+
+		self.hot_log.iter().rev().find_map(|hot_op| match &hot_op.op {
+			RegistryDelta::RemoveNode { id, snapshot } if *id == node_id => Some((RegistryDelta::AddNode { id: *id, node: snapshot.clone() }, hot_op.timestamp)),
+			_ => None,
+		})
 	}
 
 	fn ensure_network_exists(&mut self, target: RegistryTarget, network_id: NetworkId) -> Result<(), CrdtError> {
-		if !self.registry_ref(target).networks.contains_key(&network_id) {
-			self.restore_network_from_history(target, network_id)?;
+		if self.registry_ref(target).networks.contains_key(&network_id) {
+			return Ok(());
+		}
+		// The hot log is peer-local, so reviving retired state from it would make it peer-dependent.
+		let removal = self.hot_log_removal(target, network_id);
+		match removal {
+			Some((revive, timestamp)) => self.apply_op_with(target, revive, timestamp, ApplyMode::Force),
+			None => self.restore_network_from_history(target, network_id),
 		}
-		Ok(())
 	}
 
 	/// Compute the inverse of `delta` against the registry named by `target`. Retirement passes

document/graph-storage/src/history.rs

@@ -8,7 +8,7 @@
 //! construction. The only operation that introduces out-of-order deltas is [`merge`](History::merge),
 //! which re-sorts the combined set into the canonical order to restore the invariant.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use crate::{AttributesWrite, CrdtError, Delta, Rev, TimeStamp};
 
@@ -39,6 +39,11 @@ impl History {
 		self.index.contains_key(&rev)
 	}
 
+	/// Whether history holds a delta authored at `timestamp`.
+	pub fn contains_timestamp(&self, timestamp: TimeStamp) -> bool {
+		self.deltas.iter().any(|delta| delta.timestamp == timestamp)
+	}
+
 	pub fn len(&self) -> usize {
 		self.deltas.len()
 	}
@@ -64,22 +69,12 @@ impl History {
 		self.deltas.iter()
 	}
 
-	/// Absorb `incoming` (dedup by `Rev`) and canonically re-sort the whole combined history.
-	///
-	/// The sort is deterministic (topological, ties broken by `Rev`), so two peers that absorb the same
-	/// delta set produce byte-identical history, not merely two different valid orderings. This is the
-	/// history-convergence mechanism: arrival order is erased. Callers update the registry separately
-	/// (LWW apply is commutative, so the registry converges regardless of order).
-	pub fn merge(&mut self, incoming: impl IntoIterator<Item = Delta>) {
-		for delta in incoming {
-			self.push(delta);
-		}
-		self.canonical_sort();
-	}
-
 	/// Re-order `deltas` into the canonical topological order and rebuild the index: parents precede
 	/// children, and among deltas whose parents are all emitted the lowest `Rev` goes first. O(V + E).
-	fn canonical_sort(&mut self) {
+	///
+	/// Deterministic: two peers that absorb the same delta set end up with byte-identical history, not two
+	/// different valid orderings. Arrival order is erased.
+	pub fn canonical_sort(&mut self) {
 		// Unsatisfied in-history parent count per delta, plus reverse edges to decrement as parents emit.
 		let mut pending_parents: HashMap<Rev, usize> = HashMap::with_capacity(self.deltas.len());
 		let mut children: HashMap<Rev, Vec<Rev>> = HashMap::new();
@@ -119,6 +114,51 @@ impl History {
 		}
 	}
 
+	/// `roots` and everything reachable from them through all parent links. Unknown roots are skipped.
+	pub fn ancestors(&self, roots: impl IntoIterator<Item = Rev>) -> HashSet<Rev> {
+		let mut seen = HashSet::new();
+		let mut stack: Vec<Rev> = 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<Rev> {
+		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<Item = Rev>) -> 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<Rev> {