From 955ffe8ca3d33650630e27cbddb8fd35f0953a60 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 08:27:21 -0600 Subject: [PATCH 1/7] feat(catalog): identify derived summary inputs --- crates/asap_types/src/aggregation_config.rs | 48 ++- crates/asap_types/src/derived_input.rs | 322 ++++++++++++++++++ crates/asap_types/src/lib.rs | 1 + crates/asap_types/src/policy_fingerprint.rs | 9 +- crates/asap_types/src/precompute_plan.rs | 11 + .../asap_types/src/precompute_plan/catalog.rs | 11 +- crates/asap_types/src/sds.rs | 16 +- crates/asap_types/src/summary_catalog.rs | 57 +++- .../drivers/ingest/prometheus_remote_write.rs | 2 + data_plane/src/drivers/query/servers/http.rs | 1 + .../src/precompute_engine/output_sink.rs | 1 + .../sketch_db/lifecycle/eviction.rs | 1 + .../storage_engines/types/streaming_config.rs | 16 + .../tests/test_utilities/engine_factories.rs | 8 + .../summary-catalog-sds-architecture.md | 21 ++ 15 files changed, 503 insertions(+), 22 deletions(-) create mode 100644 crates/asap_types/src/derived_input.rs diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index 3e9f5cab3..8254b65b1 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -103,6 +103,8 @@ pub struct PrecomputeMaterialization { pub grouping_labels: crate::GroupingProjection, #[serde(default, skip_serializing_if = "Option::is_none")] pub partitioning: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub derived_input: Option, pub aggregated_labels: KeyByLabelNames, pub rollup_labels: KeyByLabelNames, pub original_yaml: String, @@ -200,9 +202,39 @@ impl PrecomputeMaterialization { .saturating_mul(1_000) } + pub fn source_identity(&self) -> crate::sds::DataSourceIdentity { + use crate::sds::DataSourceIdentity; + if let Some(input) = &self.derived_input { + DataSourceIdentity::Derived { + input: input.clone(), + } + } else if let Some(table_ref) = &self.table_name { + DataSourceIdentity::Table { + table_ref: table_ref.clone(), + } + } else { + DataSourceIdentity::TimeSeries { + metric: self.metric.clone(), + } + } + } + pub fn population_filter_canonical(&self) -> Result { + if let Some(input) = &self.derived_input { + input.validate()?; + if self.table_name.is_some() + || self.table_population.is_some() + || self.table_timestamp_column.is_some() + || !self.spatial_filter.is_empty() + { + return Err("derived inputs cannot also declare a raw source/filter".into()); + } + } self.effective_value_projection().validate()?; - if self.value_projection.is_some() && self.table_name.is_none() { + if self.value_projection.is_some() + && self.table_name.is_none() + && self.derived_input.is_none() + { return Err("explicit table value projection requires a table source".into()); } if let Some(column) = &self.table_timestamp_column { @@ -255,6 +287,7 @@ impl PrecomputeMaterialization { parameters, grouping_labels: grouping_labels.into(), partitioning: None, + derived_input: None, aggregated_labels, rollup_labels, original_yaml, @@ -400,6 +433,11 @@ impl PrecomputeMaterialization { table_name, value_column, ); + config.derived_input = data + .get("derived_input") + .filter(|v| !v.is_null()) + .map(|v| serde_json::from_value(v.clone())) + .transpose()?; config.partitioning = data .get("partitioning") .filter(|value| !value.is_null()) @@ -595,6 +633,11 @@ impl PrecomputeMaterialization { table_name, value_column, ); + config.derived_input = aggregation_data + .get("derived_input") + .filter(|v| !v.is_null()) + .map(|v| serde_yaml::from_value(v.clone())) + .transpose()?; config.partitioning = aggregation_data .get("partitioning") .filter(|value| !value.is_null()) @@ -640,6 +683,9 @@ impl SerializableToSink for PrecomputeMaterialization { "metric": self.metric, }); + if let Some(input) = &self.derived_input { + json["derived_input"] = serde_json::json!(input); + } // Only include numAggregatesToRetain if it's Some if let Some(num_aggregates) = self.num_aggregates_to_retain { json["numAggregatesToRetain"] = serde_json::json!(num_aggregates); diff --git a/crates/asap_types/src/derived_input.rs b/crates/asap_types/src/derived_input.rs new file mode 100644 index 000000000..5aa86fbb2 --- /dev/null +++ b/crates/asap_types/src/derived_input.rs @@ -0,0 +1,322 @@ +//! Content identity for a maintenance sub-DAG cut at existing summary inputs. +//! The executable program remains in OwnedPostAsapDag; this is its catalog key. +use std::collections::{BTreeMap, BTreeSet}; + +use planner_types::post_asap::PostAsapNodeId; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::{executable_plan::OwnedPostAsapDag, sds::SummaryDefinitionId}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DerivedInputIdentity { + pub inputs: BTreeSet, + pub program_sha256: String, +} + +impl DerivedInputIdentity { + pub fn validate(&self) -> Result<(), String> { + if self.inputs.is_empty() + || self.inputs.iter().any(|id| id.fingerprint().is_unset()) + || self.program_sha256.len() != 64 + || !self + .program_sha256 + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + { + return Err("derived input requires summary references and a canonical SHA-256".into()); + } + Ok(()) + } + + /// Hash semantic nodes and edge roles, replacing input frontiers with stable + /// catalog IDs. Query IDs, node numbering, and catalog generations are absent. + pub fn from_dag( + document: &OwnedPostAsapDag, + root: PostAsapNodeId, + frontiers: &BTreeMap, + ) -> Result { + if document.schema_version != crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION { + return Err("unsupported derived program document version".into()); + } + document.decode()?; + let mut incoming: BTreeMap<_, Vec<_>> = BTreeMap::new(); + for edge in &document.edges { + incoming.entry(edge.consumer).or_default().push(edge); + } + let nodes: BTreeMap<_, _> = document.nodes.iter().map(|n| (n.id, n)).collect(); + if frontiers.contains_key(&root) { + return Err("derived program requires a non-frontier root".into()); + } + let mut hashes = BTreeMap::new(); + let mut visiting = BTreeSet::new(); + let mut inputs = BTreeSet::new(); + let mut stack = vec![(root, false)]; + while let Some((id, finish)) = stack.pop() { + if hashes.contains_key(&id) { + continue; + } + let node = nodes + .get(&id) + .ok_or("derived program references missing node")?; + if let Some(summary) = frontiers.get(&id) { + inputs.insert(*summary); + hashes.insert(id, serde_json::json!({"summary": summary})); + continue; + } + if !finish { + if !visiting.insert(id) { + return Err("derived program has a cycle".into()); + } + stack.push((id, true)); + let edges = incoming + .get(&id) + .ok_or("derived program has an unbound input leaf")?; + for edge in edges { + stack.push((edge.producer, false)); + } + continue; + } + let mut edges = Vec::new(); + for edge in incoming.get(&id).into_iter().flatten() { + let input = hashes + .get(&edge.producer) + .ok_or("derived input was not evaluated")?; + edges.push( + serde_json::to_vec(&serde_json::json!({ + "input": input, "role": edge.role, "schema": edge.intermediate_schema, + "state": edge.data_state, "grouping": edge.grouping, "window": edge.window, + })) + .map_err(|e| e.to_string())?, + ); + } + edges.sort(); + let bytes = serde_json::to_vec(&serde_json::json!({ + "version": 1, "operator": node.operator, "payload": node.payload, + "state": node.output_state, "schema": node.output_schema, + "guarantee": node.guarantee, "inputs": edges, + })) + .map_err(|e| e.to_string())?; + hashes.insert( + id, + serde_json::json!(format!("{:x}", Sha256::digest(bytes))), + ); + visiting.remove(&id); + } + let identity = Self { + inputs, + program_sha256: hashes[&root] + .as_str() + .ok_or("derived root has no semantic hash")? + .into(), + }; + identity.validate()?; + Ok(identity) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::summary_catalog::SummaryCatalog; + use crate::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; + + fn config() -> PrecomputeMaterialization { + PrecomputeMaterialization::new( + AggregationType::Sum, + String::new(), + Default::default(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 10, + 10, + WindowKind::Tumbling, + String::new(), + "m".into(), + None, + None, + None, + ) + } + + #[test] + fn derived_source_is_distinct_and_generation_independent() { + let raw = config(); + let raw_id = SummaryDefinitionId::from(raw.policy_fingerprint()); + let mut derived = raw.clone(); + derived.derived_input = Some(DerivedInputIdentity { + inputs: BTreeSet::from([raw_id]), + program_sha256: "a".repeat(64), + }); + assert_ne!(raw.policy_fingerprint(), derived.policy_fingerprint()); + let a = + SummaryCatalog::from_materializations(1, 1, &[raw.clone(), derived.clone()]).unwrap(); + let b = SummaryCatalog::from_materializations(2, 9, &[raw, derived.clone()]).unwrap(); + assert_eq!(a.materializations, b.materializations); + assert_eq!(a.data_descriptors, b.data_descriptors); + let mut renamed = derived.clone(); + renamed.metric = "output_alias".into(); + assert_eq!(renamed.policy_fingerprint(), derived.policy_fingerprint()); + let json = serde_json::to_value(&derived).unwrap(); + let decoded: PrecomputeMaterialization = serde_json::from_value(json).unwrap(); + assert_eq!(decoded.policy_fingerprint(), derived.policy_fingerprint()); + } + + #[test] + fn catalog_rejects_missing_derived_dependencies_and_raw_source_conflicts() { + let mut derived = config(); + derived.derived_input = Some(DerivedInputIdentity { + inputs: BTreeSet::from([SummaryDefinitionId::from(derived.policy_fingerprint())]), + program_sha256: "b".repeat(64), + }); + assert!(SummaryCatalog::from_materializations(1, 1, &[derived.clone()]).is_err()); + derived.table_name = Some("table".into()); + assert!(derived.population_filter_canonical().is_err()); + } + fn program(source: u32, root: u32) -> OwnedPostAsapDag { + use crate::executable_plan::{OwnedPostAsapEdge, OwnedPostAsapNode}; + use planner_types::post_asap::{ + EdgeRole, ExecutableOperator, ExecutionDataState, GroupingEdgeCompatibility, + WindowEdgeCompatibility, + }; + let state = ExecutionDataState::MAINTENANCE_SUMMARY; + OwnedPostAsapDag { + schema_version: 1, + query_id: "query-a".into(), + root: PostAsapNodeId(root), + nodes: [source, root] + .into_iter() + .map(|id| OwnedPostAsapNode { + id: PostAsapNodeId(id), + operator: ExecutableOperator::SummaryMerge, + payload: serde_json::json!({"kind":"summary_merge"}), + output_state: state, + output_schema: serde_json::json!({"fields":[],"time_index":null}), + guarantee: None, + }) + .collect(), + edges: vec![OwnedPostAsapEdge { + producer: PostAsapNodeId(source), + consumer: PostAsapNodeId(root), + role: EdgeRole::Input, + intermediate_schema: serde_json::json!({"fields":[],"time_index":null}), + data_state: state, + grouping: GroupingEdgeCompatibility::Identical, + window: WindowEdgeCompatibility::NotApplicable, + }], + } + } + + #[test] + fn semantic_signature_ignores_node_and_query_numbering_but_not_inputs() { + let source = SummaryDefinitionId::from(config().policy_fingerprint()); + let a = program(1, 2); + let first = DerivedInputIdentity::from_dag( + &a, + a.root, + &BTreeMap::from([(PostAsapNodeId(1), source)]), + ) + .unwrap(); + let mut b = program(900, 42); + b.query_id = "another-query".into(); + b.nodes.reverse(); + let second = DerivedInputIdentity::from_dag( + &b, + b.root, + &BTreeMap::from([(PostAsapNodeId(900), source)]), + ) + .unwrap(); + assert_eq!(first, second); + b.nodes + .iter_mut() + .find(|n| n.id == b.root) + .unwrap() + .output_schema = serde_json::json!({"fields":[],"time_index":0}); + assert_ne!( + first, + DerivedInputIdentity::from_dag( + &b, + b.root, + &BTreeMap::from([(PostAsapNodeId(900), source)]) + ) + .unwrap() + ); + assert!(DerivedInputIdentity::from_dag(&a, a.root, &BTreeMap::new()).is_err()); + let mut unsupported = a.clone(); + unsupported.schema_version = 999; + assert!(DerivedInputIdentity::from_dag( + &unsupported, + unsupported.root, + &BTreeMap::from([(PostAsapNodeId(1), source)]) + ) + .is_err()); + unsupported = a.clone(); + unsupported.nodes[1].payload = serde_json::json!({"kind":"unknown_operator"}); + assert!(DerivedInputIdentity::from_dag( + &unsupported, + unsupported.root, + &BTreeMap::from([(PostAsapNodeId(1), source)]) + ) + .is_err()); + let mut cycle = a.clone(); + cycle.edges[0].producer = cycle.root; + assert!(DerivedInputIdentity::from_dag(&cycle, cycle.root, &BTreeMap::new()).is_err()); + cycle.edges[0].producer = PostAsapNodeId(999); + assert!(DerivedInputIdentity::from_dag(&cycle, cycle.root, &BTreeMap::new()).is_err()); + } + + #[test] + fn catalog_rejects_self_referential_summary() { + use crate::sds::{ + DataDescriptor, DataSourceIdentity, SummaryDescriptor, ValueProjectionIdentity, + }; + let config = config(); + let input = DerivedInputIdentity { + inputs: BTreeSet::from([SummaryDefinitionId::from(config.policy_fingerprint())]), + program_sha256: "f".repeat(64), + }; + let data = DataDescriptor::new_typed( + DataSourceIdentity::Derived { input }, + ValueProjectionIdentity::SampleValue, + "", + Vec::::new(), + "derived", + ); + assert!(SummaryCatalog::build( + 1, + 1, + [( + config.policy_fingerprint(), + SummaryDescriptor::from_config(&config).unwrap(), + data, + config.window_layout + )] + ) + .is_err()); + } + #[test] + fn installation_rejects_derived_inputs_until_immutable_consumer_is_enabled() { + use crate::precompute_plan::{PlanEnvelope, PrecomputePlan}; + let mut config = config(); + config.derived_input = Some(DerivedInputIdentity { + inputs: BTreeSet::from([SummaryDefinitionId::from(config.policy_fingerprint())]), + program_sha256: "c".repeat(64), + }); + let envelope = PlanEnvelope { + plan_id: 1, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 0, + expiry_unix_ms: None, + backend_compat: "asap-query-backend.v1".into(), + planner_revision: "test".into(), + capability_snapshot_id: "test".into(), + }; + let error = + PrecomputePlan::build(envelope, vec![config], &["producer".into()]).unwrap_err(); + assert!(error.to_string().contains("immutable maintenance consumer")); + } +} diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 3c0a23997..0832dfb12 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -2,6 +2,7 @@ pub mod accumulator_spec; pub mod accuracy; pub mod aggregation_config; pub mod aggregation_type; +pub mod derived_input; pub mod enums; pub mod executable_plan; pub mod grouping_projection; diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index f9e751651..46837a95e 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -90,7 +90,14 @@ impl PolicyFingerprint { let mut buf: Vec = Vec::with_capacity(512); // 1. metric name - buf.extend_from_slice(cfg.metric.as_bytes()); + if cfg.derived_input.is_some() { + buf.extend_from_slice(b"derived-input-v1:"); + buf.extend_from_slice( + &serde_json::to_vec(&cfg.source_identity()).expect("typed source identity"), + ); + } else { + buf.extend_from_slice(cfg.metric.as_bytes()); + } buf.push(0); // 2. aggregation_type (Serialize impl is the stable form) diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index 257d99dd4..da7b07b0c 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -334,6 +334,17 @@ impl PrecomputePlan { if !valid_ingest { return Err(PrecomputePlanError::UnsupportedIngestEndpoint); } + // Derived input identity is portable, but raw ingress cannot execute it. + // Installation stays closed until the immutable maintenance consumer is wired. + if self + .materializations + .iter() + .any(|config| config.derived_input.is_some()) + { + return Err(PrecomputePlanError::CatalogContract( + "derived summary input requires an immutable maintenance consumer".into(), + )); + } for (query_id, installed) in &self.executable_dags { if query_id != &installed.document.query_id { return Err(PrecomputePlanError::CatalogContract( diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs index 9ee907798..e9dd2a599 100644 --- a/crates/asap_types/src/precompute_plan/catalog.rs +++ b/crates/asap_types/src/precompute_plan/catalog.rs @@ -1,6 +1,6 @@ //! Catalog consistency checks for the precompute execution plan. use super::*; -use crate::sds::{DataSourceIdentity, SummaryDefinitionId, SummaryDescriptor}; +use crate::sds::{SummaryDefinitionId, SummaryDescriptor}; use crate::summary_catalog::SummaryCatalog; use planner_types::pre_asap::Source; use std::collections::BTreeSet; @@ -70,14 +70,7 @@ impl PrecomputePlan { return Err(invalid("pane origin differs from catalog definition")); } let data = &catalog.data_descriptors[&binding.data_descriptor_id]; - let expected_source = config.table_name.as_ref().map_or_else( - || DataSourceIdentity::TimeSeries { - metric: config.metric.clone(), - }, - |table_ref| DataSourceIdentity::Table { - table_ref: table_ref.clone(), - }, - ); + let expected_source = config.source_identity(); if config.table_name.is_some() && !config.grouping_labels.is_empty() && data.observation_semantics diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index 23c3cc48b..b75fd88ca 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -729,8 +729,15 @@ impl FidelityGuarantee { #[serde(deny_unknown_fields)] #[serde(rename_all = "snake_case")] pub enum DataSourceIdentity { - TimeSeries { metric: String }, - Table { table_ref: String }, + TimeSeries { + metric: String, + }, + Table { + table_ref: String, + }, + Derived { + input: crate::derived_input::DerivedInputIdentity, + }, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -829,7 +836,7 @@ impl DataDescriptor { pub fn time_series_metric(&self) -> Option<&str> { match &self.source { DataSourceIdentity::TimeSeries { metric } => Some(metric), - DataSourceIdentity::Table { .. } => None, + DataSourceIdentity::Table { .. } | DataSourceIdentity::Derived { .. } => None, } } @@ -929,6 +936,9 @@ impl DataDescriptor { &self.id } pub fn validate(&self) -> Result<(), SdsError> { + if let DataSourceIdentity::Derived { input } = &self.source { + input.validate().map_err(SdsError)?; + } self.value_projection.validate().map_err(SdsError)?; self.group_by_keys.validate().map_err(SdsError)?; if matches!(self.source, DataSourceIdentity::TimeSeries { .. }) diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 5f31ee43d..36468724d 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -103,14 +103,7 @@ impl SummaryCatalog { .map(|config| { let summary = SummaryDescriptor::from_config(config) .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; - let source = config.table_name.as_ref().map_or_else( - || DataSourceIdentity::TimeSeries { - metric: config.metric.clone(), - }, - |table_ref| DataSourceIdentity::Table { - table_ref: table_ref.clone(), - }, - ); + let source = config.source_identity(); let value_projection = config.effective_value_projection().clone(); let data = DataDescriptor::new_typed( source, @@ -253,6 +246,54 @@ impl SummaryCatalog { return Err(SummaryCatalogError::MissingDescriptor(id.as_u64())); } } + if !self + .data_descriptors + .values() + .any(|data| matches!(data.source, DataSourceIdentity::Derived { .. })) + { + return Ok(()); + } + // Dependencies must refer to this snapshot and form an acyclic graph. + let mut pending = std::collections::BTreeMap::new(); + let mut consumers: std::collections::BTreeMap<_, Vec<_>> = + std::collections::BTreeMap::new(); + for (id, binding) in &self.materializations { + let dependencies = match &self.data_descriptors[&binding.data_descriptor_id].source { + DataSourceIdentity::Derived { input } => input.inputs.clone(), + _ => Default::default(), + }; + for source in &dependencies { + if !self.materializations.contains_key(source) { + return Err(SummaryCatalogError::Descriptor( + "derived input references missing summary".into(), + )); + } + consumers.entry(*source).or_default().push(*id); + } + pending.insert(*id, dependencies.len()); + } + let mut ready: Vec<_> = pending + .iter() + .filter_map(|(id, count)| (*count == 0).then_some(*id)) + .collect(); + let mut visited = 0; + while let Some(id) = ready.pop() { + visited += 1; + for consumer in consumers.get(&id).into_iter().flatten() { + let count = pending + .get_mut(consumer) + .expect("catalog dependency target"); + *count -= 1; + if *count == 0 { + ready.push(*consumer); + } + } + } + if visited != pending.len() { + return Err(SummaryCatalogError::Descriptor( + "derived summary dependencies have a cycle".into(), + )); + } Ok(()) } } diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index c1ec49919..d0aea8b9e 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -988,6 +988,7 @@ mod tests { table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -1059,6 +1060,7 @@ mod tests { table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index c456d8e3c..af0d3a07a 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -3681,6 +3681,7 @@ aggregations: table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index b908f2a8b..46a83a42c 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -422,6 +422,7 @@ mod tests { table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, } diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs index 174c4ae5c..30713e75a 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -276,6 +276,7 @@ mod tests { table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, } diff --git a/data_plane/src/storage_engines/types/streaming_config.rs b/data_plane/src/storage_engines/types/streaming_config.rs index 91a2dc1b3..2298955db 100644 --- a/data_plane/src/storage_engines/types/streaming_config.rs +++ b/data_plane/src/storage_engines/types/streaming_config.rs @@ -133,6 +133,11 @@ impl StreamingConfig { num_aggregates_to_retain, QueryLanguage::PromQl, )?; + if config.derived_input.is_some() { + anyhow::bail!( + "legacy streaming input cannot execute a derived summary program" + ); + } // PR 5: the map key IS the policy-fingerprint u64. // `AggregationConfig::policy_fp_u64()` is the canonical // accessor for this value. @@ -191,6 +196,17 @@ mod tests { assert_eq!(cfg.storage_backend(), StorageBackend::GorillaObjectStore); } + #[test] + fn legacy_yaml_rejects_derived_summary_input() { + let data = serde_yaml::from_str::(&format!( + "aggregations:\n- aggregationType: Sum\n aggregationSubType: ''\n metric: outer\n labels: {{grouping: [], rollup: [], aggregated: []}}\n parameters: {{}}\n windowSize: 10\n windowType: tumbling\n spatialFilter: ''\n derived_input:\n inputs: [1]\n program_sha256: '{}'\n", "a".repeat(64) + )).unwrap(); + let error = StreamingConfig::from_yaml_data(&data).unwrap_err(); + assert!(error + .to_string() + .contains("legacy streaming input cannot execute")); + } + /// PR 5: a streaming-config YAML that omits `aggregationId` /// parses correctly — the backend derives identity from content /// via `PolicyFingerprint::from_config`. The map key is the diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs index 3fce2d44b..564d21d7c 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -109,6 +109,7 @@ pub fn create_engine_single_pop_with_aggregated( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -206,6 +207,7 @@ pub fn create_engine_dual_input( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -233,6 +235,7 @@ pub fn create_engine_dual_input( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -325,6 +328,7 @@ pub fn create_engine_two_metrics( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -351,6 +355,7 @@ pub fn create_engine_two_metrics( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -453,6 +458,7 @@ pub fn create_engine_three_metrics( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -532,6 +538,7 @@ pub fn create_engine_multi_timestamp( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; @@ -603,6 +610,7 @@ pub fn create_engine_multi_timestamp_with_window( table_name: None, value_projection: None, table_population: None, + derived_input: None, table_timestamp_column: None, partitioning: None, }; diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 048dcbac4..39956f7fe 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -429,3 +429,24 @@ This is an explicit lifetime transition, not cross-generation recovery of arbitr summary state. Legacy records without trustworthy catalog provenance remain unbound. Tombstone reclamation still requires coordinated removal of old physical parts and is not implemented by this transition. + +### Derived summary input identity + +A summary computed from another summary has a different data source from the +original raw table or metric. `PrecomputeMaterialization.derived_input` and +`DataSourceIdentity::Derived` use the same `DerivedInputIdentity`: the referenced +`SummaryDefinitionId`s and a SHA-256 of the maintenance program. The executable +program remains in `OwnedPostAsapDag`; the catalog does not retain another copy. + +The signature replaces materialized input frontiers with stable summary IDs and +hashes the remaining node payloads, schemas, guarantees, and edge semantics. It +excludes query names, plan-local node numbering, and catalog generations. A changed +input definition or transformation creates a new identity. Existing raw-source +identities retain their previous byte representation. Catalog validation rejects +missing input definitions and dependency cycles. + +This contract is a prerequisite, not enabled summary-over-summary execution. +Installation currently rejects derived inputs so they cannot accidentally receive +raw samples through the legacy metric router. Enabling them requires the immutable +maintenance consumer and durable output deduplication protocol; neither raw-table +substitution nor treating late correction fragments as new observations is valid. From bcbb5a40129493e6b724c18916979704a4028537 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 08:44:47 -0600 Subject: [PATCH 2/7] fix(catalog): separate derived policy domain from raw metric names --- crates/asap_types/src/derived_input.rs | 71 +++++++++++++++++-- crates/asap_types/src/policy_fingerprint.rs | 4 +- .../summary-catalog-sds-architecture.md | 3 +- 3 files changed, 71 insertions(+), 7 deletions(-) diff --git a/crates/asap_types/src/derived_input.rs b/crates/asap_types/src/derived_input.rs index 5aa86fbb2..a5f6d6dce 100644 --- a/crates/asap_types/src/derived_input.rs +++ b/crates/asap_types/src/derived_input.rs @@ -40,7 +40,20 @@ impl DerivedInputIdentity { if document.schema_version != crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION { return Err("unsupported derived program document version".into()); } - document.decode()?; + let decoded = document.decode()?; + let literals: BTreeSet<_> = decoded + .nodes + .iter() + .filter_map(|node| { + matches!( + &node.payload, + planner_types::post_asap::ExecutableOperatorPayload::Fallback { + expression: planner_types::pre_asap::QueryExpr::Literal(_), + } + ) + .then_some(node.id) + }) + .collect(); let mut incoming: BTreeMap<_, Vec<_>> = BTreeMap::new(); for edge in &document.edges { incoming.entry(edge.consumer).or_default().push(edge); @@ -70,10 +83,10 @@ impl DerivedInputIdentity { return Err("derived program has a cycle".into()); } stack.push((id, true)); - let edges = incoming - .get(&id) - .ok_or("derived program has an unbound input leaf")?; - for edge in edges { + if !incoming.contains_key(&id) && !literals.contains(&id) { + return Err("derived program has an unbound input leaf".into()); + } + for edge in incoming.get(&id).into_iter().flatten() { stack.push((edge.producer, false)); } continue; @@ -165,6 +178,22 @@ mod tests { assert_eq!(decoded.policy_fingerprint(), derived.policy_fingerprint()); } + #[test] + fn raw_utf8_metric_cannot_impersonate_derived_policy_domain() { + let mut derived = config(); + derived.derived_input = Some(DerivedInputIdentity { + inputs: BTreeSet::from([SummaryDefinitionId::from(derived.policy_fingerprint())]), + program_sha256: "d".repeat(64), + }); + let mut raw = derived.clone(); + raw.metric = format!( + "derived-input-v1:{}", + serde_json::to_string(&derived.source_identity()).unwrap() + ); + raw.derived_input = None; + assert_ne!(raw.policy_fingerprint(), derived.policy_fingerprint()); + } + #[test] fn catalog_rejects_missing_derived_dependencies_and_raw_source_conflicts() { let mut derived = config(); @@ -319,4 +348,36 @@ mod tests { PrecomputePlan::build(envelope, vec![config], &["producer".into()]).unwrap_err(); assert!(error.to_string().contains("immutable maintenance consumer")); } + #[test] + fn literal_leaves_are_hashed_without_inventing_materialization_references() { + use planner_types::{ + post_asap::{ExecutableOperator, ExecutableOperatorPayload}, + pre_asap::{QueryExpr, ScalarValue}, + }; + let mut dag = program(1, 2); + let mut literal = dag.nodes[0].clone(); + literal.id = PostAsapNodeId(3); + literal.operator = ExecutableOperator::Fallback; + literal.payload = serde_json::to_value(ExecutableOperatorPayload::Fallback { + expression: QueryExpr::Literal(ScalarValue::Int64(2)), + }) + .unwrap(); + dag.nodes.push(literal); + let mut edge = dag.edges[0].clone(); + edge.producer = PostAsapNodeId(3); + dag.edges.push(edge); + let source = SummaryDefinitionId::from(config().policy_fingerprint()); + let frontiers = BTreeMap::from([(PostAsapNodeId(1), source)]); + let first = DerivedInputIdentity::from_dag(&dag, dag.root, &frontiers).unwrap(); + assert_eq!(first.inputs, BTreeSet::from([source])); + dag.nodes[2].payload = serde_json::to_value(ExecutableOperatorPayload::Fallback { + expression: QueryExpr::Literal(ScalarValue::Int64(3)), + }) + .unwrap(); + assert_ne!( + first, + DerivedInputIdentity::from_dag(&dag, dag.root, &frontiers).unwrap() + ); + assert!(DerivedInputIdentity::from_dag(&dag, PostAsapNodeId(3), &frontiers).is_err()); + } } diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 46837a95e..38afc45ff 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -91,7 +91,9 @@ impl PolicyFingerprint { // 1. metric name if cfg.derived_input.is_some() { - buf.extend_from_slice(b"derived-input-v1:"); + // Raw policies start with UTF-8 metric bytes; 0xff is impossible + // there, so a metric cannot impersonate this source domain. + buf.extend_from_slice(b"\xffderived-input-v1:"); buf.extend_from_slice( &serde_json::to_vec(&cfg.source_identity()).expect("typed source identity"), ); diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 39956f7fe..4196f079a 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -440,7 +440,8 @@ program remains in `OwnedPostAsapDag`; the catalog does not retain another copy. The signature replaces materialized input frontiers with stable summary IDs and hashes the remaining node payloads, schemas, guarantees, and edge semantics. It -excludes query names, plan-local node numbering, and catalog generations. A changed +excludes query names, plan-local node numbering, and catalog generations. Literal +leaves are hashed directly; raw input leaves still require catalog frontiers. A changed input definition or transformation creates a new identity. Existing raw-source identities retain their previous byte representation. Catalog validation rejects missing input definitions and dependency cycles. From f73835bef0440838b08d71c889d2344ac953608d Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 09:16:11 -0600 Subject: [PATCH 3/7] fix(storage): reserve immutable output publication durably --- .../sketch_db/persistence/flusher.rs | 37 +- .../sketch_db/persistence/immutable_output.rs | 512 ++++++++++++++++++ .../sketch_db/persistence/metadata.rs | 75 ++- .../sketch_db/persistence/mod.rs | 1 + .../sketch_db/persistence/recovery.rs | 12 +- 5 files changed, 629 insertions(+), 8 deletions(-) create mode 100644 data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs diff --git a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs index e774b39e6..4c8307db2 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs @@ -32,7 +32,7 @@ use super::{PersistError, PersistResult}; /// Handle to a running flusher thread. Dropping the handle signals /// shutdown and joins the thread. pub struct FlusherHandle { - inner: Arc, + pub(super) inner: Arc, thread: Option>, } @@ -58,6 +58,14 @@ pub(crate) struct FlusherShared { pub back_pressure_wait_count: AtomicU64, } +impl FlusherShared { + pub(super) fn allocate_part_id(&self) -> PersistResult { + self.next_part_id + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1)) + .map_err(|_| PersistError::Internal("part ID exhausted".into())) + } +} + impl FlusherHandle { /// Start a flusher thread. Takes an `EpochSource` (typically the /// store itself, wrapped in `Arc`). @@ -84,12 +92,23 @@ impl FlusherHandle { { // Pick a starting part_id: one past the max currently in the // manifest (so IDs are monotonically increasing across restarts). + let reserved = sid_metadata.load_strict()?.into_iter().flat_map(|r| { + [r.pending_immutable, r.last_immutable] + .into_iter() + .flatten() + .map(|p| p.part_id) + }); let next_id = manifest .live_parts() .iter() .map(|p| p.part_id) + .chain(reserved) .max() - .map(|m| m + 1) + .map(|m| { + m.checked_add(1) + .ok_or_else(|| PersistError::Internal("part ID exhausted".into())) + }) + .transpose()? .unwrap_or(1); let shared = Arc::new(FlusherShared { @@ -369,7 +388,7 @@ fn run_tick(shared: &Arc, source: &S) -> PersistR if !snapshots.is_empty() { // Build one part for the tick. - let part_id = shared.next_part_id.fetch_add(1, Ordering::Relaxed); + let part_id = shared.allocate_part_id()?; let part_dir = part_dir_path(&parts_root(&cfg.disk_path), part_id); let entries_total: usize = snapshots.iter().map(|s| s.len()).sum(); let size_bytes_estimate: u64 = snapshots.iter().map(|s| s.approx_bytes as u64).sum(); @@ -428,6 +447,18 @@ fn run_tick(shared: &Arc, source: &S) -> PersistR .into_iter() .filter(|p| p.max_ts < cutoff) .collect(); + // Read reservations after capturing candidates: a newly published part + // cannot enter the older candidate set after this check. + let reserved: std::collections::HashSet<_> = shared + .sid_metadata + .load_strict()? + .into_iter() + .filter_map(|record| record.pending_immutable.map(|pending| pending.part_id)) + .collect(); + let expired: Vec<_> = expired + .into_iter() + .filter(|part| !reserved.contains(&part.part_id)) + .collect(); for p in &expired { shared.manifest.append_delete(p.part_id)?; let dir = part_dir_path(&parts_root(&cfg.disk_path), p.part_id); diff --git a/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs b/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs new file mode 100644 index 000000000..7264f52b8 --- /dev/null +++ b/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs @@ -0,0 +1,512 @@ +//! Crash-resumable publication of one immutable output window. The SID sidecar +//! reserves the existing part ID before writing; no second payload store is used. +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use super::flusher::{FlusherHandle, FlusherShared}; +use super::manifest::PartEntry; +use super::metadata::SidMetaRecord; +use super::part::{part_dir_path, PartReader, PartWriter}; +use super::source::{EpochSnapshot, EpochSnapshotEntry}; +use super::{PersistError, PersistResult}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ImmutableOutputReservation { + pub part_id: u64, + pub input_digest: [u8; 32], + pub payload_digest: [u8; 32], + pub start_ms: u64, + pub end_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ImmutableWindowPublication { + pub part_id: u64, + pub already_published: bool, +} + +fn invalid(message: &str) -> PersistError { + PersistError::Format(message.into()) +} + +fn fingerprint(snapshot: &EpochSnapshot) -> PersistResult<[u8; 32]> { + if snapshot.entries.is_empty() || snapshot.min_ts >= snapshot.max_ts { + return Err(invalid("immutable publication requires a nonempty window")); + } + let mut rows = Vec::with_capacity(snapshot.entries.len()); + for entry in &snapshot.entries { + if entry.start_ts != snapshot.min_ts || entry.end_ts != snapshot.max_ts { + return Err(invalid("immutable publication spans multiple windows")); + } + let label = entry.label.as_ref().map(|label| label.serialize_to_bytes()); + rows.push(( + label, + &entry.sketch_type_name, + entry.encoding_tag, + &entry.sketch_bytes, + )); + } + rows.sort(); + if rows.windows(2).any(|pair| pair[0].0 == pair[1].0) { + return Err(invalid( + "immutable publication contains duplicate label populations", + )); + } + let mut digest = Sha256::new(); + digest.update(b"asap-immutable-window-v1"); + for value in [ + snapshot.agg_id, + snapshot.min_ts, + snapshot.max_ts, + rows.len() as u64, + ] { + digest.update(value.to_le_bytes()); + } + for (label, kind, encoding, payload) in rows { + digest.update([u8::from(label.is_some())]); + let label = label.as_deref().unwrap_or_default(); + for bytes in [label, kind.as_bytes()] { + digest.update((bytes.len() as u64).to_le_bytes()); + digest.update(bytes); + } + digest.update([encoding]); + digest.update((payload.len() as u64).to_le_bytes()); + digest.update(payload); + } + Ok(digest.finalize().into()) +} + +impl FlusherHandle { + /// Reuse the existing persistence state without transferring ownership of + /// its worker thread. The store can retain a Weak reference to this Arc. + pub(crate) fn publication_handle(&self) -> std::sync::Arc { + std::sync::Arc::clone(&self.inner) + } + pub fn publish_immutable_window( + &self, + record: &SidMetaRecord, + input_digest: [u8; 32], + snapshot: &EpochSnapshot, + ) -> PersistResult { + self.inner + .publish_immutable_window(record, input_digest, snapshot) + } + pub fn resume_pending_immutable_window( + &self, + sid: u64, + ) -> PersistResult> { + self.inner.resume_pending_immutable_window(sid) + } +} + +impl FlusherShared { + /// Publish a finalized window. A retry must present exactly the reserved + /// input and payload. This does not guarantee source availability after GC. + pub fn publish_immutable_window( + &self, + record: &SidMetaRecord, + input_digest: [u8; 32], + snapshot: &EpochSnapshot, + ) -> PersistResult { + if snapshot.agg_id != record.sid || record.removed { + return Err(invalid("immutable publication SID is invalid or removed")); + } + let payload_digest = fingerprint(snapshot)?; + self.sid_metadata.transaction(|records, metadata| { + let key = record.sid.to_string(); + let mut current = records.get(&key).cloned().unwrap_or_else(|| record.clone()); + if current.removed + || current.retired_at_ms.is_some() + || current.expires_at_ms.is_some() + || current.summary_definition_id != record.summary_definition_id + || current.catalog_generation != record.catalog_generation + || current.metric_name != record.metric_name + || current.group_by_keys != record.group_by_keys + || !current.same_kind(record) + { + return Err(invalid("immutable publication metadata identity changed")); + } + if current + .completed_through_ms + .is_some_and(|end| snapshot.min_ts < end) + { + let Some(previous) = ¤t.last_immutable else { + return Err(invalid( + "completed immutable window has no retained lineage proof", + )); + }; + if previous.input_digest != input_digest + || previous.payload_digest != payload_digest + || previous.start_ms != snapshot.min_ts + || previous.end_ms != snapshot.max_ts + { + return Err(invalid( + "completed immutable retry differs from latest publication", + )); + } + self.validate_reserved_part(record.sid, previous)?; + return Ok(ImmutableWindowPublication { + part_id: previous.part_id, + already_published: true, + }); + } + let reservation = match current.pending_immutable.clone() { + Some(pending) => { + if pending.input_digest != input_digest + || pending.payload_digest != payload_digest + || pending.start_ms != snapshot.min_ts + || pending.end_ms != snapshot.max_ts + { + return Err(invalid( + "another immutable publication is pending for this SID", + )); + } + pending + } + None => { + let pending = ImmutableOutputReservation { + part_id: self.allocate_part_id()?, + input_digest, + payload_digest, + start_ms: snapshot.min_ts, + end_ms: snapshot.max_ts, + }; + current.pending_immutable = Some(pending.clone()); + records.insert(key.clone(), current.clone()); + metadata.write_records(records)?; + pending + } + }; + let path = part_dir_path(&self.cfg.disk_path.join("parts"), reservation.part_id); + if path.exists() { + // Never overwrite a reserved part silently: callers can recover a + // durable part without sources; damaged/partial parts fail closed. + self.validate_reserved_part(record.sid, &reservation)?; + } else { + PartWriter::write_part(&path, reservation.part_id, std::slice::from_ref(snapshot))?; + std::fs::File::open( + path.parent() + .ok_or_else(|| invalid("missing parts parent"))?, + )? + .sync_all()?; + self.validate_reserved_part(record.sid, &reservation)?; + } + self.finish_reserved_part(&mut current, &reservation)?; + records.insert(key, current); + metadata.write_records(records)?; + Ok(ImmutableWindowPublication { + part_id: reservation.part_id, + already_published: false, + }) + }) + } + + /// Complete a pending durable part without reconstructing its source input. + /// A reservation whose part was never completed returns an error, preserving + /// the reservation for an explicit recovery policy rather than losing data. + pub fn resume_pending_immutable_window( + &self, + sid: u64, + ) -> PersistResult> { + self.sid_metadata.transaction(|records, metadata| { + let key = sid.to_string(); + let Some(mut current) = records.get(&key).cloned() else { + return Ok(None); + }; + let Some(pending) = current.pending_immutable.clone() else { + return Ok(None); + }; + if current.removed || current.retired_at_ms.is_some() || current.expires_at_ms.is_some() + { + return Err(invalid("pending immutable SID was removed")); + } + self.validate_reserved_part(sid, &pending)?; + self.finish_reserved_part(&mut current, &pending)?; + records.insert(key, current); + metadata.write_records(records)?; + Ok(Some(ImmutableWindowPublication { + part_id: pending.part_id, + already_published: true, + })) + }) + } + + fn validate_reserved_part( + &self, + sid: u64, + pending: &ImmutableOutputReservation, + ) -> PersistResult<()> { + let reader = PartReader::open(&part_dir_path( + &self.cfg.disk_path.join("parts"), + pending.part_id, + ))?; + if reader.meta.part_id != pending.part_id { + return Err(invalid("reserved part ID mismatch")); + } + let mut entries = Vec::new(); + for index in reader.index_records() { + let row = reader.load_entry(&index)?; + if row.agg_id != sid { + return Err(invalid("reserved part contains another SID")); + } + entries.push(EpochSnapshotEntry { + start_ts: row.start_ts, + end_ts: row.end_ts, + label: row.label, + sketch_type_name: row.sketch_type_name, + encoding_tag: row.encoding_tag, + sketch_bytes: row.sketch_bytes, + }); + } + let snapshot = EpochSnapshot { + agg_id: sid, + epoch_id: 0, + min_ts: pending.start_ms, + max_ts: pending.end_ms, + entries, + approx_bytes: 0, + }; + if fingerprint(&snapshot)? != pending.payload_digest { + return Err(invalid("reserved immutable payload digest mismatch")); + } + Ok(()) + } + + fn finish_reserved_part( + &self, + current: &mut SidMetaRecord, + pending: &ImmutableOutputReservation, + ) -> PersistResult<()> { + let path = part_dir_path(&self.cfg.disk_path.join("parts"), pending.part_id); + // The reservation may have been recovered before its parent-directory + // entry was durable; make that durable before manifest publication too. + std::fs::File::open( + path.parent() + .ok_or_else(|| invalid("missing parts parent"))?, + )? + .sync_all()?; + if !self + .manifest + .live_parts() + .iter() + .any(|part| part.part_id == pending.part_id) + { + let size_bytes = std::fs::read_dir(&path)?.try_fold(0u64, |sum, entry| { + let length = entry?.metadata()?.len(); + sum.checked_add(length) + .ok_or_else(|| std::io::Error::other("part size overflow")) + })?; + self.manifest.append_add(PartEntry { + part_id: pending.part_id, + min_ts: pending.start_ms, + max_ts: pending.end_ms, + size_bytes, + })?; + } + current.completed_through_ms = Some( + current + .completed_through_ms + .unwrap_or(0) + .max(pending.end_ms), + ); + current.last_immutable = Some(pending.clone()); + current.pending_immutable = None; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::super::{ + config::SketchStorePersistenceConfig, + manifest::Manifest, + source::{EpochSource, SealedEpochRef}, + }; + use super::*; + use crate::storage_engines::{sketch_db::data::AggKind, types::AggregationType}; + use std::sync::Arc; + + struct EmptySource; + impl EpochSource for EmptySource { + fn list_sealed_epochs(&self) -> Vec { + vec![] + } + fn snapshot_sealed_epoch(&self, _: u64, _: u64) -> PersistResult> { + Ok(None) + } + fn evict_sealed_epoch(&self, _: u64, _: u64) {} + fn approx_memory_bytes(&self) -> usize { + 0 + } + } + fn handle(path: &std::path::Path) -> FlusherHandle { + let mut config = SketchStorePersistenceConfig::with_memory_limit(1024, path.into()); + config.hot_window_ms = None; + config.delete_older_than_ms = None; + FlusherHandle::start( + config, + Arc::new(Manifest::open_or_init(path).unwrap()), + Arc::new(EmptySource), + ) + .unwrap() + } + fn record() -> SidMetaRecord { + SidMetaRecord::new( + 1, + "derived".into(), + vec![], + &AggKind::ExactAgg { + agg_type: AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + 0, + ) + } + fn snapshot() -> EpochSnapshot { + EpochSnapshot { + agg_id: 1, + epoch_id: 0, + min_ts: 10, + max_ts: 20, + approx_bytes: 8, + entries: vec![EpochSnapshotEntry { + start_ts: 10, + end_ts: 20, + label: None, + sketch_type_name: "SumAccumulator".into(), + encoding_tag: 0, + sketch_bytes: vec![1, 2, 3], + }], + } + } + fn reserve(handle: &FlusherHandle, snapshot: &EpochSnapshot) -> ImmutableOutputReservation { + let pending = ImmutableOutputReservation { + part_id: handle.inner.allocate_part_id().unwrap(), + input_digest: [7; 32], + payload_digest: fingerprint(snapshot).unwrap(), + start_ms: 10, + end_ms: 20, + }; + handle + .inner + .sid_metadata + .transaction(|records, metadata| { + let mut record = record(); + record.pending_immutable = Some(pending.clone()); + records.insert("1".into(), record); + metadata.write_records(records) + }) + .unwrap(); + pending + } + #[test] + fn completed_retry_and_concurrent_publications_do_not_duplicate_parts() { + let temp = tempfile::tempdir().unwrap(); + let handle = handle(temp.path()); + std::thread::scope(|scope| { + let workers: Vec<_> = (0..4) + .map(|_| { + scope.spawn(|| { + handle + .publish_immutable_window(&record(), [7; 32], &snapshot()) + .unwrap() + }) + }) + .collect(); + let ids: Vec<_> = workers + .into_iter() + .map(|worker| worker.join().unwrap().part_id) + .collect(); + assert!(ids.iter().all(|id| *id == ids[0])); + }); + assert_eq!(handle.manifest().live_parts().len(), 1); + assert!(handle + .publish_immutable_window(&record(), [8; 32], &snapshot()) + .is_err()); + let metadata = handle.inner.sid_metadata.load_strict().unwrap(); + assert_eq!(metadata[0].completed_through_ms, Some(20)); + assert!(metadata[0].pending_immutable.is_none()); + let mut different = snapshot(); + different.entries[0].sketch_bytes.push(4); + assert!(handle + .publish_immutable_window(&record(), [7; 32], &different) + .is_err()); + } + #[test] + fn restart_preserves_reserved_part_and_resumes_without_source() { + let temp = tempfile::tempdir().unwrap(); + let mut first = handle(temp.path()); + let state = snapshot(); + let pending = reserve(&first, &state); + let path = part_dir_path(&temp.path().join("parts"), pending.part_id); + PartWriter::write_part(&path, pending.part_id, &[state]).unwrap(); + first.shutdown(); + drop(first); + let (_, recovery) = super::super::recovery::recover(temp.path()).unwrap(); + assert_eq!(recovery.orphan_parts_removed, 0); + let resumed = handle(temp.path()); + assert!(resumed.inner.allocate_part_id().unwrap() > pending.part_id); + let result = resumed.resume_pending_immutable_window(1).unwrap().unwrap(); + assert_eq!(result.part_id, pending.part_id); + assert_eq!(resumed.manifest().live_parts().len(), 1); + assert!(resumed + .resume_pending_immutable_window(1) + .unwrap() + .is_none()); + } + #[test] + fn pending_missing_part_requires_source_and_stale_upsert_preserves_reservation() { + let temp = tempfile::tempdir().unwrap(); + let handle = handle(temp.path()); + let pending = reserve(&handle, &snapshot()); + handle.inner.sid_metadata.upsert_all(&[record()]).unwrap(); + assert_eq!( + handle.inner.sid_metadata.load_strict().unwrap()[0].pending_immutable, + Some(pending.clone()) + ); + assert!(handle.resume_pending_immutable_window(1).is_err()); + assert_eq!( + handle + .publish_immutable_window(&record(), [7; 32], &snapshot()) + .unwrap() + .part_id, + pending.part_id + ); + handle.inner.sid_metadata.upsert_all(&[record()]).unwrap(); + assert!(handle.inner.sid_metadata.load_strict().unwrap()[0] + .pending_immutable + .is_none()); + } + #[test] + fn malformed_metadata_fails_before_part_publication() { + let temp = tempfile::tempdir().unwrap(); + let handle = handle(temp.path()); + std::fs::write(handle.inner.sid_metadata.path(), b"broken").unwrap(); + assert!(handle + .publish_immutable_window(&record(), [7; 32], &snapshot()) + .is_err()); + assert!(handle.manifest().live_parts().is_empty()); + } + #[test] + fn failed_manifest_leaves_recoverable_payload_and_no_completion() { + let temp = tempfile::tempdir().unwrap(); + let handle = handle(temp.path()); + let log = handle.manifest().log_path(); + let backup = log.with_extension("saved"); + std::fs::rename(&log, &backup).unwrap(); + std::fs::create_dir(&log).unwrap(); + assert!(handle + .publish_immutable_window(&record(), [7; 32], &snapshot()) + .is_err()); + let records = handle.inner.sid_metadata.load_strict().unwrap(); + let pending = records[0].pending_immutable.clone().unwrap(); + assert_eq!(records[0].completed_through_ms, None); + assert!(handle.manifest().live_parts().is_empty()); + std::fs::remove_dir(&log).unwrap(); + std::fs::rename(&backup, &log).unwrap(); + let published = handle.resume_pending_immutable_window(1).unwrap().unwrap(); + assert_eq!(published.part_id, pending.part_id); + assert_eq!(handle.manifest().live_parts().len(), 1); + } +} diff --git a/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs index c432e07f2..dfdf7b4bc 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs @@ -322,6 +322,10 @@ pub struct SidMetaRecord { /// No further publication may change a window ending at or before this bound. #[serde(default)] pub completed_through_ms: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pending_immutable: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_immutable: Option, } impl SidMetaRecord { @@ -347,9 +351,15 @@ impl SidMetaRecord { expires_at_ms: None, removed: false, completed_through_ms: None, + pending_immutable: None, + last_immutable: None, } } + pub(super) fn same_kind(&self, other: &Self) -> bool { + self.agg_kind == other.agg_kind + } + /// Reconstruct the structured [`AggKind`], or `None` for an /// unrecognized sketch kind. pub fn agg_kind(&self) -> Option { @@ -416,6 +426,10 @@ struct SidBindingRec { removed: bool, #[serde(default)] completed_through_ms: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pending_immutable: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + last_immutable: Option, } /// Version-3 normalized sidecar with authoritative catalog provenance. Descriptors appear once and SeriesId bindings hold @@ -487,6 +501,8 @@ impl SdsSidecar { expires_at_ms: record.expires_at_ms, removed: record.removed, completed_through_ms: record.completed_through_ms, + pending_immutable: record.pending_immutable, + last_immutable: record.last_immutable, }, ); } @@ -541,6 +557,8 @@ impl SdsSidecar { expires_at_ms: binding.expires_at_ms, removed: binding.removed, completed_through_ms: binding.completed_through_ms, + pending_immutable: binding.pending_immutable, + last_immutable: binding.last_immutable, }) }) .collect() @@ -646,7 +664,7 @@ impl SidMetadataStore { return Ok(()); } let mut map: HashMap = self - .load()? + .load_strict()? .into_iter() .map(|r| (r.sid.to_string(), r)) .collect(); @@ -657,6 +675,8 @@ impl SidMetadataStore { if let Some(existing) = map.get(&key) { // Lifecycle is monotone for a SeriesId. An older flush snapshot // must not resurrect a retired or removed persisted instance. + next.pending_immutable = existing.pending_immutable.clone(); + next.last_immutable = existing.last_immutable.clone(); next.removed |= existing.removed; next.completed_through_ms = existing.completed_through_ms.max(next.completed_through_ms); @@ -680,6 +700,55 @@ impl SidMetadataStore { self.write_atomic(json.as_bytes()) } + pub fn load_strict(&self) -> PersistResult> { + let bytes = match fs::read(&self.path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(error.into()), + }; + let value: serde_json::Value = serde_json::from_slice(&bytes) + .map_err(|error| PersistError::Format(format!("invalid SID metadata: {error}")))?; + if let Some(version) = value.get("schema_version") { + if !matches!(version.as_u64(), Some(2 | 3)) { + return Err(PersistError::Format( + "unsupported SID metadata version".into(), + )); + } + let sidecar: SdsSidecar = serde_json::from_value(value) + .map_err(|error| PersistError::Format(error.to_string()))?; + sidecar.into_records() + } else { + let records: HashMap = serde_json::from_value(value) + .map_err(|error| PersistError::Format(error.to_string()))?; + Ok(records.into_values().collect()) + } + } + + pub(super) fn transaction( + &self, + operation: impl FnOnce(&mut HashMap, &Self) -> PersistResult, + ) -> PersistResult { + let _writer = self + .writer + .lock() + .map_err(|_| PersistError::Internal("SID metadata writer poisoned".into()))?; + let mut records = self + .load_strict()? + .into_iter() + .map(|r| (r.sid.to_string(), r)) + .collect(); + operation(&mut records, self) + } + pub(super) fn write_records( + &self, + records: &HashMap, + ) -> PersistResult<()> { + let sidecar = SdsSidecar::from_records(records.values().cloned()); + let bytes = + serde_json::to_vec(&sidecar).map_err(|e| PersistError::Serialize(e.to_string()))?; + self.write_atomic(&bytes) + } + fn write_atomic(&self, bytes: &[u8]) -> PersistResult<()> { if let Some(parent) = self.path.parent() { fs::create_dir_all(parent)?; @@ -696,9 +765,7 @@ impl SidMetadataStore { } fs::rename(&tmp, &self.path)?; if let Some(parent) = self.path.parent() { - if let Ok(dir) = File::open(parent) { - let _ = dir.sync_all(); - } + File::open(parent)?.sync_all()?; } Ok(()) } diff --git a/data_plane/src/storage_engines/sketch_db/persistence/mod.rs b/data_plane/src/storage_engines/sketch_db/persistence/mod.rs index 349feb92e..552fe4e03 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/mod.rs @@ -27,6 +27,7 @@ pub mod source; pub mod cache; pub mod flusher; +pub mod immutable_output; pub mod recovery; pub use config::SketchStorePersistenceConfig; diff --git a/data_plane/src/storage_engines/sketch_db/persistence/recovery.rs b/data_plane/src/storage_engines/sketch_db/persistence/recovery.rs index 7fb800946..a21402608 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/recovery.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/recovery.rs @@ -34,7 +34,12 @@ pub fn recover(disk_path: &Path) -> PersistResult<(Manifest, RecoveryReport)> { // Validate every live part by reading its meta.bin header. let mut to_drop: Vec = Vec::new(); - let mut referenced: HashSet = HashSet::new(); + let pending: HashSet = super::metadata::SidMetadataStore::new(disk_path) + .load_strict()? + .into_iter() + .filter_map(|record| record.pending_immutable.map(|pending| pending.part_id)) + .collect(); + let mut referenced = pending.clone(); for entry in manifest.live_parts() { referenced.insert(entry.part_id); let part_dir = super::part::part_dir_path(&parts_root, entry.part_id); @@ -61,6 +66,11 @@ pub fn recover(disk_path: &Path) -> PersistResult<(Manifest, RecoveryReport)> { } } + if to_drop.iter().any(|part_id| pending.contains(part_id)) { + return Err(super::PersistError::Format( + "corrupt reserved immutable part".into(), + )); + } for part_id in to_drop { manifest.append_delete(part_id)?; let dir = super::part::part_dir_path(&parts_root, part_id); From c9d0f088db60eb91b9ba92d75b4376f022f5a317 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 09:16:59 -0600 Subject: [PATCH 4/7] fix(storage): reject retry after manifest retirement --- .../sketch_db/persistence/immutable_output.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs b/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs index 7264f52b8..117b350bc 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs @@ -145,6 +145,14 @@ impl FlusherShared { )); } self.validate_reserved_part(record.sid, previous)?; + if !self + .manifest + .live_parts() + .iter() + .any(|part| part.part_id == previous.part_id) + { + return Err(invalid("completed immutable part is no longer published")); + } return Ok(ImmutableWindowPublication { part_id: previous.part_id, already_published: true, @@ -432,6 +440,11 @@ mod tests { assert!(handle .publish_immutable_window(&record(), [7; 32], &different) .is_err()); + let part_id = handle.manifest().live_parts()[0].part_id; + handle.manifest().append_delete(part_id).unwrap(); + assert!(handle + .publish_immutable_window(&record(), [7; 32], &snapshot()) + .is_err()); } #[test] fn restart_preserves_reserved_part_and_resumes_without_source() { From 2fc455a5d24a847a4047ed01de9fdf0a12070224 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 09:24:19 -0600 Subject: [PATCH 5/7] docs(sds): separate immutable window section --- docs/design_docs/summary-catalog-sds-architecture.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index e2fa87306..14b388e5c 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -462,6 +462,7 @@ Installation currently rejects derived inputs so they cannot accidentally receiv raw samples through the legacy metric router. Enabling them requires the immutable maintenance consumer and durable output deduplication protocol; neither raw-table substitution nor treating late correction fragments as new observations is valid. + ### Immutable completed windows Finite Remote Write completion now fences the SummaryStore append boundary, From 1bdd1e497615297bee44959491b679a9440a92df Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 09:28:47 -0600 Subject: [PATCH 6/7] fix(storage): look up committed lineage before sketch evaluation --- .../sketch_db/persistence/immutable_output.rs | 96 ++++++++++++++++--- 1 file changed, 85 insertions(+), 11 deletions(-) diff --git a/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs b/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs index 117b350bc..468c26f02 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs @@ -29,6 +29,22 @@ fn invalid(message: &str) -> PersistError { PersistError::Format(message.into()) } +fn validate_identity(current: &SidMetaRecord, record: &SidMetaRecord) -> PersistResult<()> { + if current.sid != record.sid + || current.removed + || current.retired_at_ms.is_some() + || current.expires_at_ms.is_some() + || current.summary_definition_id != record.summary_definition_id + || current.catalog_generation != record.catalog_generation + || current.metric_name != record.metric_name + || current.group_by_keys != record.group_by_keys + || !current.same_kind(record) + { + return Err(invalid("immutable publication metadata identity changed")); + } + Ok(()) +} + fn fingerprint(snapshot: &EpochSnapshot) -> PersistResult<[u8; 32]> { if snapshot.entries.is_empty() || snapshot.min_ts >= snapshot.max_ts { return Err(invalid("immutable publication requires a nonempty window")); @@ -100,6 +116,45 @@ impl FlusherHandle { } impl FlusherShared { + /// Find the latest committed result before evaluating a potentially + /// randomized sketch again. This never creates or replaces a payload. + pub fn lookup_immutable_window( + &self, + record: &SidMetaRecord, + input_digest: [u8; 32], + start_ms: u64, + end_ms: u64, + ) -> PersistResult> { + self.sid_metadata.transaction(|records, _| { + let Some(current) = records.get(&record.sid.to_string()) else { + return Ok(None); + }; + validate_identity(current, record)?; + let Some(previous) = ¤t.last_immutable else { + return Ok(None); + }; + if previous.start_ms != start_ms || previous.end_ms != end_ms { + return Ok(None); + } + if previous.input_digest != input_digest { + return Err(invalid("immutable lookup input lineage differs")); + } + self.validate_reserved_part(record.sid, previous)?; + if !self + .manifest + .live_parts() + .iter() + .any(|part| part.part_id == previous.part_id) + { + return Err(invalid("completed immutable part is no longer published")); + } + Ok(Some(ImmutableWindowPublication { + part_id: previous.part_id, + already_published: true, + })) + }) + } + /// Publish a finalized window. A retry must present exactly the reserved /// input and payload. This does not guarantee source availability after GC. pub fn publish_immutable_window( @@ -115,17 +170,7 @@ impl FlusherShared { self.sid_metadata.transaction(|records, metadata| { let key = record.sid.to_string(); let mut current = records.get(&key).cloned().unwrap_or_else(|| record.clone()); - if current.removed - || current.retired_at_ms.is_some() - || current.expires_at_ms.is_some() - || current.summary_definition_id != record.summary_definition_id - || current.catalog_generation != record.catalog_generation - || current.metric_name != record.metric_name - || current.group_by_keys != record.group_by_keys - || !current.same_kind(record) - { - return Err(invalid("immutable publication metadata identity changed")); - } + validate_identity(¤t, record)?; if current .completed_through_ms .is_some_and(|end| snapshot.min_ts < end) @@ -522,4 +567,33 @@ mod tests { assert_eq!(published.part_id, pending.part_id); assert_eq!(handle.manifest().live_parts().len(), 1); } + #[test] + fn committed_lookup_reuses_payload_without_recomputing_randomized_state() { + let temp = tempfile::tempdir().unwrap(); + let mut first = handle(temp.path()); + let publication = first + .publish_immutable_window(&record(), [7; 32], &snapshot()) + .unwrap(); + first.shutdown(); + drop(first); + let reopened = handle(temp.path()); + let found = reopened + .inner + .lookup_immutable_window(&record(), [7; 32], 10, 20) + .unwrap() + .unwrap(); + assert_eq!(found.part_id, publication.part_id); + assert!(reopened + .inner + .lookup_immutable_window(&record(), [8; 32], 10, 20) + .is_err()); + reopened + .manifest() + .append_delete(publication.part_id) + .unwrap(); + assert!(reopened + .inner + .lookup_immutable_window(&record(), [7; 32], 10, 20) + .is_err()); + } } From 03398cbd696d03dd2285471d31122ec82136f8fe Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 09:36:52 -0600 Subject: [PATCH 7/7] fix(storage): match pending lineage atomically during recovery --- .../sketch_db/persistence/immutable_output.rs | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs b/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs index 468c26f02..0ac85e833 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/immutable_output.rs @@ -254,12 +254,32 @@ impl FlusherShared { }) } + /// Resume only the caller's pending lineage, atomically with metadata + /// validation. A different pending input cannot be acknowledged by this call. + pub fn resume_matching_immutable_window( + &self, + record: &SidMetaRecord, + input_digest: [u8; 32], + start_ms: u64, + end_ms: u64, + ) -> PersistResult> { + self.resume_immutable_window(record.sid, Some((record, input_digest, start_ms, end_ms))) + } + /// Complete a pending durable part without reconstructing its source input. /// A reservation whose part was never completed returns an error, preserving /// the reservation for an explicit recovery policy rather than losing data. pub fn resume_pending_immutable_window( &self, sid: u64, + ) -> PersistResult> { + self.resume_immutable_window(sid, None) + } + + fn resume_immutable_window( + &self, + sid: u64, + expected: Option<(&SidMetaRecord, [u8; 32], u64, u64)>, ) -> PersistResult> { self.sid_metadata.transaction(|records, metadata| { let key = sid.to_string(); @@ -273,6 +293,15 @@ impl FlusherShared { { return Err(invalid("pending immutable SID was removed")); } + if let Some((record, digest, start, end)) = expected { + validate_identity(¤t, record)?; + if pending.input_digest != digest + || pending.start_ms != start + || pending.end_ms != end + { + return Err(invalid("pending immutable recovery lineage differs")); + } + } self.validate_reserved_part(sid, &pending)?; self.finish_reserved_part(&mut current, &pending)?; records.insert(key, current); @@ -596,4 +625,35 @@ mod tests { .lookup_immutable_window(&record(), [7; 32], 10, 20) .is_err()); } + #[test] + fn matching_resume_rejects_other_lineage_without_mutating_reservation() { + let temp = tempfile::tempdir().unwrap(); + let handle = handle(temp.path()); + let state = snapshot(); + let pending = reserve(&handle, &state); + let path = part_dir_path(&temp.path().join("parts"), pending.part_id); + PartWriter::write_part(&path, pending.part_id, &[state]).unwrap(); + assert!(handle + .inner + .resume_matching_immutable_window(&record(), [8; 32], 10, 20) + .is_err()); + assert!(handle + .inner + .resume_matching_immutable_window(&record(), [7; 32], 20, 30) + .is_err()); + assert!(handle.manifest().live_parts().is_empty()); + assert_eq!( + handle.inner.sid_metadata.load_strict().unwrap()[0].pending_immutable, + Some(pending.clone()) + ); + assert_eq!( + handle + .inner + .resume_matching_immutable_window(&record(), [7; 32], 10, 20) + .unwrap() + .unwrap() + .part_id, + pending.part_id + ); + } }