Add the session operations for taking back, dropping and restoring interactions and for following a retirer

Author: TrueDoctor

document/format/src/export.rs

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

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

document/format/src/persist.rs

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

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"
+		);
 	});
 }
 

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<Rev>, author: PeerId, timestamp: TimeStamp, kind: RegistryDelta, reverse: Vec<Prior>) -> Self {
@@ -79,6 +80,17 @@ impl Delta {
 	}
 }
 
+impl RegistryDelta {
+	pub fn resource_hash(&self) -> Option<ResourceHash> {
+		match self {
+			RegistryDelta::AddResource { entry, .. } => entry.hash,
+			RegistryDelta::RemoveResource { snapshot, .. } => snapshot.hash,
+			RegistryDelta::SetResourceHash { hash, .. } => *hash,
+			_ => None,
+		}
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;

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<HotOp>,
 	/// The stamps of the hot log's ops, so a re-announced op is recognised without a scan.
 	pub(crate) hot_timestamps: HashSet<TimeStamp>,
-	/// 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<Item = HotOpId>) -> 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<HotOp>) {
+		let wanted: HashSet<HotOpId> = ids.iter().copied().collect();
+		let taken: Vec<HotOp> = 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();

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 {

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<MetadataFact>", into = "Vec<MetadataFact>")]
 pub struct HistoryMetadata {
 	facts: BTreeMap<Subject, BTreeMap<String, MetadataFact>>,
 }
 
+impl From<Vec<MetadataFact>> for HistoryMetadata {
+	fn from(facts: Vec<MetadataFact>) -> Self {
+		Self::fold(facts)
+	}
+}
+
+impl From<HistoryMetadata> for Vec<MetadataFact> {
+	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<MetadataFact> {
+		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)]);

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<K: std::hash::Hash + Eq, T>(a: &HashMap<K, Tombstone<T>>, b: &HashMap<K, Tombstone<T>>) -> 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

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<PeerId, HotSequence>,
 	/// retired ops past their author's prefix, as inclusive runs, one per gap in flight.
 	pub retired_runs: HashMap<PeerId, Vec<(HotSequence, HotSequence)>>,
+	/// Which covered ops their author took back rather than retired, as inclusive runs.
+	#[serde(default)]
+	pub taken_back: HashMap<PeerId, Vec<(HotSequence, HotSequence)>>,
 }