Give hot ops shared attributes, and close an interaction with a flagged Meta op (#4753)

* Give hot ops shared attributes, and close an interaction with a flagged Meta op

* Review fixes

* Cleanup

Author: TrueDoctor

document/format/tests/open_create.rs

@@ -123,6 +123,7 @@ fn apply_hot_op_persists_to_hot_log_and_survives_reopen() {
 			},
 			timestamp: TimeStamp { counter: 1, peer: PeerId(5) },
 			sequence: HotSequence(1),
+			attributes: Default::default(),
 		};
 		gdd.apply_hot_op(hot_op).unwrap_or_else(|error| panic!("apply_hot_op failed: {error:?}"));
 
@@ -148,6 +149,7 @@ fn retire_moves_eligible_hot_ops_to_history_and_keeps_rest() {
 			},
 			timestamp: TimeStamp { counter: 1, peer: PeerId(5) },
 			sequence: HotSequence(1),
+			attributes: Default::default(),
 		};
 		let late = HotOp {
 			op: RegistryDelta::AddNetwork {
@@ -156,6 +158,7 @@ fn retire_moves_eligible_hot_ops_to_history_and_keeps_rest() {
 			},
 			timestamp: TimeStamp { counter: 10, peer: PeerId(5) },
 			sequence: HotSequence(2),
+			attributes: Default::default(),
 		};
 		gdd.apply_hot_op(early).unwrap();
 		gdd.apply_hot_op(late).unwrap();
@@ -194,6 +197,7 @@ fn last_broadcast_rev_persists_across_reopen() {
 			},
 			timestamp: TimeStamp { counter: 1, peer: PeerId(5) },
 			sequence: HotSequence(1),
+			attributes: Default::default(),
 		};
 		gdd.apply_hot_op(op).unwrap();
 		let retired = gdd.retire(TimeStamp { counter: 1, peer: PeerId(5) }).unwrap_or_else(|error| panic!("retire failed: {error:?}"));
@@ -632,6 +636,7 @@ fn persist_path_writes_at_manifest_declared_codec_paths() {
 			},
 			timestamp: TimeStamp { counter: 1, peer: PeerId(5) },
 			sequence: HotSequence(1),
+			attributes: Default::default(),
 		};
 		gdd.apply_hot_op(hot_op).unwrap_or_else(|error| panic!("apply_hot_op failed: {error:?}"));
 

document/format/tests/staging_and_reopen.rs

@@ -54,6 +54,7 @@ fn a_reopen_keeps_unretired_hot_ops_out_of_the_retired_snapshot() {
 		},
 		timestamp: TimeStamp { counter, peer: PeerId(22) },
 		sequence: HotSequence(counter),
+		attributes: Default::default(),
 	};
 	futures::executor::block_on(async {
 		let mut gdd = GddV1::create_in(AnyContainer::Memory(MemoryBackend::new()), GddV1Layout, PeerId(22), 1, "ed".into(), "std".into()).unwrap();

document/graph-storage/src/attributes/attr.rs

@@ -67,6 +67,7 @@ pub mod network {
 }
 
 pub mod delta {
-	/// Marks the last delta of a user interaction, so the undo cursor steps per-interaction, not per-delta.
+	/// Marks the last delta of a user interaction, so the undo cursor steps per-interaction, not per-delta. Also set on
+	/// the last hot op of an interaction.
 	pub const INTERACTION_END: &str = "interaction_end";
 }

document/graph-storage/src/crdt.rs

@@ -211,6 +211,7 @@ pub enum RegistryDelta {
 	Merge {
 		extra_parents: Vec<Rev>,
 	},
+	Meta,
 	// Allow for future delta types without a model change
 	Other(Value),
 }

document/graph-storage/src/document.rs

@@ -350,7 +350,7 @@ impl Document {
 				registry.attributes.apply_delta(delta, timestamp);
 			}
 			// Merge is a structural sync point only; it mutates no registry state.
-			RegistryDelta::Merge { .. } | RegistryDelta::Other(_) => {}
+			RegistryDelta::Merge { .. } | RegistryDelta::Meta | RegistryDelta::Other(_) => {}
 		}
 		Ok(())
 	}

document/graph-storage/src/prior.rs

