Skip to content
Merged
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
96 changes: 96 additions & 0 deletions crates/graphforge-storage/src/graph_construction.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8210,6 +8210,15 @@ mod tests {
"uuid_encode.after_delta_runs",
"uuid_encode.after_manifest",
"uuid_encode.after_intent_removal",
"v4_publish.after_artifacts",
"v4_publish.after_artifacts_fsync",
"v4_publish.after_receipt_temp_fsync",
"v4_publish.after_receipt_install",
"v4_publish.after_manifest_temp_fsync",
"v4_publish.after_manifest_install",
"v4_publish.after_lock_temp_fsync",
"v4_publish.after_lock_install",
"encode.after_v4_before_inventory",
"encode.control.after_temp_fsync.inventory.json",
"encode.control.after_install.inventory.json",
] {
Expand Down Expand Up @@ -8444,6 +8453,38 @@ mod tests {
assert_eq!(encoding.evidence.prior_topology_rows_decoded, 0);
assert_eq!(encoding.evidence.retained_topology_bytes_copied, 0);
assert_eq!(encoding.evidence.membership_records, 5);
assert_eq!(encoding.evidence.ordinal_records, 3);
assert_eq!(encoding.evidence.ordinal_artifact_write_bytes, 120);
assert_eq!(encoding.evidence.ordinal_artifact_write_operations, 2);
assert_eq!(encoding.evidence.ordinal_ranges, 1);
assert_eq!(encoding.evidence.ordinal_work_operations, 3);
assert_eq!(encoding.evidence.ordinal_peak_buffer_bytes, 3 * 64 * 1024);
assert_eq!(encoding.evidence.ordinal_publication_write_operations, 3);
// Three artifact file syncs, one artifact-directory barrier, two
// barriers for each of receipt/manifest/lock, and four ancestor
// directory barriers after the complete facet is installed.
assert_eq!(encoding.evidence.ordinal_fsync_operations, 14);
let ordinal_publication_bytes = encoding
.artifacts
.iter()
.filter(|artifact| {
artifact.path.ends_with("ordinal-v4-receipt.json")
|| artifact.path.ends_with("ordinal-v4-manifest.json")
|| artifact.path.ends_with("ordinal-v4.lock")
})
.map(|artifact| artifact.bytes)
.sum::<u64>();
assert_eq!(
encoding.evidence.ordinal_publication_write_bytes,
ordinal_publication_bytes
);
assert_eq!(
encoding.evidence.ordinal_peak_temporary_bytes,
encoding
.evidence
.ordinal_artifact_write_bytes
.saturating_add(ordinal_publication_bytes)
);

let graph = root
.path()
Expand Down Expand Up @@ -8488,6 +8529,61 @@ mod tests {
assert_eq!(resumed, encoding);
}

#[test]
fn fresh_construction_publishes_selected_authenticated_v4_exact_lookup() {
let root = TempDir::new().unwrap();
let operation = Uuid::from_u128(9_321);
let target = Uuid::from_u128(9_322);
crate::open_or_initialize_project(root.path()).unwrap();
let mut session = GraphConstructionSession::open(
root.path(),
operation,
0,
GraphConstructionBudgets::default(),
)
.unwrap();
session
.append(ConstructionChunkKind::Node, "nodes", &node_batch(1, 3))
.unwrap();
session.seal().unwrap();
let shape = session.shape_canonical_with_cancellation(|| false).unwrap();
let encoding = session.encode_canonical(&shape, 1).unwrap();
session
.publish_canonical(&encoding, target, Uuid::from_u128(9_323))
.unwrap();

let selected = crate::resolve_project_generation(root.path()).unwrap();
assert_eq!(selected.generation_uuid(), target);
let authority = selected
.authenticated_v4_ordinal_authority()
.unwrap()
.expect("fresh construction publishes v4 authority");
let inventory = selected.graph_files_inventory().unwrap().unwrap();
let materialized = TempDir::new().unwrap();
let graph = materialized.path().join("graph");
std::fs::create_dir(&graph).unwrap();
crate::materialize_graph_objects(root.path(), &inventory, &graph).unwrap();
let mut handle = match authority
.open(&graph, crate::V4OrdinalIdentityLimits::default())
.unwrap()
{
crate::V4OrdinalIdentityOpen::Ready(handle) => handle,
crate::V4OrdinalIdentityOpen::RebuildRequired { found_version } => {
panic!("fresh v4 unexpectedly requires rebuild from {found_version}")
}
};
let lookup = handle.lookup_node_uuids(&[3, 1, 4, 2]).unwrap();
assert_eq!(
lookup.values,
vec![
Some(Uuid::from_u128(3)),
Some(Uuid::from_u128(1)),
None,
Some(Uuid::from_u128(2)),
]
);
}

#[test]
fn canonical_encoder_merges_bounded_heterogeneous_node_schemas() {
let root = TempDir::new().unwrap();
Expand Down
146 changes: 127 additions & 19 deletions crates/graphforge-storage/src/graph_construction_encoding.rs
Original file line number Diff line number Diff line change
Expand Up @@ -306,6 +306,36 @@ pub struct GraphConstructionEncodingEvidence {
pub membership_peak_buffer_bytes: u64,
/// Peak bytes in owned temporary v3 outputs.
pub membership_peak_temporary_bytes: u64,
/// Canonical node identities streamed into the v4 ordinal index.
#[serde(default)]
pub ordinal_records: u64,
/// Immutable v4 payload bytes written before publication metadata.
#[serde(default)]
pub ordinal_artifact_write_bytes: u64,
/// Bounded payload writes performed by the v4 stream encoder.
#[serde(default)]
pub ordinal_artifact_write_operations: u64,
/// Contiguous ordinal ranges emitted by the v4 stream encoder.
#[serde(default)]
pub ordinal_ranges: u64,
/// Cancellation/work polls performed while streaming v4 identities.
#[serde(default)]
pub ordinal_work_operations: u64,
/// Peak live v4 encoding buffer bytes.
#[serde(default)]
pub ordinal_peak_buffer_bytes: u64,
/// Peak bytes in owned temporary v4 outputs.
#[serde(default)]
pub ordinal_peak_temporary_bytes: u64,
/// v4 payload and publication durability barriers.
#[serde(default)]
pub ordinal_fsync_operations: u64,
/// v4 receipt, manifest, and authority-lock bytes written at publication.
#[serde(default)]
pub ordinal_publication_write_bytes: u64,
/// v4 receipt, manifest, and authority-lock writes at publication.
#[serde(default)]
pub ordinal_publication_write_operations: u64,
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
Expand Down Expand Up @@ -466,7 +496,21 @@ pub(crate) fn encode(
.map(ConstructionSemanticAuthority::context)
.transpose()?;

encode_nodes(
let identities_sha256 = shaped_output_sha256(shape_outputs, &shape.identities)?;
let mut index = crate::uuid_membership::encode_construction_index(
source,
&shape.identities,
identities_sha256,
&output,
generation,
shape.parent_topology_generation,
parent_index,
shape.node_count,
shape.edge_count,
cancelled,
)?;

let v4 = encode_nodes(
source,
&output,
shape,
Expand All @@ -476,10 +520,41 @@ pub(crate) fn encode(
semantic_context.as_ref(),
semantic_authority.map(|authority| &authority.bindings),
budgets,
generation,
shape.parent_topology_generation == 0,
cancelled,
&mut artifacts,
&mut evidence,
)?;
if let Some((manifest, metrics)) = v4 {
if metrics.input_records != shape.node_count {
return Err(storage("v4 construction count differs from shaped nodes"));
}
evidence.ordinal_records = metrics.input_records;
evidence.ordinal_artifact_write_bytes = metrics.artifact_bytes;
evidence.ordinal_artifact_write_operations = metrics.write_blocks;
evidence.ordinal_ranges = u64::try_from(metrics.ranges).map_err(storage)?;
evidence.ordinal_work_operations = metrics.cancellation_polls;
evidence.ordinal_peak_buffer_bytes =
u64::try_from(metrics.peak_buffer_bytes).map_err(storage)?;
let (v4_artifacts, publication) =
crate::uuid_membership::publish_v4_construction_artifacts(
&output,
&manifest,
generation,
identities_sha256,
)?;
evidence.ordinal_publication_write_bytes = publication.write_bytes;
evidence.ordinal_publication_write_operations = publication.write_operations;
evidence.ordinal_fsync_operations = metrics
.fsync_operations
.saturating_add(publication.fsync_operations);
evidence.ordinal_peak_temporary_bytes = metrics
.peak_temporary_bytes
.max(publication.peak_temporary_bytes);
crate::graph_construction::construction_failpoint("encode.after_v4_before_inventory");
index.artifacts.extend(v4_artifacts);
}
encode_edges(
source,
&output,
Expand Down Expand Up @@ -511,18 +586,6 @@ pub(crate) fn encode(
&mut evidence,
)?;

let index = crate::uuid_membership::encode_construction_index(
source,
&shape.identities,
shaped_output_sha256(shape_outputs, &shape.identities)?,
&output,
generation,
shape.parent_topology_generation,
parent_index,
shape.node_count,
shape.edge_count,
cancelled,
)?;
evidence.membership_records = index.input_records;
evidence.membership_read_bytes = index.read_bytes;
evidence.membership_write_bytes = index.final_write_bytes;
Expand Down Expand Up @@ -699,19 +762,60 @@ fn encode_nodes(
semantic_context: Option<&CompositionBindingContext>,
semantic_bindings: Option<&SemanticStorageBindings>,
budgets: GraphConstructionBudgets,
generation: u64,
build_v4: bool,
cancelled: &mut impl FnMut() -> bool,
artifacts: &mut Vec<ConstructionEncodedArtifact>,
evidence: &mut GraphConstructionEncodingEvidence,
) -> Result<(), GfError> {
) -> Result<
Option<(
crate::V4OrdinalIdentityManifest,
crate::uuid_membership::V4OrdinalBuildMetrics,
)>,
GfError,
> {
if shape.node_rows.is_empty() {
return Ok(());
if !build_v4 {
return Ok(None);
}
let graph = output
.open_child_directory(OsStr::new("graph"))
.map_err(storage)?;
let topology = graph
.open_child_directory(OsStr::new("topology"))
.map_err(storage)?;
let index = topology
.open_child_directory(OsStr::new("uuid-membership"))
.map_err(storage)?;
return crate::uuid_membership::V4OrdinalConstructionWriter::start(generation, &index)?
.finish()
.map(Some);
}
let details_name = shape
.node_details
.as_deref()
.ok_or_else(|| storage("node rows lack canonical details"))?;
let mut identities =
FixedReader::<IDENTITY_WIDTH>::open(source, shape_outputs, &shape.identities)?;
let v4_index = if build_v4 {
let graph = output
.open_child_directory(OsStr::new("graph"))
.map_err(storage)?;
let topology = graph
.open_child_directory(OsStr::new("topology"))
.map_err(storage)?;
Some(
topology
.open_child_directory(OsStr::new("uuid-membership"))
.map_err(storage)?,
)
} else {
None
};
let mut v4 = v4_index
.as_ref()
.map(|index| crate::uuid_membership::V4OrdinalConstructionWriter::start(generation, index))
.transpose()?;
let mut details = FixedReader::<NODE_DETAIL_WIDTH>::open(source, shape_outputs, details_name)?;
let rows_per_window = budgets
.max_batch_rows
Expand Down Expand Up @@ -761,10 +865,12 @@ fn encode_nodes(
.get(label)
.ok_or_else(|| storage("node label is absent from runtime catalog"))?,
};
let node_id = u64::from_be_bytes(identity[24..32].try_into().expect("fixed"));
if let Some(v4) = v4.as_mut() {
v4.push_pair(Uuid::from_bytes(uuid), node_id, cancelled)?;
}
out_uuid.push(uuid);
out_id.push(u64::from_be_bytes(
identity[24..32].try_into().expect("fixed"),
));
out_id.push(node_id);
out_type.push(type_id);
if out_uuid.len().is_multiple_of(4096) && cancelled() {
return Err(storage("construction encoding cancelled"));
Expand Down Expand Up @@ -802,7 +908,9 @@ fn encode_nodes(
cancelled,
artifacts,
evidence,
)
)?;
v4.map(crate::uuid_membership::V4OrdinalConstructionWriter::finish)
.transpose()
}

#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
Expand Down
10 changes: 5 additions & 5 deletions crates/graphforge-storage/src/graph_files.rs
Original file line number Diff line number Diff line change
Expand Up @@ -990,13 +990,13 @@ pub(crate) fn resolve_v1_inventory_entry(
resolve_v1_inventory_entry_retained(graph_root, entry).map(|resolved| resolved.path)
}

struct RetainedV1InventoryEntry {
path: PathBuf,
file: File,
identity: graphforge_filesystem::FileIdentity,
pub(crate) struct RetainedV1InventoryEntry {
pub(crate) path: PathBuf,
pub(crate) file: File,
pub(crate) identity: graphforge_filesystem::FileIdentity,
}

fn resolve_v1_inventory_entry_retained(
pub(crate) fn resolve_v1_inventory_entry_retained(
graph_root: &Path,
entry: &GraphFileEntry,
) -> Result<RetainedV1InventoryEntry, GfError> {
Expand Down
Loading