diff --git a/document/format/tests/open_create.rs b/document/format/tests/open_create.rs index d58732614d5..ae2b8a7ddfb 100644 --- a/document/format/tests/open_create.rs +++ b/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:?}")); diff --git a/document/format/tests/staging_and_reopen.rs b/document/format/tests/staging_and_reopen.rs index ab6eb513991..87dce9bcae6 100644 --- a/document/format/tests/staging_and_reopen.rs +++ b/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(); diff --git a/document/graph-storage/src/attributes/attr.rs b/document/graph-storage/src/attributes/attr.rs index 281d916d672..2c8b80ed9c9 100644 --- a/document/graph-storage/src/attributes/attr.rs +++ b/document/graph-storage/src/attributes/attr.rs @@ -68,6 +68,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"; } diff --git a/document/graph-storage/src/crdt.rs b/document/graph-storage/src/crdt.rs index 4eb23f3c054..1e18426dead 100644 --- a/document/graph-storage/src/crdt.rs +++ b/document/graph-storage/src/crdt.rs @@ -211,6 +211,7 @@ pub enum RegistryDelta { Merge { extra_parents: Vec, }, + Meta, // Allow for future delta types without a model change Other(Value), } diff --git a/document/graph-storage/src/document.rs b/document/graph-storage/src/document.rs index 0e3ca48836d..7da8ced7104 100644 --- a/document/graph-storage/src/document.rs +++ b/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(()) } diff --git a/document/graph-storage/src/prior.rs b/document/graph-storage/src/prior.rs index d66bfad8528..1d7b59e6bee 100644 --- a/document/graph-storage/src/prior.rs +++ b/document/graph-storage/src/prior.rs @@ -184,7 +184,7 @@ pub(crate) fn capture(registry: &Registry, op: &RegistryDelta) -> Vec { key: delta.key.clone(), previous: registry.attributes.get(&delta.key).cloned(), }), - RegistryDelta::Merge { .. } | RegistryDelta::Other(_) => {} + RegistryDelta::Merge { .. } | RegistryDelta::Meta | RegistryDelta::Other(_) => {} } priors } diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index 4cb6c33be72..b387e74c1a7 100644 --- a/document/graph-storage/src/session.rs +++ b/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) -> Result, CrdtError> { + self.stage_ops_with(ops.into_iter().map(|op| (op, Attributes::new()))) + } + + fn stage_ops_with(&mut self, ops: impl IntoIterator) -> Result, CrdtError> { // A batch is staged whole or not at all. - let mut pending: Vec = 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,6 +350,60 @@ 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, 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 { + let mut by_author: HashMap> = 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 { self.document.hot_log.iter().filter(|hot_op| hot_op.timestamp <= up_to).map(HotOp::id).collect() @@ -351,7 +411,7 @@ impl Session { /// 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, CrdtError> { let wanted: HashSet = ids.iter().copied().collect(); let (drained, kept): (Vec, Vec) = self.document.hot_log.drain(..).partition(|hot_op| wanted.contains(&hot_op.id())); @@ -359,7 +419,10 @@ impl Session { 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, + /// 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)] diff --git a/document/graph-storage/src/tests/bookkeeping.rs b/document/graph-storage/src/tests/bookkeeping.rs index f8a86dc7bf7..176223a965c 100644 --- a/document/graph-storage/src/tests/bookkeeping.rs +++ b/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 +} diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index b56aa8b7ad3..4121e3c515f 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/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(), } } diff --git a/node-graph/rfcs/document-format.md b/node-graph/rfcs/document-format.md index fff4b46ec59..a11ad767cbd 100644 --- a/node-graph/rfcs/document-format.md +++ b/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 }, // 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`).