@@ -184,7 +184,7 @@ pub(crate) fn capture(registry: &Registry, op: &RegistryDelta) -> Vec<Prior> {
 			key: delta.key.clone(),
 			previous: registry.attributes.get(&delta.key).cloned(),
 		}),
-		RegistryDelta::Merge { .. } | RegistryDelta::Other(_) => {}
+		RegistryDelta::Merge { .. } | RegistryDelta::Meta | RegistryDelta::Other(_) => {}
 	}
 	priors
 }

document/graph-storage/src/session.rs

@@ -3,7 +3,8 @@ use crate::NodeMetadataSource;
 #[cfg(any(feature = "conversion", test))]
 use crate::from_runtime;
 use crate::{
-	ApplyMode, Delta, Document, History, Implementation, LamportClock, NetworkId, NodeId, PeerId, Registry, RegistryDelta, RegistryTarget, ResourceEntry, Rev, TimeStamp, UserId, Value, to_value,
+	ApplyMode, Attributes, Delta, Document, History, Implementation, LamportClock, NetworkId, NodeId, PeerId, Registry, RegistryDelta, RegistryTarget, ResourceEntry, Rev, TimeStamp, UserId, Value,
+	to_value,
 };
 use graphene_resource::{ResourceHash, ResourceId};
 use serde::{Deserialize, Serialize};
