From 04c1dbf9e7efcfebf7fe614d333a853c26deacab Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Fri, 30 Oct 2026 17:11:40 +0000 Subject: [PATCH] Add the session operations for taking back, dropping and restoring interactions and for following a retirer --- document/format/src/export.rs | 1 + document/format/src/manifest.rs | 2 +- document/format/src/persist.rs | 1 + document/format/tests/migrations.rs | 8 +- document/graph-storage/src/crdt.rs | 12 + document/graph-storage/src/document.rs | 29 +- document/graph-storage/src/history.rs | 16 +- .../graph-storage/src/history_metadata.rs | 43 +- document/graph-storage/src/registry.rs | 13 +- document/graph-storage/src/schema/session.rs | 7 +- .../graph-storage/src/schema/session_state.rs | 3 + document/graph-storage/src/session.rs | 316 ++++++++++- .../graph-storage/src/tests/bookkeeping.rs | 286 +++++++++- document/graph-storage/src/tests/crdt.rs | 33 ++ document/graph-storage/src/tests/sessions.rs | 514 +++++++++++++++++- document/migrations/formats/src/lib.rs | 3 +- document/migrations/formats/src/steps/mod.rs | 1 + .../migrations/formats/src/steps/v2_to_v3.rs | 54 ++ document/migrations/formats/src/v3.rs | 44 ++ document/migrations/formats/tests/latest.rs | 1 + 20 files changed, 1354 insertions(+), 33 deletions(-) create mode 100644 document/migrations/formats/src/steps/v2_to_v3.rs create mode 100644 document/migrations/formats/src/v3.rs diff --git a/document/format/src/export.rs b/document/format/src/export.rs index 2f9d7b25356..296d726f6a7 100644 --- a/document/format/src/export.rs +++ b/document/format/src/export.rs @@ -154,6 +154,7 @@ impl Gdd { last_hot_sequence: self.session.last_hot_sequence(), retired_marks: self.session.retired_marks().clone(), clock_counter: self.session.clock_counter(), + shared: false, view_settings: self.view_settings.clone(), network_view_settings: self.network_view_settings.clone(), }; diff --git a/document/format/src/manifest.rs b/document/format/src/manifest.rs index e0e8f13aa92..956828edb38 100644 --- a/document/format/src/manifest.rs +++ b/document/format/src/manifest.rs @@ -9,7 +9,7 @@ use crate::{DEFAULT_HISTORY_CODEC, DEFAULT_HOT_LOG_CODEC, DEFAULT_REGISTRY_CODEC pub const FORMAT_MAGIC: &str = "gdd"; /// Newest format version this build can open, which older documents migrate to. `cargo run migration snapshot` bumps it when the stored types change. -pub const SUPPORTED_FORMAT_VERSION: u32 = 2; +pub const SUPPORTED_FORMAT_VERSION: u32 = 3; /// The on-disk codec for each working-copy payload, recorded so reads/writes never have to probe /// the filesystem to discover it. The manifest itself is excluded: it is always JSON, since it must diff --git a/document/format/src/persist.rs b/document/format/src/persist.rs index 882a76f77ea..e2c2e9b335e 100644 --- a/document/format/src/persist.rs +++ b/document/format/src/persist.rs @@ -292,6 +292,7 @@ impl Gdd { last_hot_sequence: self.session.last_hot_sequence(), retired_marks: self.session.retired_marks().clone(), clock_counter: self.session.clock_counter(), + shared: false, view_settings: self.view_settings.clone(), network_view_settings: self.network_view_settings.clone(), } diff --git a/document/format/tests/migrations.rs b/document/format/tests/migrations.rs index 579ccd83c67..0e65f27c52a 100644 --- a/document/format/tests/migrations.rs +++ b/document/format/tests/migrations.rs @@ -91,10 +91,14 @@ fn opening_steps_the_format_version() { let (working, layout) = gdd.into_storage(); let migrated = GddV1::open_in(working, layout).await.unwrap(); - assert_eq!(migrated.manifest().format_version, 2); + assert_eq!(migrated.manifest().format_version, manifest::SUPPORTED_FORMAT_VERSION); let (working, layout) = migrated.into_storage(); - assert_eq!(GddV1::open_in(working, layout).await.unwrap().manifest().format_version, 2, "the bump persisted"); + assert_eq!( + GddV1::open_in(working, layout).await.unwrap().manifest().format_version, + manifest::SUPPORTED_FORMAT_VERSION, + "the bump persisted" + ); }); } diff --git a/document/graph-storage/src/crdt.rs b/document/graph-storage/src/crdt.rs index e9e3b942b28..54bf6a0f07b 100644 --- a/document/graph-storage/src/crdt.rs +++ b/document/graph-storage/src/crdt.rs @@ -1,4 +1,5 @@ use crate::{Attributes, Delta, PeerId, Prior, RegistryDelta, Rev, TimeStamp, Value, attr, compute_rev}; +use graphene_resource::ResourceHash; impl Delta { pub fn new(parent: Option, author: PeerId, timestamp: TimeStamp, kind: RegistryDelta, reverse: Vec) -> Self { @@ -79,6 +80,17 @@ impl Delta { } } +impl RegistryDelta { + pub fn resource_hash(&self) -> Option { + match self { + RegistryDelta::AddResource { entry, .. } => entry.hash, + RegistryDelta::RemoveResource { snapshot, .. } => snapshot.hash, + RegistryDelta::SetResourceHash { hash, .. } => *hash, + _ => None, + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/document/graph-storage/src/document.rs b/document/graph-storage/src/document.rs index 59ca3ee5ce7..06803d259bf 100644 --- a/document/graph-storage/src/document.rs +++ b/document/graph-storage/src/document.rs @@ -2,8 +2,8 @@ use std::collections::{HashMap, HashSet}; use std::hash::Hash; use crate::{ - Attributes, CrdtError, Delta, ExportSlot, History, HotOp, HotSequence, Implementation, InputSlot, LamportClock, MAX_EXPORT_SLOTS, MAX_INPUT_SLOTS, Network, NetworkId, Node, NodeId, NodeInput, - PeerId, PeerRegistration, Prior, Registry, RegistryDelta, ResourceEntry, ResourceId, RetiredMarks, Rev, SourceValue, TimeStamp, Tombstone, Touched, UserId, + Attributes, CrdtError, Delta, ExportSlot, History, HotOp, HotOpId, HotSequence, Implementation, InputSlot, LamportClock, MAX_EXPORT_SLOTS, MAX_INPUT_SLOTS, Network, NetworkId, Node, NodeId, + NodeInput, PeerId, PeerRegistration, Prior, Registry, RegistryDelta, ResourceEntry, ResourceId, RetiredMarks, Rev, SourceValue, TimeStamp, Tombstone, Touched, UserId, }; #[derive(Clone, Debug)] @@ -16,7 +16,7 @@ pub struct Document { pub(crate) hot_log: Vec, /// The stamps of the hot log's ops, so a re-announced op is recognised without a scan. pub(crate) hot_timestamps: HashSet, - /// Which hot ops are retired into history, so a late copy is dropped. + /// Which hot ops are retired or taken back, so a late copy is dropped. 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 @@ -112,12 +112,35 @@ impl Document { Ok(()) } + /// Mark hot ops retired, dropping any the hot log holds. Returns what the dropped ops named. + pub(crate) fn mark_retired(&mut self, retired: impl IntoIterator) -> Touched { + self.retired.extend(retired); + self.drop_retired_hot_ops() + } + /// Take on a peer's marks as well as this peer's. Returns what the dropped ops named. pub(crate) fn absorb_retired_marks(&mut self, remote: &RetiredMarks) -> Touched { self.retired.absorb(remote); self.drop_retired_hot_ops() } + /// Take back hot ops for good, re-deriving the working registry without them. Returns what they named and the ops. + pub(crate) fn retract_hot_ops(&mut self, ids: &[HotOpId]) -> (Touched, Vec) { + let wanted: HashSet = ids.iter().copied().collect(); + let taken: Vec = self.hot_log.extract_if(.., |hot_op| wanted.contains(&hot_op.id())).collect(); + self.retired.take_back(ids); + + let mut touched = Touched::default(); + for hot_op in &taken { + touched.record(&hot_op.op); + self.hot_timestamps.remove(&hot_op.timestamp); + } + if !taken.is_empty() { + self.rebuild_working(); + } + (touched, taken) + } + /// Bring the stamp index back in line after the hot log was filtered as a whole. pub(crate) fn resync_hot_timestamps(&mut self) { self.hot_timestamps = self.hot_log.iter().map(|hot_op| hot_op.timestamp).collect(); diff --git a/document/graph-storage/src/history.rs b/document/graph-storage/src/history.rs index 3d7d7b00480..1e17465789d 100644 --- a/document/graph-storage/src/history.rs +++ b/document/graph-storage/src/history.rs @@ -74,14 +74,7 @@ impl History { /// (idempotent re-apply) overwrites the existing entry in place rather than appending, so the /// order and index are unchanged. pub fn push(&mut self, delta: Delta) { - match &delta.kind { - RegistryDelta::AddResource { entry, .. } => self.resource_hashes.extend(entry.hash), - RegistryDelta::RemoveResource { snapshot, .. } => self.resource_hashes.extend(snapshot.hash), - RegistryDelta::SetResourceHash { hash: Some(hash), .. } => { - self.resource_hashes.insert(*hash); - } - _ => {} - } + self.resource_hashes.extend(delta.kind.resource_hash()); if let Some(&position) = self.index.get(&delta.id) { self.deltas[position] = delta; return; @@ -212,6 +205,13 @@ impl History { } } + /// Record when a retired delta entered history (outside its `Rev`). Returns whether the delta was found. + pub(crate) fn set_retired_at(&mut self, rev: Rev, wall_ms: u64) -> bool { + let Some(&position) = self.index.get(&rev) else { return false }; + self.deltas[position].retired_at_ms = wall_ms; + true + } + /// Set a local annotation attribute (e.g. a commit message) on a retired delta in place. Excluded /// from the delta's `Rev`, so identity and the index are unchanged. Returns whether the delta was found. pub fn annotate(&mut self, rev: Rev, key: &str, value: Value, timestamp: TimeStamp) -> bool { diff --git a/document/graph-storage/src/history_metadata.rs b/document/graph-storage/src/history_metadata.rs index 4974b84a6f7..ab5dc54b413 100644 --- a/document/graph-storage/src/history_metadata.rs +++ b/document/graph-storage/src/history_metadata.rs @@ -43,12 +43,25 @@ pub mod rev_attr { pub const TAG_PREFIX: &str = "tag:"; } -/// The standing fact for each subject and key. Folding the same facts in any order gives the same record. -#[derive(Clone, Debug, Default, PartialEq)] +/// The standing fact for each subject and key. Folding the same facts in any order gives the same record, so it travels as its facts. +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +#[serde(from = "Vec", into = "Vec")] pub struct HistoryMetadata { facts: BTreeMap>, } +impl From> for HistoryMetadata { + fn from(facts: Vec) -> Self { + Self::fold(facts) + } +} + +impl From for Vec { + fn from(record: HistoryMetadata) -> Self { + record.facts().cloned().collect() + } +} + impl HistoryMetadata { pub fn is_empty(&self) -> bool { self.facts.is_empty() @@ -107,13 +120,20 @@ impl HistoryMetadata { Some(MetadataFact { subject, key, value, stamp }) } - /// Take on a fact unless the standing one has a later stamp, or the same stamp and a value that encodes no lower. - pub fn absorb(&mut self, fact: MetadataFact) { + /// Take on a fact unless the standing one has a later stamp, or the same stamp and a value that encodes no lower. Returns whether it landed. + pub fn absorb(&mut self, fact: MetadataFact) -> bool { let attributes = self.facts.entry(fact.subject).or_default(); let stands = |mine: &MetadataFact| mine.stamp.cmp(&fact.stamp).then_with(|| mine.value.to_string().cmp(&fact.value.to_string())).is_ge(); - if !attributes.get(&fact.key).is_some_and(stands) { - attributes.insert(fact.key.clone(), fact); + if attributes.get(&fact.key).is_some_and(stands) { + return false; } + attributes.insert(fact.key.clone(), fact); + true + } + + /// Take on another copy's record. Returns the facts that landed. + pub fn merge(&mut self, other: &Self) -> Vec { + other.facts().filter(|fact| self.absorb((*fact).clone())).cloned().collect() } } @@ -145,6 +165,17 @@ mod tests { assert_eq!(HistoryMetadata::fold([name("Ada", 10), name("Bea", 10)]), HistoryMetadata::fold([name("Bea", 10), name("Ada", 10)])); } + #[test] + fn merging_copies_agrees_either_way_and_only_once() { + let a = HistoryMetadata::fold([fact(ADA, user_attr::NAME, "Ada", 10)]); + let b = HistoryMetadata::fold([fact(ADA, user_attr::NAME, "Bea", 20)]); + let (mut ab, mut ba) = (a.clone(), b.clone()); + assert_eq!(ab.merge(&b).len(), 1); + assert!(ba.merge(&a).is_empty()); + assert_eq!(ab, ba); + assert!(ab.merge(&b).is_empty(), "merging again changes nothing"); + } + #[test] fn a_clock_behind_a_peers_still_gets_its_say() { let record = HistoryMetadata::fold([fact(ADA, user_attr::NAME, "Ada", 100)]); diff --git a/document/graph-storage/src/registry.rs b/document/graph-storage/src/registry.rs index 046a7e8639f..d3edd8cca5b 100644 --- a/document/graph-storage/src/registry.rs +++ b/document/graph-storage/src/registry.rs @@ -1,4 +1,4 @@ -use crate::{Attributes, NetworkId, NodeId, PeerId, Registry, ResourceId, ResourceStore, SourceKey, TimeStamp}; +use crate::{Attributes, NetworkId, NodeId, PeerId, Registry, ResourceId, ResourceStore, SourceKey, TimeStamp, Tombstone}; use std::collections::HashMap; impl Registry { @@ -25,6 +25,17 @@ impl Registry { } } + /// Whether both registries removed the same entities at the same times. Registries equal in value but not in their + /// removal marks resolve later ops differently. + pub fn removal_marks_equal(&self, other: &Self) -> bool { + fn same(a: &HashMap>, b: &HashMap>) -> bool { + a.len() == b.len() + && a.iter() + .all(|(id, mark)| b.get(id).is_some_and(|other| other.timestamp == mark.timestamp && other.placeholder == mark.placeholder)) + } + same(&self.removed_nodes, &other.removed_nodes) && same(&self.removed_networks, &other.removed_networks) && same(&self.removed_resources, &other.removed_resources) + } + /// True if both registries agree on every value-bearing field, ignoring per-slot and /// per-attribute timestamps. Mirrors `compute_deltas`'s value-only semantics, so unchanged /// state at a stamped slot doesn't count as drift. `peer_users` is excluded: it isn't diffed by diff --git a/document/graph-storage/src/schema/session.rs b/document/graph-storage/src/schema/session.rs index 153f5e5ab31..5736b7f57b7 100644 --- a/document/graph-storage/src/schema/session.rs +++ b/document/graph-storage/src/schema/session.rs @@ -28,11 +28,14 @@ pub struct HotOpId { pub sequence: HotSequence, } -/// Which hot ops are retired: each author's gap-free prefix, plus runs past it while ops are in flight. Persisted, so -/// a late copy of a retired op is dropped. +/// Which hot ops are retired or taken back, as each author's gap-free prefix plus runs past it while ops are in flight. +/// Persisted, so a late copy of such an op is dropped. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RetiredMarks { pub retired_up_to: HashMap, /// retired ops past their author's prefix, as inclusive runs, one per gap in flight. pub retired_runs: HashMap>, + /// Which covered ops their author took back rather than retired, as inclusive runs. + #[serde(default)] + pub taken_back: HashMap>, } diff --git a/document/graph-storage/src/schema/session_state.rs b/document/graph-storage/src/schema/session_state.rs index 7d8be61f942..dc2c951065b 100644 --- a/document/graph-storage/src/schema/session_state.rs +++ b/document/graph-storage/src/schema/session_state.rs @@ -51,4 +51,7 @@ pub struct SessionState { /// The Lamport counter, so a reopen never mints a spent stamp. #[serde(default)] pub clock_counter: u64, + /// Whether the document belongs in its room, so a reopen reconnects. + #[serde(default)] + pub shared: bool, } diff --git a/document/graph-storage/src/session.rs b/document/graph-storage/src/session.rs index 076a88cae15..11cc40a149d 100644 --- a/document/graph-storage/src/session.rs +++ b/document/graph-storage/src/session.rs @@ -19,6 +19,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, + /// What the ops this peer took back named. A retirement racing the take-back may still name them, and this peer may + /// hold the only copy, so they stay garbage collection roots for the life of the session. + taken_back_resources: HashSet, } impl Session { @@ -40,6 +43,7 @@ impl Session { Self { document: Document::empty(peer, user), remote_tips: HashMap::new(), + taken_back_resources: HashSet::new(), } } @@ -56,6 +60,11 @@ impl Session { (peer == self.document.peer).then_some(self.document.user).or_else(registered) } + /// Whether `peer` is this device or another registered to the same user. + pub fn is_mine(&self, peer: PeerId) -> bool { + self.user_of(peer) == Some(self.document.user) + } + pub fn registry(&self) -> &Registry { &self.document.working_registry } @@ -284,6 +293,7 @@ impl Session { ..Document::empty(peer, user) }, remote_tips: HashMap::new(), + taken_back_resources: HashSet::new(), } } @@ -305,6 +315,116 @@ impl Session { Ok(session) } + /// Drop hot ops another peer retired, without retiring them here. Until their deltas arrive, the working registry + /// shows the snapshot plus what is still hot. + pub fn discard_hot_ops(&mut self, retired: &[HotOpId]) -> Touched { + self.document.mark_retired(retired.iter().copied()) + } + + /// Take back this peer's latest interaction (its ops since the previous marker) while it is still hot, so it leaves + /// no step in history. `None` when nothing is hot or the interaction already retired. + pub fn retract_interaction(&mut self) -> Result, CrdtError> { + let peer = self.document.peer; + let mut own: Vec<&HotOp> = self.document.hot_log.iter().filter(|hot_op| hot_op.timestamp.peer == peer).collect(); + own.sort_by_key(|hot_op| hot_op.sequence); + let closed_before = match own.last() { + Some(last) if last.ends_interaction() => own.len() - 1, + Some(_) => own.len(), + None => return Ok(None), + }; + let start = own[..closed_before].iter().rposition(|hot_op| hot_op.ends_interaction()).map_or(0, |position| position + 1); + if own[start..].iter().all(|hot_op| matches!(hot_op.op, RegistryDelta::Meta)) { + return Ok(None); + } + let ids: Vec = own[start..].iter().map(|hot_op| hot_op.id()).collect(); + let retired = &self.document.retired; + // A later interaction that retired rather than was taken back is the latest, and undoes through history. + if ids.iter().any(|&id| retired.covers(id)) || !retired.took_back_all(peer, own[own.len() - 1].sequence, self.document.last_hot_sequence) { + return Ok(None); + } + let (touched, taken) = self.document.retract_hot_ops(&ids); + self.taken_back_resources.extend(taken.iter().filter_map(|hot_op| hot_op.op.resource_hash())); + let ops = taken.into_iter().map(|hot_op| hot_op.op).filter(|op| !matches!(op, RegistryDelta::Meta)).collect(); + Ok(Some(Retraction { ids, touched, ops })) + } + + /// Take back hot ops another peer retracted. See [`retract_interaction`](Self::retract_interaction). + pub fn retract_hot_ops(&mut self, ids: &[HotOpId]) -> Touched { + self.document.retract_hot_ops(ids).0 + } + + /// The last delta of this person's latest retired interaction on the head's line, authored on any of their peers. + pub fn latest_own_interaction(&self) -> Option { + let mut current = self.document.head?; + loop { + let delta = self.document.history.get(current)?; + if delta.is_interaction_end() && self.is_mine(delta.author) { + return Some(current); + } + current = delta.parent?; + } + } + + /// Drop a retired interaction out of the shared line. The head moves to its parent and the later steps are minted + /// again there with the same author, stamp and kind, so every follower derives identical revs. The originals stay + /// as a branch. Returns the move to broadcast and what the dropped ops named. + pub fn drop_interaction(&mut self, undone: Rev) -> Result<(HeadMove, Touched), CrdtError> { + let from = self.document.head; + let base = self.interaction_start_parent(undone, false).ok_or(CrdtError::NotUndoable(undone))?; + + let mut newest_first = Vec::new(); + let mut current = from; + while current != Some(base) { + let rev = current.ok_or(CrdtError::NotFoundInHistory(undone))?; + let delta = self.document.history.get(rev).ok_or(CrdtError::NotFoundInHistory(rev))?.clone(); + if matches!(delta.kind, RegistryDelta::Merge { .. }) { + return Err(CrdtError::NotUndoable(undone)); + } + current = delta.parent; + newest_first.push(delta); + } + let position = newest_first.iter().position(|delta| delta.id == undone).ok_or(CrdtError::NotFoundInHistory(undone))?; + + // Minted on a scratch fold first, so a failing copy changes nothing. + let mut scratch = self.fold_scratch_at(Some(base))?; + let mut copies = Vec::with_capacity(position); + let mut parent = Some(base); + for original in newest_first[..position].iter().rev() { + let reverse = crate::prior::capture(&scratch.working_registry, &original.kind); + scratch.apply_op_with(RegistryTarget::Working, original.kind.clone(), original.timestamp, ApplyMode::Idempotent)?; + let copy = Delta::new(parent, original.timestamp.peer, original.timestamp, original.kind.clone(), reverse); + parent = Some(copy.id); + copies.push(copy); + } + + let mut touched = Touched::default(); + newest_first[position..].iter().for_each(|delta| touched.record(&delta.kind)); + for (original, copy) in newest_first[..position].iter().rev().zip(&mut copies) { + self.document.history.push(copy.clone()); + self.carry_attributes(original, copy.id); + if let Some(carried) = self.document.history.get(copy.id) { + *copy = carried.clone(); + } + } + self.document.head = parent; + self.document.retired_snapshot = scratch.working_registry; + self.document.history.canonical_sort(); + + self.document.redo_stack.clear(); + self.document.rebuild_working(); + if let Some(head) = self.document.head { + self.publish_up_to(head); + } + Ok(( + HeadMove { + from, + head: self.document.head, + copies, + }, + touched, + )) + } + /// Move the shared head back to its ancestor `target`, leaving the line since as a branch. Returns the move to /// broadcast and what the steps left behind named. Only the retirer does this; a guest asks it to. pub fn move_head_to(&mut self, target: Rev) -> Result<(HeadMove, Touched), CrdtError> { @@ -390,6 +510,31 @@ impl Session { Ok(touched) } + /// Mint a dropped interaction's deltas again on top of the line as one interaction. They keep their + /// original stamps, so a field someone wrote since keeps the newer value. Returns the new revs. + pub fn restore_interaction(&mut self, dropped: Rev) -> Result, CrdtError> { + let base = self.interaction_start_parent(dropped, false).ok_or(CrdtError::NotUndoable(dropped))?; + let mut originals = Vec::new(); + let mut current = Some(dropped); + while current != Some(base) { + let rev = current.ok_or(CrdtError::NothingToRedo)?; + let delta = self.document.history.get(rev).ok_or(CrdtError::NotFoundInHistory(rev))?.clone(); + current = delta.parent; + originals.push(delta); + } + originals.reverse(); + let revs = self.commit_ops_authored_at(originals.iter().map(|delta| (delta.kind.clone(), Some(delta.timestamp))), true)?; + for (original, &rev) in originals.iter().zip(&revs) { + self.carry_attributes(original, rev); + } + if let Some(&last) = revs.last() { + self.mark_interaction_end(last); + } + self.document.history.canonical_sort(); + self.document.rebuild_working(); + Ok(revs) + } + /// Which hot ops are done with, for a peer catching up and for persisting. See [`RetiredMarks`]. pub fn retired_marks(&self) -> &RetiredMarks { &self.document.retired @@ -412,12 +557,17 @@ impl Session { } fn fold_at(&self, head: Option) -> Result { + Ok(self.fold_scratch_at(head)?.working_registry) + } + + /// A scratch document whose working registry is the fold of `head`'s ancestry. + fn fold_scratch_at(&self, head: Option) -> Result { let reachable = self.document.history.ancestors(head); let mut scratch = Document::empty(self.document.peer, self.document.user); for delta in self.document.history.iter().filter(|delta| reachable.contains(&delta.id)) { scratch.apply_op_with(RegistryTarget::Working, delta.kind.clone(), delta.timestamp, ApplyMode::Idempotent)?; } - Ok(scratch.working_registry) + Ok(scratch) } pub fn history_len(&self) -> usize { @@ -634,6 +784,16 @@ impl Session { Ok(true) } + /// Retire one closed interaction, coarsened, marking its last delta as the interaction's end. Every op is retired, + /// the ones coarsening drops included. Returns the new revs. + pub fn retire_interaction(&mut self, interaction: &ClosedInteraction) -> Result, CrdtError> { + let revs = self.retire_hot_ops_with(&interaction.ops, true)?; + if let Some(&last) = revs.last() { + self.mark_interaction_end(last); + } + Ok(revs) + } + /// 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() @@ -643,12 +803,17 @@ impl Session { /// /// One retired delta per hot op, except a `Meta` op, which commits nothing. pub fn retire_hot_ops(&mut self, ids: &[HotOpId]) -> Result, CrdtError> { + self.retire_hot_ops_with(ids, false) + } + + fn retire_hot_ops_with(&mut self, ids: &[HotOpId], coarsened: bool) -> 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())); self.document.hot_log = kept; self.document.resync_hot_timestamps(); self.document.retired.extend(drained.iter().map(HotOp::id)); + let drained = if coarsened { coarsen(drained) } else { drained }; let ops = drained .into_iter() .filter(|hot_op| !matches!(hot_op.op, RegistryDelta::Meta)) @@ -656,7 +821,7 @@ impl Session { self.commit_ops_authored_at(ops, true) } - /// Promote every hot op stamped at or before `up_to`, whatever its author. + /// Promote every hot op stamped at or before `up_to`, whatever its author and whether or not its interaction is closed. pub fn retire(&mut self, up_to: TimeStamp) -> Result, CrdtError> { let ids = self.hot_ops_up_to(up_to); self.retire_hot_ops(&ids) @@ -668,6 +833,16 @@ impl Session { self.document.history.mark_interaction_end(rev, timestamp); } + /// Give a minted copy the attributes and retirement time of the delta it stands for. + fn carry_attributes(&mut self, original: &Delta, copy: Rev) { + if let Some(wall_ms) = original.retired_at() { + self.document.history.set_retired_at(copy, wall_ms); + } + for (key, value) in original.attributes.live() { + self.document.history.annotate(copy, key, value.value.clone(), value.timestamp); + } + } + /// Low-level: set a local annotation attribute (e.g. a commit message) on a retired delta in place. /// Excluded from the delta's content-addressed `Rev`, so identity is unchanged. Returns whether the /// delta was found. The `Gdd` layer re-persists the affected history frame after calling this. @@ -810,7 +985,7 @@ impl Session { self.document.history.verify() } - /// Every resource hash referenced by the current registry *or* anywhere in history. Undo removes a + /// Every resource hash referenced by the current registry, the hot log, or anywhere in history. Undo removes a /// interaction's `AddResource` from the working registry, so a redoable (or re-undoable) interaction's /// resources no longer appear in `registry().resources` even though redo still needs them. Resource GC /// must keep this whole set alive, not just the current head's, or undo then redo loses declaration @@ -818,9 +993,17 @@ impl Session { pub fn all_referenced_resource_hashes(&self) -> HashSet { let mut hashes: HashSet = self.document.working_registry.resources.values().filter_map(|entry| entry.hash).collect(); hashes.extend(self.document.history.resource_hashes()); + hashes.extend(self.document.hot_log.iter().filter_map(|hot_op| hot_op.op.resource_hash())); hashes } + /// Every referenced resource hash plus what the ops this peer took back named. Garbage collection must keep these. + pub fn resource_gc_roots(&self) -> HashSet { + let mut roots = self.all_referenced_resource_hashes(); + roots.extend(&self.taken_back_resources); + roots + } + /// Every proto-node declaration resource referenced by the current registry or anywhere in history, /// with its content hash, or `None` where the hash never resolved. pub fn all_declaration_resources(&self) -> HashMap> { @@ -922,6 +1105,16 @@ impl Session { pub fn restore_hot_sequence(&mut self, sequence: HotSequence) { self.document.last_hot_sequence = self.document.last_hot_sequence.max(sequence); } + + /// What the ops this peer took back named. See [`resource_gc_roots`](Self::resource_gc_roots). + pub fn taken_back_resources(&self) -> &HashSet { + &self.taken_back_resources + } + + /// Restore [`taken_back_resources`](Self::taken_back_resources) into a session that replaced this one. + pub fn restore_taken_back_resources(&mut self, hashes: HashSet) { + self.taken_back_resources.extend(hashes); + } } /// Errors from `Session::commit_from_runtime`. @@ -989,7 +1182,11 @@ fn coalesce(runs: &mut Vec<(HotSequence, HotSequence)>) { } impl RetiredMarks { - /// Whether history already holds this hot op. + pub fn is_empty(&self) -> bool { + self.retired_up_to.is_empty() && self.retired_runs.is_empty() + } + + /// Whether this hot op is retired or taken back. pub fn covers(&self, id: HotOpId) -> bool { self.retired_up_to.get(&id.peer).is_some_and(|&through| id.sequence <= through) || self @@ -1007,9 +1204,28 @@ impl RetiredMarks { for (&peer, runs) in &remote.retired_runs { self.retired_runs.entry(peer).or_default().extend(runs.iter().copied()); } + for (&peer, runs) in &remote.taken_back { + let taken_back = self.taken_back.entry(peer).or_default(); + taken_back.extend(runs.iter().copied()); + coalesce(taken_back); + } self.compact(); } + pub(crate) fn take_back(&mut self, ids: &[HotOpId]) { + for id in ids { + let runs = self.taken_back.entry(id.peer).or_default(); + runs.push((id.sequence, id.sequence)); + coalesce(runs); + } + self.extend(ids.iter().copied()); + } + + /// Whether `peer` took back every op it authored after `after`, through `through`. + pub(crate) fn took_back_all(&self, peer: PeerId, after: HotSequence, through: HotSequence) -> bool { + after >= through || self.taken_back.get(&peer).is_some_and(|runs| runs.iter().any(|&(start, end)| start <= after.next() && through <= end)) + } + /// Record newly retired ops. pub fn extend(&mut self, retired: impl IntoIterator) { for id in retired { @@ -1058,6 +1274,8 @@ pub enum CrdtError { NothingToUndo, #[error("Nothing to redo")] NothingToRedo, + #[error("Interaction {0} cannot be dropped: it is the first or spans a merge")] + NotUndoable(Rev), #[error("Cursor is at {found:?}, the move expected {expected:?}")] CursorMismatch { expected: Option, found: Option }, #[error("Delta {0} is not on the head's line")] @@ -1072,7 +1290,6 @@ pub enum CrdtError { RevMismatch { stored: Rev, expected: Rev }, } -/// A shared head move, as [`Session::move_head_to`] produces it and [`Session::apply_head_move`] follows it. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct HeadMove { pub from: Option, @@ -1098,3 +1315,92 @@ impl MergeOutcome { } } } + +#[derive(Clone, PartialEq, Eq, Hash)] +enum FieldKey { + NodeInput(NodeId, u32), + NodeInputAttribute(NodeId, u32, String), + NodeAttribute(NodeId, String), + NodeImplementation(NodeId), + NodeInputs(NodeId), + NetworkExport(NetworkId, u32), + NetworkAttribute(NetworkId, String), + ResourceHash(ResourceId), + Source(ResourceId, crate::SourceKey), + DocumentAttribute(String), +} + +/// The field a write lands on, `None` for an op that is not a plain field write. +fn field_of(op: &RegistryDelta) -> Option { + Some(match op { + RegistryDelta::ChangeNodeInput { id, index, .. } => FieldKey::NodeInput(*id, *index), + RegistryDelta::ChangeNodeInputAttribute { id, index, delta } => FieldKey::NodeInputAttribute(*id, *index, delta.key.clone()), + RegistryDelta::ChangeNodeAttribute { id, delta } => FieldKey::NodeAttribute(*id, delta.key.clone()), + RegistryDelta::SetNodeImplementation { id, .. } => FieldKey::NodeImplementation(*id), + RegistryDelta::SetNodeInputs { id, .. } => FieldKey::NodeInputs(*id), + RegistryDelta::SetNetworkExport { id, index, .. } => FieldKey::NetworkExport(*id, *index), + RegistryDelta::ChangeNetworkAttribute { id, delta } => FieldKey::NetworkAttribute(*id, delta.key.clone()), + RegistryDelta::SetResourceHash { id, .. } => FieldKey::ResourceHash(*id), + RegistryDelta::AddSource { id, key, .. } | RegistryDelta::RemoveSource { id, key } => FieldKey::Source(*id, *key), + RegistryDelta::ChangeDocumentAttribute { delta } => FieldKey::DocumentAttribute(delta.key.clone()), + _ => return None, + }) +} + +/// The nodes or networks a write refers to beyond its own field. A reference is evidence the target exists at the +/// write's stamp, so a write that carries one only drops when the write superseding it carries the same. +fn references_of(op: &RegistryDelta) -> Vec { + let of_input = |input: &crate::NodeInput| match input { + crate::NodeInput::Node { id, .. } => Some(id.0), + _ => None, + }; + match op { + RegistryDelta::ChangeNodeInput { new_input, .. } => of_input(new_input).into_iter().collect(), + RegistryDelta::SetNodeInputs { inputs, .. } => inputs.iter().filter_map(|slot| of_input(&slot.input)).collect(), + RegistryDelta::SetNetworkExport { export, .. } => export.as_ref().and_then(of_input).into_iter().collect(), + RegistryDelta::SetNodeImplementation { + implementation: Implementation::Network(network), + .. + } => vec![network.0], + _ => Vec::new(), + } +} + +/// Coarsens one interaction's ops. A field write that a later write in it supersedes has no effect on the fold and +/// is dropped, unless it refers to something the later write does not. A whole-list input write supersedes every +/// earlier input write on its node. Structural ops stay. +fn coarsen(mut ops: Vec) -> Vec { + // The author's order. The hot log holds arrival order, which a re-announcement after a lapsed link scrambles. + ops.sort_by_key(|hot_op| hot_op.timestamp); + let mut keep = vec![true; ops.len()]; + let mut latest: HashMap = HashMap::new(); + for (index, hot_op) in ops.iter().enumerate() { + let Some(field) = field_of(&hot_op.op) else { continue }; + let references = references_of(&hot_op.op); + let supersedes = |earlier: &HotOp| references_of(&earlier.op).iter().all(|reference| references.contains(reference)); + + if let FieldKey::NodeInputs(id) = field { + latest.retain(|key, &mut earlier| { + let same_node = matches!(key, FieldKey::NodeInput(node, _) | FieldKey::NodeInputAttribute(node, _, _) | FieldKey::NodeInputs(node) if *node == id); + let superseded = same_node && supersedes(&ops[earlier]); + keep[earlier] &= !superseded; + !superseded + }); + } else if let Some(&earlier) = latest.get(&field) + && supersedes(&ops[earlier]) + { + keep[earlier] = false; + } + latest.insert(field, index); + } + ops.into_iter().zip(keep).filter_map(|(hot_op, keep)| keep.then_some(hot_op)).collect() +} + +#[derive(Clone, Debug)] +pub struct Retraction { + /// The ids every peer drops. + pub ids: Vec, + pub touched: Touched, + /// The ops, marker aside, for a redo to stage afresh. + pub ops: Vec, +} diff --git a/document/graph-storage/src/tests/bookkeeping.rs b/document/graph-storage/src/tests/bookkeeping.rs index 661f96f8e80..f29cc2cbf25 100644 --- a/document/graph-storage/src/tests/bookkeeping.rs +++ b/document/graph-storage/src/tests/bookkeeping.rs @@ -1,4 +1,4 @@ -use crate::{AttributeDelta, CrdtError, HotOpId, Implementation, Network, NetworkId, Node, NodeId, NodeInput, PeerId, RegistryDelta, Session, UserId, Value}; +use crate::{AttributeDelta, CrdtError, HotOpId, Implementation, Network, NetworkId, Node, NodeId, NodeInput, PeerId, RegistryDelta, ResourceId, Session, UserId, Value}; pub(super) fn set_attribute(key: &str, value: u32) -> RegistryDelta { RegistryDelta::ChangeDocumentAttribute { @@ -43,6 +43,232 @@ fn retirement_takes_the_ops_under_the_cutoff_in_any_applied_order() { assert!(host.registry().value_equal(host.retired_registry())); } +/// Only closed interactions retire. An author's ops past its last marker stay hot however old, while a +/// later interaction from someone else retires around them. +#[test] +fn only_closed_interactions_retire_and_an_open_one_stays_hot() { + 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 opens an interaction and leaves it open, stamped earlier than 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_interaction(&closed[0]).expect("retire"); + assert_eq!(revs.len(), 2, "the marker commits no delta"); + assert!(host.history().any(|delta| delta.id == revs[1] && delta.is_interaction_end()), "the interaction retires as one"); + assert_eq!(host.hot_log().len(), 2, "the guest's ops stay hot"); + assert!(host.hot_log().iter().all(|hot_op| hot_op.timestamp.peer == PeerId(2))); + assert!(!host.retired_registry().networks.contains_key(&NetworkId(9))); + assert!(host.registry().networks.contains_key(&NetworkId(9))); + + // The guest closes. Its interaction retires after the host's although it is stamped before it. + 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)); + host.retire_interaction(&closed[0]).expect("retire"); + assert!(host.hot_log().is_empty()); + assert!(host.registry().value_equal(host.retired_registry())); +} + +/// Retracting a hot interaction removes its ops for good and re-derives what they touched from the snapshot +/// and the other hot ops, so a concurrent write to the same entity survives. The ops never retire, and a late copy is dropped. +#[test] +fn a_retracted_interaction_leaves_no_trace_and_keeps_what_others_wrote() { + let mut host = Session::with_peer(PeerId(1)); + host.commit_op_for_test(add_network(3)).expect("base"); + host.document.working_registry = host.document.retired_snapshot.clone(); + + // A guest's write to the network and the host's own gesture on the same network, interleaved. + let guest_write = crate::HotOp { + op: RegistryDelta::ChangeNetworkAttribute { + id: NetworkId(3), + delta: AttributeDelta { + key: "guest".into(), + value: Some(Value::from(serde_json::json!(1))), + }, + }, + timestamp: crate::TimeStamp { counter: 40, peer: PeerId(2) }, + sequence: crate::HotSequence(1), + attributes: Default::default(), + }; + host.stage_ops([RegistryDelta::ChangeNetworkAttribute { + id: NetworkId(3), + delta: AttributeDelta { + key: "host".into(), + value: Some(Value::from(serde_json::json!(1))), + }, + }]) + .expect("host op"); + host.replay_hot_op(guest_write.clone()).expect("guest op"); + host.stage_ops([add_network(9)]).expect("host op"); + let own: Vec = host.hot_log().iter().filter(|hot_op| hot_op.timestamp.peer == PeerId(1)).map(crate::HotOp::id).collect(); + + let crate::Retraction { ids, touched, ops } = host.retract_interaction().expect("retract").expect("the host's interaction is hot"); + assert_eq!(ops.len(), 3, "RegisterPeer and the two writes come back for a redo, the marker does not"); + assert_eq!(ids, own, "the whole open interaction is taken back"); + assert!(touched.networks.contains(&NetworkId(3)) && touched.networks.contains(&NetworkId(9))); + assert_eq!(host.hot_log().len(), 1, "only the guest's op stays"); + assert!(!host.registry().networks.contains_key(&NetworkId(9))); + let network = &host.registry().networks[&NetworkId(3)]; + assert!(!network.attributes.contains_key("host"), "the host's write is gone"); + assert!(network.attributes.contains_key("guest"), "the guest's concurrent write survives"); + assert!(host.retract_interaction().expect("nothing left").is_none()); + + // A late copy of a retracted op is dropped, and nothing of it ever retires. + let late = crate::HotOp { + op: add_network(9), + timestamp: crate::TimeStamp { counter: 2, peer: PeerId(1) }, + sequence: own[2].sequence, + attributes: Default::default(), + }; + host.replay_hot_op(late.clone()).expect("dropped, not an error"); + assert_eq!(host.hot_log().len(), 1); + assert!(host.closed_interactions().is_empty()); + + // Another peer learns of the retraction from the marks alone. + let mut guest = Session::with_peer(PeerId(2)); + guest.commit_op_for_test(add_network(3)).expect("base"); + guest.document.working_registry = guest.document.retired_snapshot.clone(); + guest.replay_hot_op(late).expect("the host's op reached the guest"); + assert!(guest.registry().networks.contains_key(&NetworkId(9))); + let touched = guest.absorb_retired_marks(host.retired_marks()); + assert!(touched.networks.contains(&NetworkId(9))); + assert!(!guest.registry().networks.contains_key(&NetworkId(9))); + assert!(guest.hot_log().is_empty()); +} + +/// An author's ops can reach the retirer out of order through a re-announcement after a lapsed link. +/// Coarsening keeps the newest write by stamp, not log position, so the retired delta matches every working registry. +#[test] +fn an_interaction_that_arrived_out_of_order_retires_to_its_newest_write() { + let mut author = Session::with_peer(PeerId(2)); + author.stage_ops(vec![set_attribute("name", 1), set_attribute("name", 2)]).expect("stage"); + author.end_interaction().expect("close"); + let ops = author.hot_log().to_vec(); + + let mut in_order = Session::with_peer(PeerId(1)); + let mut reversed = Session::with_peer(PeerId(1)); + for hot_op in &ops { + in_order.replay_hot_op(hot_op.clone()).expect("replay"); + } + for hot_op in ops.iter().rev() { + reversed.replay_hot_op(hot_op.clone()).expect("replay"); + } + for host in [&mut in_order, &mut reversed] { + let closed = host.closed_interactions(); + assert_eq!(closed.len(), 1); + host.retire_interaction(&closed[0]).expect("retire"); + assert!(host.registry().value_equal(host.retired_registry()), "the retired value is the one the working registry showed"); + } + assert_eq!(reversed.retired_registry(), in_order.retired_registry(), "arrival order decides nothing"); +} + +/// An interaction retires coarsened. Of the writes to one field only the newest becomes a delta. A whole-list input +/// write supersedes the slot writes before it. A write wiring an input to a node stays while what supersedes it wires +/// elsewhere. The fold is the one every op would have produced. +#[test] +fn an_interaction_retires_to_one_delta_per_field() { + let attribute = |value: u32| set_attribute("name", value); + let wire = |target: u64| RegistryDelta::ChangeNodeInput { + id: NodeId(1), + index: 0, + new_input: crate::NodeInput::Node { id: NodeId(target), index: 0 }, + }; + let value = |value: u32| RegistryDelta::ChangeNodeInput { + id: NodeId(1), + index: 1, + new_input: crate::NodeInput::Value { + value: Value::from(serde_json::json!(value)), + exposed: false, + }, + }; + let mut host = Session::with_peer(PeerId(1)); + host.commit_op_for_test(add_network(3)).expect("network"); + host.document.working_registry = host.document.retired_snapshot.clone(); + let node = |id: u64| RegistryDelta::AddNode { + id: NodeId(id), + node: crate::Node::new(NetworkId(3), crate::Implementation::ProtoNode(ResourceId::from(7)), 2), + }; + let slot_write = RegistryDelta::ChangeNodeInput { + id: NodeId(5), + index: 0, + new_input: crate::NodeInput::Value { + value: Value::from(serde_json::json!(1)), + exposed: false, + }, + }; + let whole_list = RegistryDelta::SetNodeInputs { + id: NodeId(5), + inputs: vec![crate::InputSlot::unset(crate::TimeStamp::ORIGIN)], + }; + let ops = vec![ + node(1), + node(2), + node(4), + node(5), + attribute(1), + value(1), + attribute(2), + wire(2), + value(2), + wire(4), + attribute(3), + value(3), + slot_write, + whole_list, + ]; + let mut raw = host.clone(); + raw.stage_ops(ops.clone()).expect("stage"); + host.stage_ops(ops).expect("stage"); + host.end_interaction().expect("close"); + raw.end_interaction().expect("close"); + + let closed = host.closed_interactions(); + assert_eq!(closed.len(), 1); + let coarsened = host.retire_interaction(&closed[0]).expect("retire"); + let raw_ids: Vec = raw.hot_log().iter().map(crate::HotOp::id).collect(); + let uncoarsened = raw.retire_hot_ops(&raw_ids).expect("retire raw"); + + assert_eq!(uncoarsened.len(), 15, "RegisterPeer, four additions and ten writes"); + assert_eq!( + coarsened.len(), + 10, + "RegisterPeer, four additions, the newest attribute, the newest value, both wires, and the whole list: {coarsened:?}" + ); + assert!(host.hot_log().is_empty()); + assert!(host.retired_registry().value_equal(raw.retired_registry()), "the fold is unchanged"); + assert!(host.registry().value_equal(host.retired_registry()), "the working registry, built from every op, agrees"); + assert_eq!(host.retired_registry(), raw.retired_registry(), "stamps included: the survivor keeps its own"); + let kinds: Vec<&RegistryDelta> = host.history().map(|delta| &delta.kind).collect(); + assert!(kinds.iter().filter(|kind| matches!(kind, RegistryDelta::ChangeDocumentAttribute { .. })).count() == 1); + assert!( + kinds.iter().filter(|kind| matches!(kind, RegistryDelta::ChangeNodeInput { index: 0, .. })).count() == 2, + "the wire to node 2 stays as evidence node 2 exists" + ); + assert!(kinds.iter().filter(|kind| matches!(kind, RegistryDelta::ChangeNodeInput { index: 1, .. })).count() == 1); + assert!(host.history().last().is_some_and(|delta| delta.is_interaction_end())); +} + #[test] fn newest_peer_registration_wins_in_any_order() { // Registration rides the staging path, so the steps go through the hot log and retire. @@ -78,6 +304,26 @@ fn a_batch_past_the_last_sequence_is_refused_whole() { assert!(session.hot_log().is_empty()); } +#[test] +fn a_retraction_settles_its_sequences_so_later_interactions_stay_contiguous() { + let mut session = Session::with_peer(PeerId(1)); + session.stage_ops([set_attribute("a", 1)]).unwrap(); + session.end_interaction().unwrap(); + let up_to = session.hot_log().last().unwrap().timestamp; + session.retire(up_to).unwrap(); + session.stage_ops([set_attribute("b", 1)]).unwrap(); + session.retract_interaction().unwrap(); + + session.stage_ops([set_attribute("c", 1)]).unwrap(); + session.end_interaction().unwrap(); + let closed = session.closed_interactions(); + assert!(closed[0].contiguous, "the retracted sequences are not a gap"); + session.retire_interaction(&closed[0]).unwrap(); + let marks = session.retired_marks(); + assert!(marks.retired_runs.is_empty(), "retired and retracted together leave one prefix"); + assert_eq!(marks.retired_up_to[&PeerId(1)], session.last_hot_sequence()); +} + #[test] fn a_reopened_session_drops_a_late_copy_of_a_retired_op() { let op = crate::HotOp { @@ -264,3 +510,41 @@ 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 } + +#[test] +fn undo_leaves_an_older_hot_interaction_when_the_latest_retired_out_of_order() { + let mut session = Session::with_peer(PeerId(1)); + for key in ["a", "b"] { + session.stage_ops([set_attribute(key, 1)]).unwrap(); + session.end_interaction().unwrap(); + } + let closed = session.closed_interactions(); + session.retire_interaction(&closed[1]).unwrap(); + + assert!(session.retract_interaction().unwrap().is_none(), "the latest interaction retired, so undo goes through history"); + assert!(session.registry().attributes.contains_key("a"), "the older hot interaction stays"); +} + +#[test] +fn undo_takes_back_the_interaction_before_one_already_taken_back() { + let mut session = Session::with_peer(PeerId(1)); + for key in ["a", "b", "c"] { + session.stage_ops([set_attribute(key, 1)]).unwrap(); + session.end_interaction().unwrap(); + } + let closed = session.closed_interactions(); + session.retire_interaction(&closed[1]).unwrap(); + session.retract_interaction().unwrap().expect("the latest interaction is hot"); + + assert!(session.retract_interaction().unwrap().is_none(), "the one before retired, so undo goes through history"); + assert!(session.registry().attributes.contains_key("a")); + + let mut twice = Session::with_peer(PeerId(1)); + for key in ["a", "b"] { + twice.stage_ops([set_attribute(key, 1)]).unwrap(); + twice.end_interaction().unwrap(); + } + twice.retract_interaction().unwrap().expect("b is hot"); + twice.retract_interaction().unwrap().expect("a is hot and now the latest"); + assert!(twice.hot_log().is_empty()); +} diff --git a/document/graph-storage/src/tests/crdt.rs b/document/graph-storage/src/tests/crdt.rs index 422e81f1829..a5767891b7c 100644 --- a/document/graph-storage/src/tests/crdt.rs +++ b/document/graph-storage/src/tests/crdt.rs @@ -753,6 +753,39 @@ fn all_referenced_resource_hashes_survives_undo() { assert!(session.all_referenced_resource_hashes().contains(&changed), "the undone hash change keeps its bytes for redo"); } +/// A hot hash a later hot op overwrote is in no registry, but retiring puts it in history. +#[test] +fn all_referenced_resource_hashes_keeps_an_overwritten_hot_hash() { + use crate::ResourceId; + + let mut session = Session::with_peer(PeerId(1)); + let (first, second) = (ResourceHash::from(&b"first"[..]), ResourceHash::from(&b"second"[..])); + let id = ResourceId::new(); + session.stage_embedded_resource(id, first).expect("stage resource"); + let hot_ops = session.stage_ops([crate::RegistryDelta::SetResourceHash { id, hash: Some(second) }]).expect("stage hash change"); + assert!(session.all_referenced_resource_hashes().contains(&first), "the overwritten hash is still named while hot"); + + session.retire(hot_ops.last().expect("staged").timestamp).expect("retire"); + assert!(session.all_referenced_resource_hashes().contains(&first), "and from history once retired"); +} + +/// A take-back can lose to a host retiring the same ops, so what a taken-back step named stays a root once nothing +/// else names it. +#[test] +fn resource_gc_roots_keep_what_a_taken_back_step_named() { + use crate::ResourceId; + + let mut session = Session::with_peer(PeerId(1)); + let hash = ResourceHash::from(&b"bytes"[..]); + session.stage_embedded_resource(ResourceId::new(), hash).expect("stage resource"); + session.end_interaction().expect("end interaction"); + session.retract_interaction().expect("retract").expect("a hot step to take back"); + assert!(!session.all_referenced_resource_hashes().contains(&hash), "nothing names the hash once taken back"); + + session.stage_embedded_resource(ResourceId::new(), ResourceHash::from(&b"other"[..])).expect("a fresh edit"); + assert!(session.resource_gc_roots().contains(&hash)); +} + /// A commit that produces no deltas must not touch the redo stack. Redo is only abandoned by a real /// new edit; a no-op commit (here `embed_resource_sources` over an empty id set) leaving it cleared /// would silently disable redo after an undo. diff --git a/document/graph-storage/src/tests/sessions.rs b/document/graph-storage/src/tests/sessions.rs index 6014ee04780..6c370943137 100644 --- a/document/graph-storage/src/tests/sessions.rs +++ b/document/graph-storage/src/tests/sessions.rs @@ -1,7 +1,9 @@ -//! Peers exchanging retired history: syncing, following and merging lines, moving the shared head, and undo against what -//! other peers hold. +//! Peers exchanging retired history, moving the shared head, and undoing against what other peers hold. -use crate::{CrdtError, Delta, History, Implementation, MergeOutcome, Network, NetworkId, Node, NodeId, PeerId, RegistryDelta, ResourceEntry, ResourceHash, ResourceId, Rev, Session, UserId, Value}; +use crate::{ + AttributeDelta, CrdtError, Delta, History, HotOp, HotOpId, Implementation, InputSlot, MergeOutcome, Network, NetworkId, Node, NodeId, NodeInput, PeerId, Registry, RegistryDelta, ResourceEntry, + ResourceHash, ResourceId, Rev, Session, TimeStamp, UserId, Value, +}; use std::collections::HashSet; use super::bookkeeping::{add_network, set_attribute}; @@ -382,6 +384,465 @@ fn a_concurrent_remove_and_attribute_change_commute() { assert_eq!(merged([&removal, &change]), merged([&change, &removal])); } +fn sync_pair(peers: &mut [Session], from: usize, to: usize) -> MergeOutcome { + let known = peers[to].known_revs(); + let deltas = peers[from].deltas_unknown_to(known); + peers[to].merge(deltas).expect("merge") +} + +/// The retired snapshot with the hot log folded on top, which the working registry must equal. +fn rebuilt_working(session: &Session) -> Registry { + let mut copy = session.clone(); + copy.document.rebuild_working(); + copy.document.working_registry +} + +struct Lcg(u64); +impl Lcg { + fn below(&mut self, bound: u64) -> u64 { + self.0 = self.0.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + (self.0 >> 33) % bound + } +} + +/// A local op valid under `Live` against `registry`. It only names entities this peer has seen and indices in range. +fn local_op(rng: &mut Lcg, registry: &Registry) -> Option { + let pick = |rng: &mut Lcg, ids: Vec| (!ids.is_empty()).then(|| ids[rng.below(ids.len() as u64) as usize]); + let mut live_networks: Vec = registry.networks.keys().map(|id| id.0).collect(); + live_networks.sort_unstable(); + let mut seen_networks: Vec = registry + .networks + .keys() + .chain(registry.removed_networks.iter().filter(|(_, m)| !m.placeholder).map(|(id, _)| id)) + .map(|id| id.0) + .collect(); + seen_networks.sort_unstable(); + let mut live_nodes: Vec = registry.node_instances.keys().map(|id| id.0).collect(); + live_nodes.sort_unstable(); + let mut seen_nodes: Vec = registry + .node_instances + .keys() + .chain(registry.removed_nodes.iter().filter(|(_, m)| !m.placeholder).map(|(id, _)| id)) + .map(|id| id.0) + .collect(); + seen_nodes.sort_unstable(); + let content = |id: u64| registry.node_or_removed(NodeId(id)).expect("seen"); + let value = |rng: &mut Lcg| NodeInput::Value { + value: Value::from(serde_json::json!(rng.below(50))), + exposed: false, + }; + let input = |rng: &mut Lcg| match pick(rng, live_nodes.clone()) { + Some(id) if rng.below(2) == 0 => NodeInput::Node { id: NodeId(id), index: 0 }, + _ => value(rng), + }; + let attribute = |rng: &mut Lcg, key: &str| AttributeDelta { + key: key.into(), + value: (rng.below(4) > 0).then(|| Value::from(serde_json::json!(rng.below(50)))), + }; + Some(match rng.below(9) { + 0 | 1 => RegistryDelta::ChangeDocumentAttribute { + delta: { + let key = ["x", "y"][rng.below(2) as usize]; + attribute(rng, key) + }, + }, + 2 => { + let id = 1 + rng.below(3); + if registry.networks.contains_key(&NetworkId(id)) { + return None; + } + RegistryDelta::AddNetwork { + id: NetworkId(id), + network: Network::default(), + } + } + 3 => { + let id = pick(rng, live_networks.clone())?; + RegistryDelta::RemoveNetwork { + id: NetworkId(id), + snapshot: registry.networks[&NetworkId(id)].clone(), + } + } + 4 => { + let id = 1 + rng.below(4); + if registry.node_instances.contains_key(&NodeId(id)) { + return None; + } + let network = pick(rng, seen_networks.clone())?; + RegistryDelta::AddNode { + id: NodeId(id), + node: Node::new(NetworkId(network), Implementation::Network(NetworkId(network)), 1 + rng.below(2) as usize), + } + } + 5 => { + let id = pick(rng, live_nodes.clone())?; + RegistryDelta::RemoveNode { + id: NodeId(id), + snapshot: registry.node_instances[&NodeId(id)].clone(), + } + } + 6 => RegistryDelta::ChangeNodeAttribute { + id: NodeId(pick(rng, seen_nodes.clone())?), + delta: { + let key = ["name", "lock"][rng.below(2) as usize]; + attribute(rng, key) + }, + }, + 7 => { + let id = pick(rng, seen_nodes.clone())?; + let len = content(id).inputs().len() as u64; + if len == 0 { + return None; + } + RegistryDelta::ChangeNodeInput { + id: NodeId(id), + index: rng.below(len) as u32, + new_input: input(rng), + } + } + _ => { + let id = pick(rng, seen_nodes.clone())?; + let inputs = (0..1 + rng.below(3)) + .map(|_| InputSlot { + input: input(rng), + timestamp: TimeStamp::ORIGIN, + attributes: Default::default(), + }) + .collect(); + RegistryDelta::SetNodeInputs { id: NodeId(id), inputs } + } + }) +} + +fn check_invariants(session: &Session, context: &str) { + assert_eq!( + session.retired_registry(), + &session.snapshot_from_history().unwrap(), + "{context}: retired snapshot is not the fold of the head's line" + ); + assert_eq!(session.registry(), &rebuilt_working(session), "{context}: working registry is not snapshot + hot log"); +} + +/// A session rebuilt as a reopen does. It loads the retired snapshot, history, head, redo stack and node counter, +/// then the hot sequence, clock and retired marks from `session.json`, then replays the hot log. +fn reopened(s: &Session) -> Session { + let mut r = Session::load( + s.peer(), + s.user(), + s.retired_registry().clone(), + s.cloned_deltas(), + s.head_rev(), + s.redo_stack().to_vec(), + s.next_node_counter(), + ); + r.restore_hot_sequence(s.last_hot_sequence()); + r.absorb_retired_marks(s.retired_marks()); + for hot_op in s.hot_log() { + r.replay_hot_op(hot_op.clone()).unwrap(); + } + r.restore_clock_counter(s.clock_counter()); + r +} + +/// Undo `session` only if every delta it would take off the line is unknown to all `others` (the silent zone). +fn undo_is_silent(session: &Session, others: &[&Session]) -> bool { + let mut probe = session.clone(); + let before = probe.document.history.ancestors(probe.head_rev()); + probe.undo().unwrap(); + let after = probe.document.history.ancestors(probe.head_rev()); + before.difference(&after).all(|rev| others.iter().all(|other| !other.document.history.contains(*rev))) +} + +/// Three peers of different users stage, broadcast, retire, undo, redo, sync and reopen in random order. +/// After every action the retired snapshot must be the fold of the head's line, and the working registry must be +/// snapshot + hot log. A reopen must restore both exactly. After flushing and syncing to quiescence, all retired +/// registries must be identical, stamps included. +#[test] +fn peers_keep_the_invariants_and_converge_under_random_undo_redo_sync_and_reopen() { + const PEERS: usize = 3; + for seed in 0..300u64 { + let mut rng = Lcg(seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) ^ 0xD1B5); + let mut peers: Vec = (1..=PEERS as u64).map(|id| Session::with_peer(PeerId(id))).collect(); + // Shared base so every peer has a boundary to undo back to. + step(&mut peers[0], set_attribute("base", 0)); + for i in 1..PEERS { + sync_pair(&mut peers, 0, i); + } + let mut in_flight: Vec> = vec![Vec::new(); PEERS]; // hot ops on their way to peer i + let mut log = Vec::new(); + + for action in 0..50 { + let me = rng.below(PEERS as u64) as usize; + let kind = rng.below(12); + log.push(format!("{me}:{kind}")); + let context = format!("seed {seed}, action {action}, trace {log:?}"); + match kind { + 0..=2 => { + let ops: Vec = (0..1 + rng.below(3)).filter_map(|_| local_op(&mut rng, peers[me].registry())).collect(); + if let Ok(staged) = peers[me].stage_ops(ops) { + for hot_op in staged { + for other in (0..PEERS).filter(|&other| other != me) { + if rng.below(4) == 0 { + in_flight[other].push(hot_op.clone()); + } else { + peers[other].replay_hot_op(hot_op.clone()).unwrap(); + } + } + } + } + } + 3 | 4 => { + let stamps: Vec = peers[me].hot_log().iter().map(|hot_op| hot_op.timestamp).collect(); + if stamps.is_empty() { + continue; + } + let up_to = stamps[rng.below(stamps.len() as u64) as usize]; + let ids: Vec = peers[me].hot_ops_up_to(up_to); + let revs = peers[me].retire(up_to).unwrap(); + if let Some(&last) = revs.last() + && rng.below(4) > 0 + { + peers[me].mark_interaction_end(last); + } + for other in (0..PEERS).filter(|&other| other != me) { + peers[other].discard_hot_ops(&ids); + } + } + 5 => { + if peers[me].can_undo() { + let others: Vec<&Session> = peers.iter().enumerate().filter(|(i, _)| *i != me).map(|(_, s)| s).collect(); + if undo_is_silent(&peers[me], &others) { + peers[me].undo().unwrap(); + } + } + } + 6 => { + if peers[me].can_redo() { + peers[me].redo().unwrap(); + } + } + 7 | 8 => { + let from = (me + 1 + rng.below(PEERS as u64 - 1) as usize) % PEERS; + sync_pair(&mut peers, from, me); + } + 9 => { + let fresh = reopened(&peers[me]); + assert_eq!(fresh.retired_registry(), peers[me].retired_registry(), "reopen snapshot, {context}"); + assert_eq!(fresh.registry(), peers[me].registry(), "reopen working, {context}"); + assert_eq!(fresh.retired_marks(), peers[me].retired_marks(), "reopen marks, {context}"); + peers[me] = fresh; + } + _ => { + for hot_op in std::mem::take(&mut in_flight[me]) { + peers[me].replay_hot_op(hot_op).unwrap(); + } + } + } + for (i, peer) in peers.iter().enumerate() { + check_invariants(peer, &format!("peer {i}, {context}")); + } + } + + // Deliver everything, let each peer retire what it holds with the others discarding, then sync to quiescence. + for i in 0..PEERS { + for hot_op in std::mem::take(&mut in_flight[i]) { + peers[i].replay_hot_op(hot_op).unwrap(); + } + } + for me in 0..PEERS { + if let Some(up_to) = peers[me].hot_log().iter().map(|hot_op| hot_op.timestamp).max() { + let ids = peers[me].hot_ops_up_to(up_to); + peers[me].retire(up_to).unwrap(); + for other in (0..PEERS).filter(|&other| other != me) { + peers[other].discard_hot_ops(&ids); + } + } + } + for _ in 0..2 * PEERS { + for from in 0..PEERS { + for to in 0..PEERS { + if from != to { + sync_pair(&mut peers, from, to); + } + } + } + } + let context = format!("seed {seed}, trace {log:?}"); + assert!(peers[0].hot_log().is_empty(), "{context}: peer 0 still holds hot ops"); + for i in 1..PEERS { + assert_eq!(peers[0].head_rev(), peers[i].head_rev(), "{context}: heads of 0 and {i} differ after syncing to quiescence"); + assert_eq!(peers[0].retired_registry(), peers[i].retired_registry(), "{context}: retired registries of 0 and {i} diverged"); + assert!(peers[i].hot_log().is_empty(), "{context}: peer {i} still holds hot ops"); + } + } +} + +/// A peer's op that was seen and then discarded before its delta arrived is in neither history nor the hot log, so a +/// reopen takes the clock from `session.json`, so a later write here must still be stamped above it. +#[test] +fn a_write_after_a_reopen_beats_a_discarded_op_it_saw() { + let mut a = Session::with_peer(PeerId(1)); + step(&mut a, set_attribute("base", 0)); + let mut b = Session::with_peer(PeerId(2)); + sync(&mut a, &mut b); + for _ in 0..10 { + b.stage_ops([set_attribute("unrelated", 0)]).unwrap(); + } + let seen = b.stage_ops([set_attribute("k", 1)]).unwrap(); + for hot_op in b.hot_log().to_vec() { + a.replay_hot_op(hot_op).unwrap(); + } + let up_to = seen.last().unwrap().timestamp; + let ids = b.hot_ops_up_to(up_to); + b.retire(up_to).unwrap(); + a.discard_hot_ops(&ids); + + // A saw k = 1, reopens, then writes k = 2. + let mut a = reopened(&a); + let written = a.stage_ops([set_attribute("k", 2)]).unwrap(); + a.retire(written.last().unwrap().timestamp).unwrap(); + sync(&mut b, &mut a); + assert_eq!( + a.retired_registry().attributes.get("k").map(|value| value.value.clone()), + Some(Value::from(serde_json::json!(2))), + "A's write came after it saw k = 1, but is stamped below it" + ); +} + +/// This peer's own op is retired by another and discarded here before the delta arrives. After a reopen the next op +/// must not get its stamp, since two ops sharing a stamp tie and peers holding both would show different values. +#[test] +fn a_reopen_after_discarding_an_own_op_never_reuses_its_stamp() { + let mut a = Session::with_peer(PeerId(1)); + let registered = a.stage_ops([set_attribute("base", 0)]).unwrap(); + a.retire(registered.last().unwrap().timestamp).unwrap(); + let mut b = Session::with_peer(PeerId(2)); + sync(&mut a, &mut b); + + let first = a.stage_ops([set_attribute("k", 1)]).unwrap(); + for hot_op in &first { + b.replay_hot_op(hot_op.clone()).unwrap(); + } + let up_to = first.last().unwrap().timestamp; + let ids = b.hot_ops_up_to(up_to); + b.retire(up_to).unwrap(); + a.discard_hot_ops(&ids); + + let mut a = reopened(&a); + let second = a.stage_ops([set_attribute("k", 2)]).unwrap(); + let reused = second.iter().any(|new| first.iter().any(|old| old.timestamp == new.timestamp)); + // Both peers hold the same two ops. B has the first as a delta and the second hot, and A matches once B's delta lands. + for hot_op in &second { + b.replay_hot_op(hot_op.clone()).unwrap(); + } + assert!(matches!(sync(&mut b, &mut a), MergeOutcome::FastForward(_))); + let k = |registry: &Registry| registry.attributes.get("k").map(|value| value.value.clone()); + assert_eq!(k(a.registry()), k(b.registry()), "same ops, different live value; reused stamp: {reused}"); + assert_eq!(a.registry(), &rebuilt_working(&a), "A's working registry is not snapshot + hot log"); + assert!(!reused, "a fresh op reused the stamp of the discarded one"); +} + +/// Undoing a retired step drops it out of the shared line and mints later steps again on its parent. Every peer +/// following the head move converges, a field a later step wrote keeps that value, and redo brings the step back +/// as a copy on top. +#[test] +fn a_dropped_interaction_leaves_the_line_and_the_room_follows() { + fn catch_up(guest: &mut Session, host: &Session) { + let missing: Vec = host.cloned_deltas().into_iter().filter(|delta| guest.delta(delta.id).is_none()).collect(); + guest.merge(missing).expect("merge"); + } + let mut host = Session::with_peer(PeerId(1)); + host.commit_op_for_test(set_attribute("base", 0)).expect("base"); + let base = host.head_rev().expect("rev"); + host.mark_interaction_end(base); + + // The guest's step, retired by the host, writes its own key and one the host writes after it. + let mut guest = Session::load(PeerId(2), UserId(2), host.retired_registry().clone(), host.cloned_deltas(), host.head_rev(), Vec::new(), 0); + let guest_ops = [set_attribute("guest", 1), set_attribute("shared", 1)]; + let hot = guest.stage_ops(guest_ops).expect("stage"); + for hot_op in &hot { + host.replay_hot_op(hot_op.clone()).expect("the guest's ops reach the host"); + } + let guest_ids: Vec = hot.iter().map(HotOp::id).collect(); + let revs = host.retire_hot_ops(&guest_ids).expect("retire"); + let undone = *revs.last().expect("the guest's step"); + host.mark_interaction_end(undone); + catch_up(&mut guest, &host); + guest.discard_hot_ops(&guest_ids); + + // A later step by the host writes the shared key. + host.commit_op_for_test(set_attribute("shared", 2)).expect("later step"); + let later = host.head_rev().expect("rev"); + host.mark_interaction_end(later); + host.document.history.set_retired_at(later, 1_234); + catch_up(&mut guest, &host); + assert_eq!(guest.head_rev(), host.head_rev()); + assert_eq!(guest.latest_own_interaction(), Some(undone)); + + let (moved, touched) = host.drop_interaction(undone).expect("drop"); + assert_eq!(moved.from, Some(later)); + assert_eq!(moved.copies.len(), 1, "the later step is minted again"); + assert_eq!(moved.copies[0].parent, Some(base), "the copy hangs off the dropped step's parent"); + assert!(moved.copies[0].is_interaction_end(), "the copy keeps its interaction end"); + assert_eq!(moved.copies[0].retired_at(), Some(1_234), "the copy keeps when the original was retired"); + assert!(touched.is_empty(), "document attributes name no graph entity"); + let snapshot = host.retired_registry(); + assert!(snapshot.attributes.get("guest").is_none_or(|value| value.deleted), "the dropped step's write is gone"); + assert_eq!( + snapshot.attributes.get("shared").map(|value| &value.value), + Some(&Value::from(serde_json::json!(2))), + "the later step's write stands" + ); + assert!(host.delta(undone).is_some() && host.delta(later).is_some(), "the dropped line stays as a branch"); + assert!(host.registry().value_equal(host.retired_registry())); + + guest.apply_head_move(&moved).expect("follow"); + assert_eq!(guest.head_rev(), host.head_rev()); + assert_eq!(guest.retired_registry(), host.retired_registry()); + assert_eq!(guest.history().map(|delta| delta.id).collect::>(), host.history().map(|delta| delta.id).collect::>()); + assert!( + matches!(guest.apply_head_move(&moved), Err(CrdtError::CursorMismatch { .. })), + "a move only applies from where it started" + ); + + // The redone step returns as a copy on top, its older stamp losing the shared key. + let restored = host.restore_interaction(undone).expect("restore"); + assert_eq!(restored.len(), 3, "RegisterPeer and the two writes come back"); + assert!(host.retired_registry().attributes.get("guest").is_some_and(|value| value.value == Value::from(serde_json::json!(1)))); + assert_eq!(host.retired_registry().attributes.get("shared").map(|value| &value.value), Some(&Value::from(serde_json::json!(2)))); + catch_up(&mut guest, &host); + assert_eq!(guest.head_rev(), host.head_rev()); + assert_eq!(guest.retired_registry(), host.retired_registry()); +} + +/// Undo of a retired step names the person's latest step, not the device's. A fresh copy of the document under a new +/// peer id still undoes what its user did from the old one, and nobody else's. +#[test] +fn undo_in_a_session_finds_the_users_step_from_another_peer() { + let mut host = Session::with_identity(PeerId(1), UserId(1)); + host.commit_op_for_test(set_attribute("base", 0)).expect("base"); + let base = host.head_rev().expect("rev"); + host.mark_interaction_end(base); + + let mut old_device = Session::load(PeerId(2), UserId(7), host.retired_registry().clone(), host.cloned_deltas(), host.head_rev(), Vec::new(), 0); + let hot = old_device.stage_ops([set_attribute("mine", 1)]).expect("stage"); + for hot_op in &hot { + host.replay_hot_op(hot_op.clone()).expect("the step reaches the host"); + } + let ids: Vec = hot.iter().map(HotOp::id).collect(); + let revs = host.retire_hot_ops(&ids).expect("retire"); + let step = *revs.last().expect("the step"); + host.mark_interaction_end(step); + + let new_device = Session::load(PeerId(3), UserId(7), host.retired_registry().clone(), host.cloned_deltas(), host.head_rev(), Vec::new(), 0); + assert_eq!(new_device.latest_own_interaction(), Some(step), "the same person under a new peer id owns the step"); + assert!(new_device.is_mine(PeerId(2))); + + let stranger = Session::load(PeerId(4), UserId(8), host.retired_registry().clone(), host.cloned_deltas(), host.head_rev(), Vec::new(), 0); + assert_eq!(stranger.latest_own_interaction(), None, "someone else has nothing of theirs to undo"); + assert!(!stranger.is_mine(PeerId(2))); +} + /// The retirer moves the head back: the line since stays as a branch, and a follower folds to the same state. #[test] fn moving_the_head_back_leaves_the_line_since_as_a_branch_everyone_follows() { @@ -539,3 +1000,50 @@ fn a_head_move_publishes_the_new_head() { guest.apply_head_move(&moved).expect("follow"); assert!(!guest.can_undo(), "the follower holds the same line"); } + +/// Dropping hot ops another peer retired re-derives the working registry without them until their deltas arrive. +#[test] +fn discarded_hot_ops_leave_the_working_registry_at_the_snapshot() { + let op = HotOp { + op: set_attribute("k", 1), + timestamp: TimeStamp { counter: 3, peer: PeerId(2) }, + sequence: crate::HotSequence(1), + attributes: Default::default(), + }; + let mut session = Session::with_peer(PeerId(1)); + session.replay_hot_op(op.clone()).unwrap(); + session.discard_hot_ops(&[op.id()]); + assert!(session.hot_log().is_empty()); + assert_eq!(session.registry(), session.retired_registry()); +} + +#[test] +fn a_drop_whose_copy_fails_leaves_the_session_unchanged() { + let mut session = Session::with_peer(PeerId(1)); + session.commit_op_for_test(add_network(3)).expect("base"); + let base = session.head_rev().expect("rev"); + session.mark_interaction_end(base); + session.commit_op_for_test(set_attribute("dropped", 1)).expect("dropped step"); + let undone = session.head_rev().expect("rev"); + session.mark_interaction_end(undone); + + // A later step no fold can apply, so minting its copy fails. + let op = RegistryDelta::SetNetworkExport { + id: NetworkId(3), + index: crate::ids::MAX_EXPORT_SLOTS as u32, + export: None, + }; + let reverse = crate::prior::capture(session.retired_registry(), &op); + let timestamp = TimeStamp { counter: 100, peer: PeerId(1) }; + let later = Delta::new(Some(undone), PeerId(1), timestamp, op, reverse); + let head = later.id; + session.document.history.push(later); + session.document.head = Some(head); + + let (registry, retired, history_len) = (session.registry().clone(), session.retired_registry().clone(), session.history_len()); + assert!(matches!(session.drop_interaction(undone), Err(CrdtError::ExportSlotOutOfBounds(_)))); + assert_eq!(session.head_rev(), Some(head)); + assert_eq!(session.history_len(), history_len); + assert_eq!(session.retired_registry(), &retired); + assert_eq!(session.registry(), ®istry); +} diff --git a/document/migrations/formats/src/lib.rs b/document/migrations/formats/src/lib.rs index 5c2e03bdc95..e3fa7cb8df9 100644 --- a/document/migrations/formats/src/lib.rs +++ b/document/migrations/formats/src/lib.rs @@ -6,6 +6,7 @@ pub mod v1; pub mod v2; -pub use v2 as latest; +pub mod v3; +pub use v3 as latest; mod steps; diff --git a/document/migrations/formats/src/steps/mod.rs b/document/migrations/formats/src/steps/mod.rs index 63207ecdd48..04a962dcf2b 100644 --- a/document/migrations/formats/src/steps/mod.rs +++ b/document/migrations/formats/src/steps/mod.rs @@ -1,3 +1,4 @@ //! One format step per version, each registering itself. mod v1_to_v2; +mod v2_to_v3; diff --git a/document/migrations/formats/src/steps/v2_to_v3.rs b/document/migrations/formats/src/steps/v2_to_v3.rs new file mode 100644 index 00000000000..c648a2dd4ec --- /dev/null +++ b/document/migrations/formats/src/steps/v2_to_v3.rs @@ -0,0 +1,54 @@ +//! Format step 2 to 3, generated by `cargo run migration snapshot`. + +use crate::{v2, v3}; +use migration_core::{MigrationError, Upgrade, format_migration}; + +format_migration! { + 2 => 3, + session: v2::SessionState, +} + +impl Upgrade for v2::RetiredMarks { + type Next = v3::RetiredMarks; + fn upgrade(self) -> Result { + let v2::RetiredMarks { retired_up_to, retired_runs } = self; + Ok(v3::RetiredMarks { + retired_up_to, + retired_runs, + taken_back: Default::default(), + }) + } +} + +impl Upgrade for v2::SessionState { + type Next = v3::SessionState; + fn upgrade(self) -> Result { + let v2::SessionState { + peer_id, + user_id, + head_rev, + last_broadcast_rev, + redo_stack, + next_node_counter, + last_hot_sequence, + view_settings, + network_view_settings, + retired_marks, + clock_counter, + } = self; + Ok(v3::SessionState { + peer_id, + user_id, + head_rev, + last_broadcast_rev, + redo_stack, + next_node_counter, + last_hot_sequence, + view_settings, + network_view_settings, + retired_marks: Upgrade::upgrade(retired_marks)?, + clock_counter, + shared: false, + }) + } +} diff --git a/document/migrations/formats/src/v3.rs b/document/migrations/formats/src/v3.rs new file mode 100644 index 00000000000..eaf3fe96840 --- /dev/null +++ b/document/migrations/formats/src/v3.rs @@ -0,0 +1,44 @@ +//! Format version 3, frozen by `cargo run migration snapshot`. Never edit it: old documents decode through these definitions. + +pub use super::v2::{ + AttributeDelta, AttributeValue, Attributes, Delta, ExportSlot, HotOp, HotOpId, HotSequence, Implementation, InputSlot, Network, NetworkField, NetworkId, Node, NodeField, NodeId, NodeInput, + NonFinitePriority, PeerId, PeerRegistration, Prior, Priority, ProtoNode, Registry, RegistryDelta, ResourceEntry, ResourceStore, Rev, SourceKey, SourceValue, TimeStamp, Tombstone, UserId, Value, +}; +use serde::{Deserialize, Serialize}; +use std::collections::{BTreeMap, HashMap}; + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RetiredMarks { + pub retired_up_to: HashMap, + pub retired_runs: HashMap>, + #[serde(default)] + pub taken_back: HashMap>, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct SessionState { + #[serde(default)] + pub peer_id: PeerId, + #[serde(default)] + pub user_id: UserId, + #[serde(default)] + pub head_rev: Option, + #[serde(default)] + pub last_broadcast_rev: Option, + #[serde(default)] + pub redo_stack: Vec, + #[serde(default)] + pub next_node_counter: u64, + #[serde(default)] + pub last_hot_sequence: HotSequence, + #[serde(default)] + pub view_settings: BTreeMap, + #[serde(default)] + pub network_view_settings: BTreeMap>, + #[serde(default)] + pub retired_marks: RetiredMarks, + #[serde(default)] + pub clock_counter: u64, + #[serde(default)] + pub shared: bool, +} diff --git a/document/migrations/formats/tests/latest.rs b/document/migrations/formats/tests/latest.rs index a5bed6019a1..68b2b1a0076 100644 --- a/document/migrations/formats/tests/latest.rs +++ b/document/migrations/formats/tests/latest.rs @@ -229,6 +229,7 @@ fn session_state_encodes_identically() { network_view_settings: BTreeMap::from([(ROOT_NETWORK, BTreeMap::from([("ui::nav".to_string(), Value::Float(2.))]))]), retired_marks: session.retired_marks().clone(), clock_counter: session.clock_counter(), + shared: false, }; assert_eq!(through_frozen_json::(&state), serde_json::to_string(&state).unwrap());