Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions document/graph-storage/src/crdt.rs
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ impl Delta {

/// Op payload. Timestamps live on the wrapping `Delta` — one per delta, applied to all LWW-eligible
/// writes within. See `notes/document-format-collaboration.md`.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum RegistryDelta {
AddNode {
id: NodeId,
Expand Down Expand Up @@ -190,7 +190,7 @@ pub enum RegistryDelta {
}

/// `value: None` means remove. The timestamp comes from the wrapping `Delta`.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct AttributeDelta {
pub key: String,
pub value: Option<serde_json::Value>,
Expand Down
237 changes: 211 additions & 26 deletions document/graph-storage/src/from_runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -182,33 +182,40 @@ fn convert_resources(resources: &graphene_resource::ResourceRegistry, referenced
if !referenced.contains(&id) {
continue;
}
let Some(info) = resources.info(&id) else { continue };

let mut entry = crate::ResourceEntry {
hash: info.hash.copied(),
hash_timestamp: TimeStamp::ORIGIN,
..Default::default()
};
for (position, source) in info.sources.iter().enumerate() {
let key = crate::SourceKey {
priority: crate::Priority::new(position as f64).expect("enumerate index is finite"),
peer,
};
let body = serde_json::to_value(source).map_err(|error| ConversionError::SerializationError(error.to_string()))?;
entry.set_source(
key,
crate::SourceValue {
source: body,
timestamp: TimeStamp::ORIGIN,
},
);
if let Some(entry) = convert_resource_entry(resources, id, peer)? {
registry.resources.insert(id, entry);
}

registry.resources.insert(id, entry);
}
Ok(())
}

/// Snapshot one runtime resource into a storage entry, or `None` if the runtime has no such resource.
pub fn convert_resource_entry(resources: &graphene_resource::ResourceRegistry, id: ResourceId, peer: PeerId) -> Result<Option<crate::ResourceEntry>, ConversionError> {
let Some(info) = resources.info(&id) else { return Ok(None) };

let mut entry = crate::ResourceEntry {
hash: info.hash.copied(),
hash_timestamp: TimeStamp::ORIGIN,
..Default::default()
};
for (position, source) in info.sources.iter().enumerate() {
let key = crate::SourceKey {
priority: crate::Priority::new(position as f64).expect("enumerate index is finite"),
peer,
};
let body = serde_json::to_value(source).map_err(|error| ConversionError::SerializationError(error.to_string()))?;
entry.set_source(
key,
crate::SourceValue {
source: body,
timestamp: TimeStamp::ORIGIN,
},
);
}

Ok(Some(entry))
}

/// Collect the `ResourceId`s referenced by `TaggedValue::Resource` inputs anywhere in the network
/// (recursively through nested networks). These are the resources the document actually uses; the
/// runtime cache may hold more (history-retained orphans) that shouldn't be snapshotted into storage.
Expand Down Expand Up @@ -279,7 +286,7 @@ fn convert_network<M: NodeMetadataSource + ?Sized>(
metadata_path,
runtime_node_id: *runtime_node_id,
};
let mut node = convert_node(doc_node, location, registry, ctx)?;
let mut node = convert_node(doc_node, location, registry, ctx, true)?;
node.attributes.set(node::ORIGINAL_NODE_ID, serde_json::json!(runtime_node_id.0), TimeStamp::ORIGIN);
registry.node_instances.insert(global_id, node);
}
Expand Down Expand Up @@ -351,7 +358,13 @@ struct NodeLocation<'a> {
runtime_node_id: RuntimeNodeId,
}