@@ -146,15 +147,19 @@ impl Session {
 	/// `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, since registration rides a real edit.
 	pub(crate) fn stage_ops(&mut self, ops: impl IntoIterator<Item = RegistryDelta>) -> Result<Vec<HotOp>, CrdtError> {
+		self.stage_ops_with(ops.into_iter().map(|op| (op, Attributes::new())))
+	}
+
+	fn stage_ops_with(&mut self, ops: impl IntoIterator<Item = (RegistryDelta, Attributes)>) -> Result<Vec<HotOp>, CrdtError> {
 		// A batch is staged whole or not at all.
-		let mut pending: Vec<RegistryDelta> = ops.into_iter().collect();
+		let mut pending: Vec<(RegistryDelta, Attributes)> = ops.into_iter().collect();
 		if pending.is_empty() {
 			return Ok(Vec::new());
 		}
 
 		let (peer, user) = (self.document.peer, self.document.user);
 		if self.document.working_registry.peer_users.get(&peer).map(|registration| registration.user) != Some(user) {
-			pending.insert(0, RegistryDelta::RegisterPeer { peer, user });
+			pending.insert(0, (RegistryDelta::RegisterPeer { peer, user }, Attributes::new()));
 		}
 
 		// A reused sequence would fall under the retired marks, so a batch running past the end is refused.
@@ -164,12 +169,13 @@ impl Session {
 
 		let sequence_before = self.document.last_hot_sequence;
 		let mut staged = Vec::with_capacity(pending.len());
-		for op in pending {
+		for (op, attributes) in pending {
 			let sequence = HotSequence(self.document.last_hot_sequence.0 + 1);
 			let hot_op = HotOp {
 				op,
 				timestamp: self.document.clock.tick(),
 				sequence,
+				attributes,
 			};
 			if let Err(error) = self.document.stage_hot_op(hot_op.clone(), ApplyMode::Strict) {
 				// The run continues where it was, so no sequence goes unused.
@@ -344,22 +350,79 @@ impl Session {
 		self.document.history.len()
 	}
 
+	/// Closes this peer's open interaction with a flagged [`RegistryDelta::Meta`] op. Returns the staged ops, a
+	/// re-registration possibly before the marker; empty when nothing is open.
+	pub fn end_interaction(&mut self) -> Result<Vec<HotOp>, CrdtError> {
+		let peer = self.document.peer;
+		let open = self
+			.document
+			.hot_log
+			.iter()
+			.rev()
+			.find(|hot_op| hot_op.timestamp.peer == peer)
+			.is_some_and(|hot_op| !hot_op.ends_interaction());
+		if !open {
+			return Ok(Vec::new());
+		}
+		let mut closing = Attributes::new();
+		closing.set(crate::attr::delta::INTERACTION_END, Value::Bool(true), TimeStamp::ORIGIN);
+		self.stage_ops_with([(RegistryDelta::Meta, closing)])
+	}
+
+	/// Every author's closed interactions in the hot log, the earliest closed first.
+	pub fn closed_interactions(&self) -> Vec<ClosedInteraction> {
+		let mut by_author: HashMap<PeerId, Vec<&HotOp>> = HashMap::new();
+		for hot_op in &self.document.hot_log {
+			by_author.entry(hot_op.timestamp.peer).or_default().push(hot_op);
+		}
+
+		let mut closed = Vec::new();
+		for (author, mut ops) in by_author {
+			ops.sort_by_key(|hot_op| hot_op.sequence);
+			let mut expected = self.document.retired.retired_up_to.get(&author).copied().unwrap_or(HotSequence::NONE).next();
+			let mut current = Vec::new();
+			let mut contiguous = true;
+			for hot_op in ops {
+				// Retired past a gap: not in the log, but not missing either.
+				while expected < hot_op.sequence && self.document.retired.covers(HotOpId { peer: author, sequence: expected }) {
+					expected = expected.next();
+				}
+				contiguous &= hot_op.sequence == expected;
+				expected = hot_op.sequence.next();
+				current.push(hot_op.id());
+				if hot_op.ends_interaction() {
+					closed.push(ClosedInteraction {
+						author,
+						ops: std::mem::take(&mut current),
+						closed_at: hot_op.timestamp,
+						contiguous,
+					});
+				}
+			}
+		}
+		closed.sort_by_key(|interaction| interaction.closed_at);
+		closed
+	}
+
 	/// The hot ops stamped at or before `up_to`, which [`retire`](Self::retire) drains.
 	pub fn hot_ops_up_to(&self, up_to: TimeStamp) -> Vec<HotOpId> {
 		self.document.hot_log.iter().filter(|hot_op| hot_op.timestamp <= up_to).map(HotOp::id).collect()
 	}
 
 	/// Promote the given hot ops into retired deltas in hot-log order, each keeping its authored stamp.
 	///
-	/// One retired delta per hot op.
+	/// One retired delta per hot op, except a `Meta` op, which commits nothing.
 	pub fn retire_hot_ops(&mut self, ids: &[HotOpId]) -> Result<Vec<Rev>, CrdtError> {
 		let wanted: HashSet<HotOpId> = ids.iter().copied().collect();
 		let (drained, kept): (Vec<HotOp>, Vec<HotOp>) = self.document.hot_log.drain(..).partition(|hot_op| wanted.contains(&hot_op.id()));
 		self.document.hot_log = kept;
 		self.document.resync_hot_timestamps();
 		self.document.retired.extend(drained.iter().map(HotOp::id));
 
-		let ops = drained.into_iter().map(|hot_op| (hot_op.op, Some(hot_op.timestamp)));
+		let ops = drained
+			.into_iter()
+			.filter(|hot_op| !matches!(hot_op.op, RegistryDelta::Meta))
+			.map(|hot_op| (hot_op.op, Some(hot_op.timestamp)));
 		self.commit_ops_authored_at(ops, true)
 	}
 
@@ -627,9 +690,16 @@ pub struct HotOp {
 	pub timestamp: TimeStamp,
 	/// Position in its author's run. See [`HotSequence`].
 	pub sequence: HotSequence,
+	/// Shared with every peer, unlike `Delta::attributes`. Dropped at retirement.
+	#[serde(default)]
+	pub attributes: Attributes,
 }
 
 impl HotOp {
+	pub fn ends_interaction(&self) -> bool {
+		self.attributes.get_or(crate::attr::delta::INTERACTION_END, false)
+	}
+
 	/// Identifies the op in the retired marks, which track a contiguous prefix per author.
 	pub fn id(&self) -> HotOpId {
 		HotOpId {
@@ -639,6 +709,17 @@ impl HotOp {
 	}
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ClosedInteraction {
+	pub author: PeerId,
+	/// In the author's order, the marker last.
+	pub ops: Vec<HotOpId>,
+	/// The marker's stamp.
+	pub closed_at: TimeStamp,
+	/// No gap in the author's sequence since its retired frontier.
+	pub contiguous: bool,
+}
+
 /// 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)]

document/graph-storage/src/tests/bookkeeping.rs

@@ -25,6 +25,7 @@ fn retirement_takes_the_ops_under_the_cutoff_in_any_applied_order() {
 		op,
 		timestamp: crate::TimeStamp { counter, peer: PeerId(peer) },
 		sequence: crate::HotSequence(1),
+		attributes: Default::default(),
 	};
 	let (late, own) = (hot(add_network(9), 50, 2), hot(set_attribute("x", 1), 20, 3));
 	host.replay_hot_op(late.clone()).expect("the late-stamped op lands first");
@@ -83,6 +84,7 @@ fn a_reopened_session_drops_a_late_copy_of_a_retired_op() {
 		op: set_attribute("k", 1),
 		timestamp: crate::TimeStamp { counter: 3, peer: PeerId(2) },
 		sequence: crate::HotSequence(1),
+		attributes: Default::default(),
 	};
 	let mut host = Session::with_peer(PeerId(1));
 	host.replay_hot_op(op.clone()).unwrap();
@@ -165,6 +167,7 @@ fn a_retired_late_copy_still_advances_the_sequence_and_the_clock() {
 		op: set_attribute("k", 1),
 		timestamp: crate::TimeStamp { counter: 50, peer: PeerId(1) },
 		sequence: crate::HotSequence(7),
+		attributes: Default::default(),
 	};
 	let mut marks = crate::RetiredMarks::default();
 	marks.extend([own.id()]);
@@ -176,3 +179,88 @@ fn a_retired_late_copy_still_advances_the_sequence_and_the_clock() {
 	assert!(next.sequence > own.sequence, "a fresh sequence");
 	assert!(next.timestamp > own.timestamp, "a later stamp");
 }
+
+#[test]
+fn only_closed_interactions_are_listed_and_the_marker_commits_nothing() {
+	let mut host = Session::with_peer(PeerId(1));
+	let guest_op = |counter: u64, sequence: u64, op: RegistryDelta| crate::HotOp {
+		op,
+		timestamp: crate::TimeStamp { counter, peer: PeerId(2) },
+		sequence: crate::HotSequence(sequence),
+		attributes: Default::default(),
+	};
+
+	assert!(host.end_interaction().expect("nothing to close").is_empty(), "no op of this peer is open");
+
+	// The guest's open interaction is stamped before everything the host does.
+	host.replay_hot_op(guest_op(10, 1, add_network(9))).expect("guest op");
+	host.replay_hot_op(guest_op(11, 2, set_attribute("guest", 1))).expect("guest op");
+
+	host.stage_ops([set_attribute("host", 1)]).expect("host op");
+	let marker = host.end_interaction().expect("close").pop().expect("the host's interaction was open");
+	assert!(matches!(marker.op, RegistryDelta::Meta) && marker.ends_interaction());
+	assert!(host.end_interaction().expect("nothing more to close").is_empty(), "closing twice stages nothing");
+
+	let closed = host.closed_interactions();
+	assert_eq!(closed.len(), 1, "the guest's open interaction is not listed: {closed:?}");
+	assert_eq!(closed[0].author, PeerId(1));
+	assert!(closed[0].contiguous);
+	assert_eq!(closed[0].ops.len(), 3, "RegisterPeer, the attribute write and the marker");
+
+	let revs = host.retire_hot_ops(&closed[0].ops).expect("retire");
+	assert_eq!(revs.len(), 2, "the marker commits no delta");
+	assert_eq!(host.hot_log().len(), 2, "the guest's ops stay hot");
+
+	host.replay_hot_op(closing(guest_op(12, 3, RegistryDelta::Meta))).expect("guest marker");
+	let closed = host.closed_interactions();
+	assert_eq!(closed.len(), 1);
+	assert_eq!(closed[0].author, PeerId(2));
+	assert_eq!(closed[0].ops.len(), 3);
+}
+
+#[test]
+fn an_interaction_with_a_gap_is_not_contiguous_until_the_gap_fills() {
+	let mut host = Session::with_peer(PeerId(1));
+	let guest_op = |counter: u64, sequence: u64, op: RegistryDelta| crate::HotOp {
+		op,
+		timestamp: crate::TimeStamp { counter, peer: PeerId(2) },
+		sequence: crate::HotSequence(sequence),
+		attributes: Default::default(),
+	};
+	host.replay_hot_op(guest_op(10, 1, add_network(9))).expect("guest op");
+	host.replay_hot_op(closing(guest_op(12, 3, RegistryDelta::Meta))).expect("guest marker, op 2 still in flight");
+	host.replay_hot_op(closing(guest_op(13, 4, set_attribute("guest", 2)))).expect("a later interaction");
+
+	let closed = host.closed_interactions();
+	assert_eq!(closed.len(), 2);
+	assert!(!closed[0].contiguous);
+	assert!(!closed[1].contiguous, "the gap is behind the later interaction too");
+
+	host.replay_hot_op(guest_op(11, 2, set_attribute("guest", 1))).expect("the late op");
+	let closed = host.closed_interactions();
+	assert!(closed.iter().all(|interaction| interaction.contiguous));
+	assert_eq!(closed[0].ops.len(), 3);
+}
+
+#[test]
+fn a_flagged_write_closes_its_interaction() {
+	let mut host = Session::with_peer(PeerId(1));
+	let guest_op = |counter: u64, sequence: u64, op: RegistryDelta| crate::HotOp {
+		op,
+		timestamp: crate::TimeStamp { counter, peer: PeerId(2) },
+		sequence: crate::HotSequence(sequence),
+		attributes: Default::default(),
+	};
+	host.replay_hot_op(guest_op(10, 1, add_network(9))).expect("guest op");
+	assert!(host.closed_interactions().is_empty(), "the interaction is open");
+
+	host.replay_hot_op(closing(guest_op(11, 2, set_attribute("guest", 1)))).expect("flagged write");
+	let closed = host.closed_interactions();
+	assert_eq!(closed.len(), 1);
+	assert_eq!(closed[0].ops.len(), 2, "both writes, and no marker of its own");
+}
+
+fn closing(mut hot_op: crate::HotOp) -> crate::HotOp {
+	hot_op.attributes.set(crate::attr::delta::INTERACTION_END, Value::Bool(true), crate::TimeStamp::ORIGIN);
+	hot_op
+}

document/graph-storage/src/tests/crdt.rs

@@ -41,6 +41,7 @@ fn apply_hot_op_advances_clock_past_observed_timestamp() {
 		op: remove_node_op(NodeId(99)),
 		timestamp: observed,
 		sequence: crate::HotSequence(1),
+		attributes: Default::default(),
 	};
 
 	document
@@ -978,6 +979,7 @@ fn hot_op(op: RegistryDelta, counter: u64, peer: u64, sequence: u64) -> HotOp {
 		op,
 		timestamp: ts(counter, peer),
 		sequence,
+		attributes: Default::default(),
 	}
 }
 

node-graph/rfcs/document-format.md

@@ -123,6 +123,7 @@ pub enum RegistryDelta {
     AddResource     { id: ResourceId, entry: ResourceEntry },           // whole-entry; reverse of RemoveResource
     RemoveResource  { id: ResourceId, snapshot: ResourceEntry },        // snapshot for O(1) reverse
     Merge           { extra_parents: Vec<Rev> },                        // joins divergent tips; registry no-op
+    Meta,                                                              
     Other(Value),                                                      // forward-compatible escape hatch
 }
 
@@ -157,6 +158,8 @@ One timestamp per `Delta` applies to every LWW-eligible write inside its `kind`.
 
 `Delta.attributes` is a type-erased annotation bucket (the same shape as the registry's attribute buckets) for mutable, local-only labels: the `interaction_end` marker that bounds undo units, and later commit messages. It is **excluded from `id`** so that annotating a delta never changes its content-addressed identity. An inline write sets it before the delta's history frame is persisted, while a later relabel rewrites that frame.
 
+`HotOp.attributes` is the same bucket on a hot op, but shared: it travels with the op to every peer and is left behind at retirement. The same `interaction_end` key marks the last op of its author's interaction, so the retirer takes the author's ops through it as one unit. When the boundary comes after the last write was sent, a data-less `RegistryDelta::Meta` op carries the flag; retirement drops `Meta` ops. The retirer marks the unit's last retired delta with `Delta.attributes`' own `interaction_end`, so the boundary survives without the hot op.
+
 ## History as a tree
 
 History is a multi-parent DAG. Branching is implicit: every concurrent or out-of-sync edit creates a branch by virtue of sharing a parent with another delta. Divergent tips are rejoined by an explicit `Merge` delta listing the joined tips, which is a registry no-op (it only collapses the tips so `head` stays a single `Rev`).