Add document-format crate (#4234)

* Add document-format crate

* Adapt document-format to merged graph-storage and consolidate errors

* Fix resource write ordering, codec error masking, and export docs

* Remove document-format RFC; tracked in a dedicated PR

* Restructure document-format into gdd/persist/export/resource modules

Split the 963-line lib.rs by responsibility: persist path into persist.rs, resource I/O into resource.rs, export engine into export.rs. Rename the GddV1 layout struct to GddV1Layout and add a GddV1 = Gdd<GddV1Layout> alias.

* Feature-split document-format and graph-storage for minimal builds

* Address PR review: O(1) history-delta lookup and no-default-features test build

* Address PR review: propagate apply_hot_op persist error and assert resource hash

* Make decompression from archive streaming

* Share archive export body between export and export_to_bytes

* Review cleanup

* Export documents with un-retired hot ops losslessly

* Persist last_broadcast_rev in session.json, drop unused last_retired_at

* Move peer_id from manifest to session.json

* Rename gesture to interaction in document-format storage layer

document/graph-storage/src/lib.rs

@@ -21,14 +21,18 @@ pub mod to_runtime;
 pub use attributes::*;
 pub use crdt::*;
 pub use document::*;
-pub use from_runtime::{RuntimeConversion, decode_declaration, encode_declaration};
 pub use history::History;
 pub use ids::*;
-pub use metadata_source::{InputMetadataEntry, NetworkMetadataEntry, NoMetadata, NodeMetadataEntry, NodeMetadataSource, Position};
 pub use model::*;
 pub use registry::*;
 pub use resources::*;
 pub use session::*;
+
+#[cfg(any(feature = "conversion", test))]
+pub use from_runtime::{RuntimeConversion, decode_declaration, encode_declaration};
+#[cfg(any(feature = "conversion", test))]
+pub use metadata_source::{InputMetadataEntry, NetworkMetadataEntry, NoMetadata, NodeMetadataEntry, NodeMetadataSource, Position};
+#[cfg(any(feature = "conversion", test))]
 pub use to_runtime::Declarations;
 
 #[cfg(test)]

document/graph-storage/src/session.rs

@@ -1,5 +1,8 @@
+#[cfg(any(feature = "conversion", test))]
+use crate::NodeMetadataSource;
+#[cfg(any(feature = "conversion", test))]
 use crate::from_runtime;
-use crate::{ApplyMode, Delta, Document, History, LamportClock, NetworkId, NodeId, NodeMetadataSource, PeerId, Registry, RegistryDelta, RegistryTarget, ResourceEntry, Rev, TimeStamp, UserId};
+use crate::{ApplyMode, Delta, Document, History, LamportClock, NetworkId, NodeId, PeerId, Registry, RegistryDelta, RegistryTarget, ResourceEntry, Rev, TimeStamp, UserId};
 use graphene_resource::{ResourceHash, ResourceId};
 use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, HashSet};
@@ -19,6 +22,7 @@ impl Session {
 	/// Mints a fresh `PeerId` from the process-wide UUID generator and wraps an empty `Document`.
 	/// Two peers in the same process will collide (the generator is seeded once); use `with_peer`
 	/// in tests where determinism matters.
+	#[cfg(any(feature = "conversion", test))]
 	pub fn new() -> Self {
 		Self::with_peer(PeerId(core_types::uuid::generate_uuid()))
 	}
@@ -51,6 +55,12 @@ impl Session {
 		&self.document.working_registry
 	}
 
+	/// The registry after applying retired history only, without the unretired hot tail. Persisted as the
+	/// snapshot alongside `history` + hot log so a reopen restores the same retired-then-hot layering.
+	pub fn retired_registry(&self) -> &Registry {
+		&self.document.retired_snapshot
+	}
+
 	/// Diff the current registry against a fresh conversion of `network`, then commit each emitted
 	/// op as its own `Delta` on the local chain. One `clock.tick()` per op (strictly causal within
 	/// a commit). Returns the new `Rev`s in commit order (empty if nothing changed) plus the
@@ -107,15 +117,15 @@ impl Session {
 			ops.push(RegistryDelta::AddSource { id, key, source: embedded.clone() });
 		}
 
-		// Caller contract: this runs on a throwaway export clone with no unretired hot ops, so the
-		// working registry equals the snapshot. Overwriting working with the advanced snapshot below
-		// would otherwise drop hot-zone edits, so reject the call rather than corrupt state.
-		if !self.document.hot_log.is_empty() {
-			return Err(CrdtError::HotLogNotEmpty);
-		}
-
+		// These are retired deltas, so `commit_ops` advances the retired snapshot and history. The working
+		// registry sits at `retired_snapshot + hot tail`, so mirror each committed delta onto it with its own
+		// timestamp rather than cloning the snapshot over it, which would discard any unretired hot-zone edits.
 		let revs = self.commit_ops(ops, false)?;
-		self.document.working_registry = self.document.retired_snapshot.clone();
+		for &rev in &revs {
+			let Some(delta) = self.document.history.get(rev) else { continue };
+			let (kind, timestamp) = (delta.kind.clone(), delta.timestamp);
+			self.document.apply_op_idempotent(kind, timestamp)?;
+		}
 		Ok(revs)
 	}
 
@@ -432,6 +442,12 @@ impl Session {
 		self.document.history.iter()
 	}
 
