From 2752c31fb3a768fb00ddcb375bc130717ca4063e Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 11:57:20 -0600 Subject: [PATCH 1/3] Acquire complete immutable input cohorts before maintenance publication --- .../precompute_engine/maintenance_runtime.rs | 8 +- .../sketch_db/index/maintenance.rs | 180 ++++++++++++++++++ 2 files changed, 186 insertions(+), 2 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index cbdaf19f..08b0298f 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -608,8 +608,12 @@ pub fn execute_completed_maintenance( let generation = store .active_catalog_generation() .ok_or("maintenance requires an authoritative catalog")?; - let frozen = - store.read_frozen_exact_windows(source_sid, source, &generation, &expected, group)?; + let mut cohort = store.read_frozen_exact_cohort( + &generation, + &derived.inputs, + &[(source_sid, source, expected, group.clone())], + )?; + let frozen = cohort.pop().ok_or("immutable input cohort is empty")?; let (dag, key, states) = prepare_frozen_maintenance_sink(installed, configs, sink, &frozen, window)?; let digest = key diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs index 53e6064f..c49af73a 100644 --- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -16,6 +16,49 @@ pub(crate) struct FrozenExactWindows { } impl SketchStore { + /// Acquire a complete immutable read set before any output is reserved. + /// Individual payloads are immutable; checking the shared generation again + /// after all reads prevents a catalog transition from mixing incarnations. + pub(crate) fn read_frozen_exact_cohort( + &self, + generation: &Arc, + expected_definitions: &BTreeSet, + requests: &[( + u64, + SummaryDefinitionId, + BTreeSet<(u64, u64)>, + BTreeMap, + )], + ) -> Result, String> { + if requests.is_empty() || requests.len() > 65_536 { + return Err("immutable input cohort has invalid population count".into()); + } + let supplied = requests + .iter() + .map(|(_, definition, _, _)| *definition) + .collect(); + if expected_definitions != &supplied { + return Err("immutable input cohort differs from installed input definitions".into()); + } + let mut ordered: Vec<_> = requests.iter().collect(); + ordered.sort_by(|left, right| (left.1, left.0, &left.3).cmp(&(right.1, right.0, &right.3))); + if ordered + .windows(2) + .any(|pair| (pair[0].1, pair[0].0, &pair[0].3) == (pair[1].1, pair[1].0, &pair[1].3)) + { + return Err("immutable input cohort repeats a physical population".into()); + } + self.validate_routed_catalog_generation(Some(generation.as_ref()))?; + let inputs = ordered + .into_iter() + .map(|(sid, definition, windows, group)| { + self.read_frozen_exact_windows(*sid, *definition, generation, windows, group) + }) + .collect::, _>>()?; + self.validate_routed_catalog_generation(Some(generation.as_ref()))?; + Ok(inputs) + } + fn durable_maintenance_population_ids( &self, definition: SummaryDefinitionId, @@ -411,6 +454,102 @@ mod tests { use crate::storage_engines::types::PrecomputedOutput; use asap_types::traits::SerializableToSink; + #[test] + fn cohort_requires_every_durable_source_in_one_catalog_generation() { + // Neither a missing second window nor a new catalog may yield a + // partially acquired cohort, even when the first source is complete. + let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_str(include_str!( + "../../../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + let plan = snapshot.compile().unwrap(); + let mut first = plan.precompute_plan.materializations[0].clone(); + first.aggregation_type = asap_types::AggregationType::Sum; + first.aggregation_sub_type = "sum".into(); + first.grouping_labels = std::iter::empty::().collect(); + let mut second = first.clone(); + second.metric = "cohort_second".into(); + let configs = [first, second]; + let catalog = + asap_types::summary_catalog::SummaryCatalog::from_materializations(1, 1, &configs) + .unwrap(); + let store = Arc::new(SketchStore::new()); + store + .install_summary_catalog(Arc::new(catalog.clone())) + .unwrap(); + let generation = store.active_catalog_generation().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let mut config = persistence::config::SketchStorePersistenceConfig::with_memory_limit( + 1 << 24, + directory.path().to_path_buf(), + ); + config.delete_older_than_ms = None; + config.hot_window_ms = None; + let mut persistence = store.start_persistence(config).unwrap(); + let mut requests = Vec::new(); + for (index, config) in configs.iter().enumerate() { + let definition = config.policy_fingerprint().into(); + let coordinate = asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: definition, + time_range: HalfOpenTimeRange { + start_ms: 0, + end_ms: 1000, + }, + group_values: BTreeMap::new(), + }; + let revision = store + .admit_summary_updates(&generation, BTreeSet::from([coordinate.clone()])) + .unwrap(); + let mut output = PrecomputedOutput::new(0, 1000, None, config.policy_fingerprint()); + output.catalog_generation = Some(Arc::clone(&generation)); + let mut sum = SumAccumulator::new(); + sum.update(5.0 + index as f64); + let sid = 900 + index as u64; + store + .publish_admitted_summary_update( + &generation, + &coordinate, + revision, + revision, + 2000, + |writer| writer.ingest_precompute_with_series_id(sid, config, &output, &sum), + ) + .unwrap(); + requests.push(( + sid, + definition, + BTreeSet::from([(0, 1000)]), + BTreeMap::new(), + )); + } + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !store.seal_finite_summary_input(&generation).unwrap() { + assert!(std::time::Instant::now() < deadline); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + let definitions = requests.iter().map(|request| request.1).collect(); + let parts = persistence.manifest.live_parts().len(); + let cohort = store + .read_frozen_exact_cohort(&generation, &definitions, &requests) + .unwrap(); + assert_eq!(cohort.len(), 2); + assert!(cohort.iter().all(|input| input.generation == generation)); + requests[1].2.insert((1000, 2000)); + assert!(store + .read_frozen_exact_cohort(&generation, &definitions, &requests) + .is_err()); + assert_eq!(persistence.manifest.live_parts().len(), parts); + let mut next = catalog; + next.plan_version += 1; + store.install_summary_catalog(Arc::new(next)).unwrap(); + requests[1].2.remove(&(1000, 2000)); + assert!(store + .read_frozen_exact_cohort(&generation, &definitions, &requests) + .is_err()); + persistence.shutdown(); + } + #[test] fn one_sid_with_two_populations_cannot_publish_a_partial_global_summary() { let mut fixture: serde_json::Value = serde_json::from_str(include_str!( @@ -502,6 +641,47 @@ mod tests { ) .unwrap(); assert!(!frozen.singleton_population_complete); + // The read set is deterministic and all-or-nothing, independently of + // the later routing decision (which still rejects this global reduce). + let request = |name: &str| { + ( + 700, + source_id, + BTreeSet::from([(0, 60_000)]), + BTreeMap::from([("instance".to_string(), name.to_string())]), + ) + }; + let cohort = store + .read_frozen_exact_cohort( + &generation, + &BTreeSet::from([source_id]), + &[request("b"), request("a")], + ) + .unwrap(); + assert_eq!(cohort.len(), 2); + assert_eq!(cohort[0].group["instance"], "a"); + assert_eq!(cohort[1].group["instance"], "b"); + assert!(store + .read_frozen_exact_cohort( + &generation, + &BTreeSet::from([source_id]), + &[request("a"), request("a")] + ) + .is_err()); + assert!(store + .read_frozen_exact_cohort( + &generation, + &BTreeSet::from([source_id, target.policy_fingerprint().into()]), + &[request("a")] + ) + .is_err()); + assert!(store + .read_frozen_exact_cohort( + &generation, + &BTreeSet::from([source_id]), + &[request("a"), request("absent")] + ) + .is_err()); let parts_before = persistence.manifest.live_parts().len(); let resolver = crate::drivers::ingest::series_resolver::SeriesIdResolver::new(); for _ in 0..2 { From 190964c8a2c0eeb6306225d6a862bf4b7f281bda Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:05:10 -0600 Subject: [PATCH 2/3] Hold cohort lifetimes through immutable registration and publication --- .../precompute_engine/maintenance_runtime.rs | 4 +- .../sketch_db/index/maintenance.rs | 149 ++++++++++++++---- .../storage_engines/sketch_db/index/mod.rs | 67 +++++--- 3 files changed, 169 insertions(+), 51 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 08b0298f..290f064a 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -635,7 +635,7 @@ pub fn execute_completed_maintenance( if store.recover_frozen_maintenance_output( target_sid, target_config, - &frozen, + std::slice::from_ref(&frozen), digest, window, )? { @@ -665,7 +665,7 @@ pub fn execute_completed_maintenance( target_config, &output, state.as_ref(), - &frozen, + std::slice::from_ref(&frozen), digest, ) } diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs index c49af73a..c48680b4 100644 --- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -277,24 +277,71 @@ impl SketchStore { } impl SketchStore { + fn validate_frozen_cohort_bindings<'a>( + &self, + config: &asap_types::PrecomputeMaterialization, + sources: &'a [FrozenExactWindows], + instances: &HashMap, + ) -> Result<&'a Arc, String> { + let generation = &sources + .first() + .ok_or("immutable publication input cohort is empty")? + .generation; + let definitions = sources.iter().map(|source| source.definition).collect(); + if config + .derived_input + .as_ref() + .is_none_or(|input| input.inputs != definitions) + { + return Err( + "immutable publication input definitions differ from installed identity".into(), + ); + } + let mut populations = BTreeSet::new(); + for source in sources { + if &source.generation != generation + || !populations.insert((source.definition, source.sid, &source.group)) + { + return Err( + "immutable publication input cohort has mixed generations or duplicates".into(), + ); + } + let binding = instances + .get(&source.sid) + .ok_or("immutable source was removed")?; + if binding.metadata.policy_fp != source.definition.fingerprint() + || binding.catalog_generation.as_deref() != Some(generation.as_ref()) + || !binding.metadata.is_writable() + { + return Err("immutable source identity or lifetime changed".into()); + } + } + self.validate_routed_catalog_generation(Some(generation.as_ref()))?; + Ok(generation) + } + pub(crate) fn recover_frozen_maintenance_output( &self, sid: u64, config: &asap_types::PrecomputeMaterialization, - source: &FrozenExactWindows, + sources: &[FrozenExactWindows], digest: [u8; 32], window: (u64, u64), ) -> Result { - self.validate_routed_catalog_generation(Some(source.generation.as_ref()))?; + let _admission = self + .admission + .read() + .map_err(|_| "admission registry poisoned")?; let instances = self .instances .read() .map_err(|_| "instance registry poisoned")?; + let generation = self.validate_frozen_cohort_bindings(config, sources, &instances)?; let Some(binding) = instances.get(&sid) else { return Ok(false); }; if binding.metadata.policy_fp != config.policy_fingerprint() - || binding.catalog_generation.as_deref() != Some(source.generation.as_ref()) + || binding.catalog_generation.as_deref() != Some(generation.as_ref()) || !binding.metadata.is_writable() { return Err("immutable output identity or lifetime changed".into()); @@ -345,41 +392,37 @@ impl SketchStore { config: &asap_types::PrecomputeMaterialization, output: &crate::storage_engines::types::PrecomputedOutput, state: &dyn AggregateCore, - source: &FrozenExactWindows, + sources: &[FrozenExactWindows], input_digest: [u8; 32], ) -> Result { use persistence::source::{EpochSnapshot, EpochSnapshotEntry}; - if config - .derived_input - .as_ref() - .is_none_or(|derived| derived.inputs != BTreeSet::from([source.definition])) - || output.policy_fp != config.policy_fingerprint() - || output.catalog_generation.as_deref() != Some(source.generation.as_ref()) + // One guard excludes source retirement through target registration + // and commit; admission excludes catalog transitions in the same span. + let _admission = self + .admission + .read() + .map_err(|_| "admission registry poisoned")?; + let mut instances = self + .instances + .write() + .map_err(|_| "instance registry poisoned")?; + let generation = self.validate_frozen_cohort_bindings(config, sources, &instances)?; + if output.policy_fp != config.policy_fingerprint() + || output.catalog_generation.as_deref() != Some(generation.as_ref()) { return Err("derived publication differs from its installed input identity".into()); } let labels = self - .register_precompute_output(sid, config, output) + .register_precompute_output_with_instances(sid, config, output, &mut instances) .ok_or("derived output registration failed")?; let _mutation = self.begin_state_mutation(); - let instances = self - .instances - .read() - .map_err(|_| "instance registry poisoned")?; - let source_binding = instances - .get(&source.sid) - .ok_or("immutable source was removed")?; let binding = instances.get(&sid).ok_or("derived output was removed")?; - if source_binding.metadata.policy_fp != source.definition.fingerprint() - || source_binding.catalog_generation.as_deref() != Some(source.generation.as_ref()) - || source_binding.metadata.status() == AggStatus::Expired - || !binding.metadata.is_writable() + if !binding.metadata.is_writable() || binding.metadata.policy_fp != output.policy_fp - || binding.catalog_generation.as_deref() != Some(source.generation.as_ref()) + || binding.catalog_generation.as_deref() != Some(generation.as_ref()) { return Err("derived publication physical lifetime changed".into()); } - self.validate_routed_catalog_generation(Some(source.generation.as_ref()))?; let mut completed = self .completed_windows .write() @@ -470,7 +513,15 @@ mod tests { first.grouping_labels = std::iter::empty::().collect(); let mut second = first.clone(); second.metric = "cohort_second".into(); - let configs = [first, second]; + let mut target = first.clone(); + target.derived_input = Some(asap_types::derived_input::DerivedInputIdentity { + inputs: BTreeSet::from([ + first.policy_fingerprint().into(), + second.policy_fingerprint().into(), + ]), + program_sha256: "0".repeat(64), + }); + let configs = [first, second, target]; let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations(1, 1, &configs) .unwrap(); @@ -488,7 +539,7 @@ mod tests { config.hot_window_ms = None; let mut persistence = store.start_persistence(config).unwrap(); let mut requests = Vec::new(); - for (index, config) in configs.iter().enumerate() { + for (index, config) in configs.iter().take(2).enumerate() { let definition = config.policy_fingerprint().into(); let coordinate = asap_types::sds::SummaryInstanceCoordinates { summary_definition_id: definition, @@ -540,6 +591,37 @@ mod tests { .read_frozen_exact_cohort(&generation, &definitions, &requests) .is_err()); assert_eq!(persistence.manifest.live_parts().len(), parts); + let target = &configs[2]; + let mut output = PrecomputedOutput::new(0, 1000, None, target.policy_fingerprint()); + output.catalog_generation = Some(Arc::clone(&generation)); + let mut sum = SumAccumulator::new(); + sum.update(11.0); + assert!(store + .publish_frozen_maintenance_output(902, target, &output, &sum, &cohort, [42; 32]) + .unwrap()); + assert!(store + .recover_frozen_maintenance_output(902, target, &cohort, [42; 32], (0, 1000)) + .unwrap()); + let published_parts = persistence.manifest.live_parts().len(); + store.force_expire(901).unwrap(); + assert!(store + .publish_frozen_maintenance_output(903, target, &output, &sum, &cohort, [42; 32]) + .is_err()); + assert!(store + .recover_frozen_maintenance_output(902, target, &cohort, [42; 32], (0, 1000)) + .is_err()); + assert!(!store.instances.read().unwrap().contains_key(&903)); + assert_eq!(persistence.manifest.live_parts().len(), published_parts); + assert!(!store + .persistence_metadata + .read() + .unwrap() + .as_ref() + .unwrap() + .load_strict() + .unwrap() + .iter() + .any(|record| record.sid == 903)); let mut next = catalog; next.plan_version += 1; store.install_summary_catalog(Arc::new(next)).unwrap(); @@ -831,7 +913,7 @@ mod tests { let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations( 1, 1, - &[source_config, target.clone()], + &[source_config.clone(), target.clone()], ) .unwrap(); let store = Arc::new(SketchStore::new()); @@ -847,6 +929,11 @@ mod tests { let generation = store.active_catalog_generation().unwrap(); let mut output = PrecomputedOutput::new(0, 1000, None, target.policy_fingerprint()); output.catalog_generation = Some(Arc::clone(&generation)); + let mut source_output = output.clone(); + source_output.policy_fp = source_config.policy_fingerprint(); + store + .register_precompute_output(600, &source_config, &source_output) + .unwrap(); store .register_precompute_output(601, &target, &output) .unwrap(); @@ -883,7 +970,13 @@ mod tests { singleton_population_complete: false, }; assert!(store - .recover_frozen_maintenance_output(601, &target, &input, [7; 32], (0, 1000)) + .recover_frozen_maintenance_output( + 601, + &target, + std::slice::from_ref(&input), + [7; 32], + (0, 1000) + ) .unwrap()); assert_eq!( store.completed_windows.read().unwrap().get(&601), diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index e9ef1234..c7c29c90 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -818,6 +818,15 @@ impl SketchStore { /// where the two indexes were written under separate sequential /// locks). See the index-field doc comments for the full invariant. pub fn register(&self, meta: SketchInstanceMetadata) { + let mut instances = self.instances.write().unwrap(); + self.register_with_instances(meta, &mut instances); + } + + fn register_with_instances( + &self, + meta: SketchInstanceMetadata, + instances: &mut HashMap, + ) -> bool { let sid = meta.sid; let policy_fp = meta.policy_fp; // A non-legacy materialization must resolve through the installed @@ -827,7 +836,7 @@ impl SketchStore { Ok(instance) => instance, Err(error) => { tracing::warn!(sid, %error, "rejecting SummaryStore registration outside the active SummaryCatalog"); - return; + return false; } }; let metric_name = instance @@ -835,10 +844,9 @@ impl SketchStore { .time_series_metric() .map(str::to_owned); // Fixed lock order: instances → policy_to_series_ids → metric_to_series_ids. - let mut instances = self.instances.write().unwrap(); if self.removed_sids.read().unwrap().contains_key(&sid) { tracing::warn!(sid, "rejecting reuse of a removed summary instance ID"); - return; + return false; } let mut policy_idx = self.policy_to_series_ids.write().unwrap(); let mut metric_idx = self.metric_to_series_ids.write().unwrap(); @@ -849,6 +857,7 @@ impl SketchStore { if let Some(metric_name) = metric_name { metric_idx.entry(metric_name).or_default().insert(sid); } + true } /// Install one authoritative catalog snapshot for future registrations. @@ -2973,13 +2982,24 @@ impl SketchStore { sid: u64, agg_cfg: &asap_types::PrecomputeMaterialization, output: &crate::storage_engines::types::PrecomputedOutput, + ) -> Option> { + let mut instances = self.instances.write().ok()?; + self.register_precompute_output_with_instances(sid, agg_cfg, output, &mut instances) + } + + fn register_precompute_output_with_instances( + &self, + sid: u64, + agg_cfg: &asap_types::PrecomputeMaterialization, + output: &crate::storage_engines::types::PrecomputedOutput, + instances: &mut HashMap, ) -> Option> { let (_attrs_fp, label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output); let key_names = &agg_cfg.grouping_labels.names(); let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg); let (capability, accuracy) = agg_kind.capability_and_accuracy(); - match self.instance(sid) { + match instances.get(&sid) { None => { if self.active_catalog_generation().as_deref() != output.catalog_generation.as_deref() @@ -3009,23 +3029,28 @@ impl SketchStore { // unconditionally `None`, which meant ASAP-tier ExactAgg // state was reachable only through the legacy precompute // query path; capability-matching couldn't see it. - self.register(SketchInstanceMetadata { - sid, - metric_name: agg_cfg.metric.clone(), - group_by_keys, - capability: Some(capability), - agg_kind, - accuracy, - first_seen_unix_ms: output.start_timestamp as i64, - retired_at_ms: None, - expires_at_ms: None, - // Trust the caller's output — it carries the - // policy fingerprint computed at emit time - // (precompute worker / backfill processor). - // Falling back to `from_config(&agg_cfg)` here - // would also be correct but redundant. - policy_fp: output.policy_fp, - }); + if !self.register_with_instances( + SketchInstanceMetadata { + sid, + metric_name: agg_cfg.metric.clone(), + group_by_keys, + capability: Some(capability), + agg_kind, + accuracy, + first_seen_unix_ms: output.start_timestamp as i64, + retired_at_ms: None, + expires_at_ms: None, + // Trust the caller's output — it carries the + // policy fingerprint computed at emit time + // (precompute worker / backfill processor). + // Falling back to `from_config(&agg_cfg)` here + // would also be correct but redundant. + policy_fp: output.policy_fp, + }, + instances, + ) { + return None; + } } Some(existing) if !existing.is_writable() || existing.policy_fp != output.policy_fp => { return None; From 537bc86265b0ce0f0a8729285c20d43cbfe7d437 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 12:09:30 -0600 Subject: [PATCH 3/3] Bind immutable receipts to canonical complete input lineage --- .../precompute_engine/maintenance_runtime.rs | 151 +++++++++++++++--- 1 file changed, 129 insertions(+), 22 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 290f064a..bb8d3079 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -381,6 +381,67 @@ fn merge_inputs(inputs: &[Arc]) -> Result Result<[u8; 32], String> { + let generation = &inputs + .first() + .ok_or("immutable input cohort is empty")? + .generation; + if inputs + .iter() + .map(|input| input.definition) + .collect::>() + != expected.inputs + { + return Err("immutable lineage differs from installed input definitions".into()); + } + let mut ordered: Vec<_> = inputs.iter().collect(); + ordered.sort_by(|left, right| { + (left.definition, left.sid, &left.group).cmp(&(right.definition, right.sid, &right.group)) + }); + if ordered.windows(2).any(|pair| { + (pair[0].definition, pair[0].sid, &pair[0].group) + == (pair[1].definition, pair[1].sid, &pair[1].group) + }) { + return Err("immutable lineage repeats a physical population".into()); + } + let multiple = ordered.len() > 1; + let mut lineage = Sha256::new(); + if multiple { + lineage.update(b"immutable-maintenance-input-v2"); + lineage.update((ordered.len() as u64).to_be_bytes()); + } else { + // Preserve the existing durable single-input receipt identity. + lineage.update(b"immutable-maintenance-input-v1"); + } + for input in ordered { + if &input.generation != generation || input.windows.is_empty() { + return Err("immutable lineage has mixed generations or empty windows".into()); + } + lineage.update(input.sid.to_be_bytes()); + let metadata = + serde_json::to_vec(&(&input.definition, &input.generation, &input.group, expected)) + .map_err(|error| error.to_string())?; + if multiple { + lineage.update((metadata.len() as u64).to_be_bytes()); + } + lineage.update(metadata); + if multiple { + lineage.update((input.windows.len() as u64).to_be_bytes()); + } + for ((start, end), state) in &input.windows { + lineage.update(start.to_be_bytes()); + lineage.update(end.to_be_bytes()); + let bytes = state.serialize_to_bytes(); + lineage.update((bytes.len() as u64).to_be_bytes()); + lineage.update(bytes); + } + } + Ok(lineage.finalize().into()) +} + /// Evaluate one installed maintenance sink over an immutable physical source /// incarnation. Durable publication is a separate existing-store transaction; /// the in-memory scheduler cache here never claims durable exactly-once writes. @@ -453,28 +514,12 @@ fn prepare_frozen_maintenance_sink( if &actual_input != expected_input { return Err("immutable sink input program differs from its catalog identity".into()); } - let mut lineage = Sha256::new(); - lineage.update(b"immutable-maintenance-input-v1"); - lineage.update(input.sid.to_be_bytes()); - lineage.update( - serde_json::to_vec(&( - &input.definition, - &input.generation, - &input.group, - expected_input, - )) - .map_err(|error| error.to_string())?, - ); - let mut states = Vec::with_capacity(input.windows.len()); - for ((start, end), state) in &input.windows { - lineage.update(start.to_be_bytes()); - lineage.update(end.to_be_bytes()); - let bytes = state.serialize_to_bytes(); - lineage.update((bytes.len() as u64).to_be_bytes()); - lineage.update(&bytes); - states.push((*end as i64, Arc::clone(state))); - } - let digest: [u8; 32] = lineage.finalize().into(); + let digest = frozen_cohort_lineage(std::slice::from_ref(input), expected_input)?; + let states = input + .windows + .iter() + .map(|((_, end), state)| (*end as i64, Arc::clone(state))) + .collect(); let key = MaterializationCommitKey { plan_id: input.generation.plan_id, plan_version: input.generation.plan_version, @@ -1356,6 +1401,68 @@ mod tests { asap_types::PolicyFingerprint(value).into() } + #[test] + fn cohort_lineage_is_order_independent_and_binds_every_input() { + use crate::storage_engines::sketch_db::index::FrozenExactWindows; + let make = |sid, id, value| { + let mut state = crate::precompute_engine::operators::SumAccumulator::new(); + state.update(value); + FrozenExactWindows { + sid, + definition: definition(id), + generation: Arc::new(asap_types::sds::CatalogGeneration { + schema_version: 2, + plan_id: 1, + plan_version: 1, + snapshot_sha256: "0".repeat(64), + }), + group: BTreeMap::from([("instance".into(), sid.to_string())]), + windows: BTreeMap::from([((0, 1000), Arc::new(state) as SummaryState)]), + singleton_population_complete: false, + } + }; + let expected = asap_types::derived_input::DerivedInputIdentity { + inputs: BTreeSet::from([definition(1), definition(2)]), + program_sha256: "1".repeat(64), + }; + let baseline = + frozen_cohort_lineage(&[make(10, 1, 3.0), make(20, 2, 5.0)], &expected).unwrap(); + assert_eq!( + baseline, + frozen_cohort_lineage(&[make(20, 2, 5.0), make(10, 1, 3.0)], &expected).unwrap() + ); + assert_ne!( + baseline, + frozen_cohort_lineage(&[make(10, 1, 3.0), make(20, 2, 6.0)], &expected).unwrap() + ); + assert_ne!( + baseline, + frozen_cohort_lineage(&[make(10, 1, 3.0), make(21, 2, 5.0)], &expected).unwrap() + ); + let mut changed = make(20, 2, 5.0); + changed.group.insert("instance".into(), "other".into()); + assert_ne!( + baseline, + frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).unwrap() + ); + let mut changed = make(20, 2, 5.0); + let state = changed.windows.remove(&(0, 1000)).unwrap(); + changed.windows.insert((1000, 2000), state); + assert_ne!( + baseline, + frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).unwrap() + ); + let mut changed = make(20, 2, 5.0); + Arc::make_mut(&mut changed.generation).plan_version += 1; + assert!(frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).is_err()); + assert!(frozen_cohort_lineage(&[make(10, 1, 3.0)], &expected).is_err()); + assert!(frozen_cohort_lineage( + &[make(10, 1, 3.0), make(10, 1, 3.0), make(20, 2, 5.0)], + &expected + ) + .is_err()); + } + fn node(id: u32) -> ExecutableDagNode { ExecutableDagNode { id: PostAsapNodeId(id),