Skip to content
Closed
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> {
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> {
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