+	/// The retired delta for `rev`, or `None` if it isn't in history. O(1) lookup, for callers that
+	/// already hold the revs they want (e.g. persisting a freshly-retired batch) and don't need a scan.
+	pub fn delta(&self, rev: Rev) -> Option<&Delta> {
+		self.document.history.get(rev)
+	}
+
 	/// Verify the retired history loaded from an untrusted source: content-addressed ids match their
 	/// recomputed hashes, and the deltas are topologically ordered. See [`History::verify`].
 	pub fn verify_history(&self) -> Result<(), CrdtError> {
@@ -465,6 +481,19 @@ impl Session {
 		self.document.head
 	}
 
+	/// The latest retired commit broadcast to at least one peer. Commits after it are silently
+	/// rewritable; commits at or before it are published. `None` until broadcast transport lands.
+	pub fn last_broadcast_rev(&self) -> Option<Rev> {
+		self.document.last_broadcast_rev
+	}
+
+	/// Advance the published frontier to `rev` as commits are broadcast. The frontier is monotonic, so
+	/// this only moves it forward (never back to `None`). Set by the (future) broadcast transport;
+	/// persisted in `session.json` so the silent/published boundary survives a reopen.
+	pub fn publish_up_to(&mut self, rev: Rev) {
+		self.document.last_broadcast_rev = Some(rev);
+	}
+
 	/// Test-only: every retired delta, cloned, for feeding one session's branch into another's `merge`.
 	#[cfg(test)]
 	pub(crate) fn cloned_deltas(&self) -> Vec<Delta> {
@@ -488,6 +517,7 @@ impl Session {
 }
 
 /// Errors from `Session::commit_from_runtime`.
+#[cfg(any(feature = "conversion", test))]
 #[derive(Debug, thiserror::Error)]
 pub enum CommitError {
 	#[error("Failed to convert runtime network: {0}")]
@@ -496,6 +526,7 @@ pub enum CommitError {
 	Crdt(#[from] CrdtError),
 }
 
+#[cfg(any(feature = "conversion", test))]
 impl Default for Session {
 	fn default() -> Self {
 		Self::new()
@@ -537,8 +568,6 @@ pub enum CrdtError {
 	/// PeerId is already registered to a different UserId.
 	#[error("Peer {0:?} is already registered to a different user")]
 	PeerRegistrationConflict(PeerId),
-	#[error("Operation requires an empty hot log")]
-	HotLogNotEmpty,
 	#[error("Delta stored under {stored} hashes to {expected}")]
 	RevMismatch { stored: Rev, expected: Rev },
 }

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

@@ -777,20 +777,33 @@ fn no_op_commit_preserves_redo_stack() {
 	assert!(session.can_redo(), "a no-op commit must not clear the redo stack");
 }
 
-/// `embed_resource_sources` overwrites the working registry with the snapshot, valid only when no
-/// unretired hot ops are present. Called with a non-empty hot log it must error rather than silently
-/// drop the hot-zone edits.
+/// `embed_resource_sources` commits its `AddSource` deltas as retired, then mirrors them onto the
+/// working registry. With unretired hot ops present it must keep the hot-zone edits (export of a
+/// mid-interaction document is lossless) rather than clobbering the working registry with the snapshot.
 #[test]
-fn embed_resource_sources_rejects_unretired_hot_ops() {
+fn embed_resource_sources_preserves_unretired_hot_ops() {
 	let mut session = Session::with_peer(PeerId(1));
 	let resources = graphene_resource::ResourceRegistry::new();
 
-	// Stage without retiring, leaving hot ops in the log.
-	session.stage_from_runtime(&tiny_network(), &NoMetadata, &resources).expect("stage");
-	assert!(!session.hot_log().is_empty(), "staging should leave unretired hot ops");
+	// Retire a base so the network's nodes live in the retired snapshot.
+	session.stage_from_runtime(&tiny_network(), &NoMetadata, &resources).expect("stage base");
+	let base_up_to = session.hot_log().last().expect("staged base").timestamp;
+	session.retire(base_up_to).expect("retire base");
 
-	let result = session.embed_resource_sources(std::iter::empty::<ResourceId>());
-	assert!(matches!(result, Err(crate::CrdtError::HotLogNotEmpty)), "expected HotLogNotEmpty, got {result:?}");
+	// Stage an embedded resource without retiring, leaving it in the hot log (the working registry now
+	// holds it, the retired snapshot does not).
+	let hash = ResourceHash::from(&b"hot-resource"[..]);
+	let id = ResourceId::new();
+	session.stage_embedded_resource(id, hash).expect("stage resource");
+	assert!(!session.hot_log().is_empty(), "staging should leave unretired hot ops");
+	assert!(session.registry().resources.contains_key(&id), "working registry should hold the hot resource");
+
+	session.embed_resource_sources(std::iter::empty::<ResourceId>()).expect("embed tolerates a non-empty hot log");
+
+	// The hot-zone resource survives in the working registry (not reset to the snapshot), and the hot log
+	// is untouched so a later retire still promotes it.
+	assert!(session.registry().resources.contains_key(&id), "hot resource must survive the embed");
+	assert!(!session.hot_log().is_empty(), "embed must not drain the hot log");
 }
 
 /// A delta's `Rev` is content-addressed, so two byte-equal deltas must hash identically regardless