fn convert_node<M: NodeMetadataSource + ?Sized>(doc_node: &DocumentNode, location: NodeLocation<'_>, registry: &mut Registry, ctx: &mut ConversionContext<'_, M>) -> Result<Node, ConversionError> {
fn convert_node<M: NodeMetadataSource + ?Sized>(
doc_node: &DocumentNode,
location: NodeLocation<'_>,
registry: &mut Registry,
ctx: &mut ConversionContext<'_, M>,
recurse: bool,
) -> Result<Node, ConversionError> {
let NodeLocation {
network_id,
parent_path,
Expand Down Expand Up @@ -383,7 +396,7 @@ fn convert_node<M: NodeMetadataSource + ?Sized>(doc_node: &DocumentNode, locatio
} else {
metadata_path
};
let implementation = convert_implementation(&doc_node.implementation, &node_path, child_metadata_path, registry, ctx)?;
let implementation = convert_implementation(&doc_node.implementation, &node_path, child_metadata_path, registry, ctx, recurse)?;

// Defaults match `DocumentNode::default()`; `to_runtime` rehydrates absent keys from the same defaults.
let mut attributes = crate::Attributes::new();
Expand All @@ -410,6 +423,29 @@ fn convert_node<M: NodeMetadataSource + ?Sized>(doc_node: &DocumentNode, locatio
})
}

/// Public form of the node ui-attribute encoding, for staging paths that encode single nodes.
pub fn encode_node_ui_attributes<M: NodeMetadataSource + ?Sized>(
attributes: &mut crate::Attributes,
metadata: &M,
metadata_path: &[RuntimeNodeId],
runtime_node_id: RuntimeNodeId,
timestamp: TimeStamp,
) -> Result<(), ConversionError> {
write_ui_attributes(attributes, metadata, metadata_path, runtime_node_id, timestamp)
}

/// Public form of the per-input ui-attribute encoding, matching `encode_node_ui_attributes`.
pub fn encode_input_ui_attributes<M: NodeMetadataSource + ?Sized>(
attributes: &mut crate::Attributes,
metadata: &M,
metadata_path: &[RuntimeNodeId],
runtime_node_id: RuntimeNodeId,
input_index: usize,
timestamp: TimeStamp,
) -> Result<(), ConversionError> {
write_ui_input_attributes(attributes, metadata, metadata_path, runtime_node_id, input_index, timestamp)
}

fn write_ui_attributes<M: NodeMetadataSource + ?Sized>(
attributes: &mut crate::Attributes,
metadata: &M,
Expand Down Expand Up @@ -533,6 +569,7 @@ fn convert_implementation<M: NodeMetadataSource + ?Sized>(
child_metadata_path: &[RuntimeNodeId],
registry: &mut Registry,
ctx: &mut ConversionContext<'_, M>,
recurse: bool,
) -> Result<Implementation, ConversionError> {
Ok(match implementation {
DocumentNodeImplementation::ProtoNode(identifier) => {
Expand Down Expand Up @@ -564,10 +601,158 @@ fn convert_implementation<M: NodeMetadataSource + ?Sized>(
// `to_runtime` -> `from_runtime` round trip reproduces the same `NetworkId` (and thus the
// same node-path hashes underneath it).
let nested_network_id = current_node_path.owned_network_id(ctx.peer);
convert_network(nested_network, nested_network_id, Some(current_node_path), child_metadata_path, registry, ctx)?;
if recurse {
convert_network(nested_network, nested_network_id, Some(current_node_path), child_metadata_path, registry, ctx)?;
}
Implementation::Network(nested_network_id)
}
// TODO: Support Extract in the Registry format.
DocumentNodeImplementation::Extract => return Err(ConversionError::UnsupportedImplementation),
})
}

/// Derives the stable storage IDs for entities addressed by runtime network paths, walking the
/// same path-hash chain as a full conversion so scoped and whole-document conversions agree.
pub struct PathResolver {
peer: PeerId,
}

impl PathResolver {
pub fn new(peer: PeerId) -> Self {
Self { peer }
}

/// The storage `NetworkId` of the network at `local_path` (`ROOT_NETWORK` for the empty path).
pub fn network_id(&self, local_path: &[RuntimeNodeId]) -> NetworkId {
match self.owner_path(local_path) {
None => ROOT_NETWORK,
Some(owner) => owner.owned_network_id(self.peer),
}
}

/// The storage `NodeId` of the node with `local_id` inside the network at `local_path`.
pub fn node_id(&self, local_path: &[RuntimeNodeId], local_id: RuntimeNodeId) -> NodeId {
let owner = self.owner_path(local_path);
child_path(owner.as_ref(), self.network_id(local_path), local_id).to_global_id(self.peer)
}

/// Converts one runtime input inside the network at `local_path` to its storage form, resolving
/// node references to their stable global IDs.
pub fn convert_input_at(&self, input: &GraphCraftNodeInput, local_path: &[RuntimeNodeId]) -> Result<NodeInput, ConversionError> {
let owner = self.owner_path(local_path);
convert_input(input, owner.as_ref(), self.network_id(local_path), self.peer)
}

/// The `NodePath` of the node owning the network at `local_path`, or `None` for the root network.
fn owner_path(&self, local_path: &[RuntimeNodeId]) -> Option<NodePath> {
let mut owner: Option<NodePath> = None;
for &local_id in local_path {
let network_id = match &owner {
None => ROOT_NETWORK,
Some(owner) => owner.owned_network_id(self.peer),
};
owner = Some(child_path(owner.as_ref(), network_id, local_id));
}
owner
}
}

/// Converts a chosen subset of runtime entities into a scratch [`Registry`] through the same
/// encoders as a whole-document conversion, so staging can derive deltas for just those entities.
pub struct ScopedConversion<'m, M: NodeMetadataSource + ?Sized> {
ctx: ConversionContext<'m, M>,
resolver: PathResolver,
}

impl<'m, M: NodeMetadataSource + ?Sized> ScopedConversion<'m, M> {
pub fn new(metadata: &'m M, peer: PeerId) -> Self {
Self {
ctx: ConversionContext {
declaration_ids: HashMap::new(),
declaration_bytes: HashMap::new(),
network_ids: HashMap::new(),
metadata,
peer,
},
resolver: PathResolver::new(peer),
}
}

/// Converts the node with `local_id` in the network at `local_path` into `registry` under its
/// stable IDs. `recurse` also converts the contents of any nested network the node implements;
/// without it only the node itself is converted, with its nested network referenced by ID.
pub fn convert_node_at(&mut self, registry: &mut Registry, local_path: &[RuntimeNodeId], local_id: RuntimeNodeId, doc_node: &DocumentNode, recurse: bool) -> Result<(), ConversionError> {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: Adding a nested-network node can leave a resource-valued network export unresolved because the scoped conversion exposes the export but the staging path only discovers resources from node inputs. Including resource references from the converted network export slots (or exposing an equivalent scoped helper) would keep structural additions self-contained.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At document/graph-storage/src/from_runtime.rs, line 684:

<comment>Adding a nested-network node can leave a resource-valued network export unresolved because the scoped conversion exposes the export but the staging path only discovers resources from node inputs. Including resource references from the converted network export slots (or exposing an equivalent scoped helper) would keep structural additions self-contained.</comment>

<file context>
@@ -564,10 +601,158 @@ fn convert_implementation<M: NodeMetadataSource + ?Sized>(
+	/// Converts the node with `local_id` in the network at `local_path` into `registry` under its
+	/// stable IDs. `recurse` also converts the contents of any nested network the node implements;
+	/// without it only the node itself is converted, with its nested network referenced by ID.
+	pub fn convert_node_at(&mut self, registry: &mut Registry, local_path: &[RuntimeNodeId], local_id: RuntimeNodeId, doc_node: &DocumentNode, recurse: bool) -> Result<(), ConversionError> {
+		let owner_path = self.resolver.owner_path(local_path);
+		let location = NodeLocation {
</file context>

let owner_path = self.resolver.owner_path(local_path);
let location = NodeLocation {
network_id: self.resolver.network_id(local_path),
parent_path: owner_path.as_ref(),
metadata_path: local_path,
runtime_node_id: local_id,
};

let mut node = convert_node(doc_node, location, registry, &mut self.ctx, recurse)?;
node.attributes.set(node::ORIGINAL_NODE_ID, serde_json::json!(local_id.0), TimeStamp::ORIGIN);
registry.node_instances.insert(self.resolver.node_id(local_path, local_id), node);
Ok(())
}

/// Converts the entry of the network at `local_path` (its export slots and network-level
/// attributes, not its nodes) into `registry`.
pub fn convert_network_entry_at(&mut self, registry: &mut Registry, local_path: &[RuntimeNodeId], node_network: &NodeNetwork) -> Result<(), ConversionError> {
let owner_path = self.resolver.owner_path(local_path);
let network_id = self.resolver.network_id(local_path);

let exports = node_network
.exports
.iter()
.map(|export| {
Ok(ExportSlot {
target: Some(convert_input(export, owner_path.as_ref(), network_id, self.ctx.peer)?),
timestamp: TimeStamp::ORIGIN,
})
})
.collect::<Result<Vec<_>, ConversionError>>()?;

let mut attributes = crate::Attributes::new();
write_ui_network_attributes(&mut attributes, self.ctx.metadata, local_path, TimeStamp::ORIGIN)?;
write_scope_injections(&mut attributes, node_network, owner_path.as_ref(), network_id, self.ctx.peer, TimeStamp::ORIGIN)?;

registry.networks.insert(network_id, Network { exports, attributes });
Ok(())
}

/// The extracted proto-node declaration bytes, for the caller to persist into its byte store.
pub fn finish(self) -> DeclarationBytes {
self.ctx.declaration_bytes
}
}

/// The `TaggedValue::Resource` IDs referenced by a stored node's value inputs.
///
/// Peeks at the externally-tagged serde shape (`{"Resource": <id>}`) rather than deserializing the
/// whole `TaggedValue`, which can be arbitrarily large. `resource_ref_shape_matches_serde` asserts
/// this stays in lockstep with the real serialization.
pub fn node_value_resource_refs(node: &Node) -> impl Iterator<Item = ResourceId> + '_ {
node.inputs.iter().filter_map(|slot| value_resource_ref(&slot.input))
}

/// The `TaggedValue::Resource` ID referenced by a stored value input, if any.
pub fn value_resource_ref(input: &NodeInput) -> Option<ResourceId> {
match input {
NodeInput::Value { value, .. } => value.get("Resource").and_then(|id| serde_json::from_value(id.clone()).ok()),
_ => None,
}
}

#[cfg(test)]
mod resource_ref_shape {
use super::*;

#[test]
fn resource_ref_shape_matches_serde() {
let id = ResourceId::from_hash(&ResourceHash::from(b"shape test".as_slice()));
let value = serde_json::to_value(TaggedValue::Resource(id)).expect("TaggedValue::Resource serializes");
let parsed: Option<ResourceId> = value.get("Resource").and_then(|inner| serde_json::from_value(inner.clone()).ok());
assert_eq!(parsed, Some(id), "The shape peek in node_value_resource_refs must match TaggedValue's serde form");
}
}
5 changes: 4 additions & 1 deletion document/graph-storage/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,10 @@ pub use resources::*;
pub use session::*;

#[cfg(any(feature = "conversion", test))]
pub use from_runtime::{RuntimeConversion, decode_declaration, encode_declaration};
pub use from_runtime::{
PathResolver, RuntimeConversion, ScopedConversion, convert_resource_entry, decode_declaration, encode_declaration, encode_input_ui_attributes, encode_node_ui_attributes, node_value_resource_refs,
value_resource_ref,
};
#[cfg(any(feature = "conversion", test))]
pub use metadata_source::{InputMetadataEntry, NetworkMetadataEntry, NoMetadata, NodeMetadataEntry, NodeMetadataSource, Position};
#[cfg(any(feature = "conversion", test))]
Expand Down
6 changes: 6 additions & 0 deletions document/graph-storage/src/session.rs
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,12 @@ impl Session {
Ok(revs)
}

/// Stages ops computed outside the session (an incremental runtime projection) as hot ops, the
/// same way `stage_from_runtime` stages a diff's ops.
pub fn stage_computed_ops(&mut self, ops: Vec<RegistryDelta>) -> Result<Vec<HotOp>, CrdtError> {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P3: stage_computed_ops takes Vec<RegistryDelta> while stage_ops (the method it delegates to) accepts impl IntoIterator<Item = RegistryDelta>. Matching the more flexible signature would let callers pass arrays, slices, iterator adapters, or other collections without an intermediate allocation.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At document/graph-storage/src/session.rs, line 134:

<comment>`stage_computed_ops` takes `Vec<RegistryDelta>` while `stage_ops` (the method it delegates to) accepts `impl IntoIterator<Item = RegistryDelta>`. Matching the more flexible signature would let callers pass arrays, slices, iterator adapters, or other collections without an intermediate allocation.</comment>

<file context>
@@ -129,6 +129,12 @@ impl Session {
 
+	/// Stages ops computed outside the session (an incremental runtime projection) as hot ops, the
+	/// same way `stage_from_runtime` stages a diff's ops.
+	pub fn stage_computed_ops(&mut self, ops: Vec<RegistryDelta>) -> Result<Vec<HotOp>, CrdtError> {
+		self.stage_ops(ops)
+	}
</file context>

self.stage_ops(ops)
}

/// Apply each op as a hot op with a freshly-ticked timestamp, returning the staged frames in
/// order. Each tick is strictly later than the last, so the final frame carries the latest
/// timestamp, which is what the caller passes to `retire`.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,9 @@ mod caches;
#[cfg(test)]
mod characterization_tests;
mod deserialization;
pub mod editor_delta;
#[cfg(test)]
mod editor_delta_tests;
mod hit_tests;
mod layout;
mod memo_network;
Expand Down
Loading
Loading