From cd9388664e8858bad4dbb605eb480f2e2b8eb810 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:13:59 -0600 Subject: [PATCH 01/16] feat(erp): define catalog-scoped population observation envelope --- crates/asap_types/src/erp_observation.rs | 79 ++++++++++++++++++++++++ crates/asap_types/src/lib.rs | 1 + 2 files changed, 80 insertions(+) create mode 100644 crates/asap_types/src/erp_observation.rs diff --git a/crates/asap_types/src/erp_observation.rs b/crates/asap_types/src/erp_observation.rs new file mode 100644 index 000000000..9bdde9b77 --- /dev/null +++ b/crates/asap_types/src/erp_observation.rs @@ -0,0 +1,79 @@ +//! Versioned runtime evidence about the inputs of one installed summary. +//! Shape is generic so the transport contract does not depend on a planner. +use crate::sds::{CatalogGeneration, SummaryDefinitionId, SummaryInstanceId}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ErpObservationInputSemantics { + ScalarSampleValue, + UnitSampleFrequency, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ErpPopulationObservation { + pub population_id: SummaryInstanceId, + pub shape: Shape, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ErpPopulationObservations { + pub schema_version: u32, + pub catalog_generation: CatalogGeneration, + pub summary_definition_id: SummaryDefinitionId, + pub observed_at_unix_ms: u64, + pub window_start_ms: i64, + pub window_end_ms: i64, + pub input_semantics: ErpObservationInputSemantics, + /// Any overflow or unsupported input invalidates the entire envelope. + /// Invalid envelopes contain no population fits. + pub invalid_reason: Option, + pub populations: Vec>, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ErpObservationFreshness { + pub max_age_ms: u64, + pub max_future_skew_ms: u64, +} + +impl ErpPopulationObservations { + pub fn validate_identity_and_freshness( + &self, + expected_generation: &CatalogGeneration, + expected_definition: SummaryDefinitionId, + now_ms: u64, + freshness: ErpObservationFreshness, + ) -> Result<(), &'static str> { + if self.schema_version != 1 + || &self.catalog_generation != expected_generation + || self.summary_definition_id != expected_definition + { + return Err("ERP observation belongs to a different catalog or summary"); + } + if self.invalid_reason.is_some() + || self.populations.is_empty() + || self.window_start_ms >= self.window_end_ms + { + return Err("ERP observation is incomplete or invalid"); + } + if freshness.max_age_ms == 0 + || self.observed_at_unix_ms > now_ms.saturating_add(freshness.max_future_skew_ms) + || now_ms.saturating_sub(self.observed_at_unix_ms) > freshness.max_age_ms + { + return Err("ERP observation is stale or future-dated"); + } + let mut seen = std::collections::BTreeSet::new(); + if self + .populations + .iter() + .any(|p| !seen.insert(&p.population_id)) + { + return Err("ERP observation repeats a population"); + } + Ok(()) + } +} diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index ccf828726..5b82e68e8 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -3,6 +3,7 @@ pub mod accuracy; pub mod aggregation_config; pub mod aggregation_type; pub mod enums; +pub mod erp_observation; pub mod executable_plan; pub mod key_by_label_names; pub mod monitor_spec; From eb304610420203317ea7afb1cf7a1748bfe98134 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:00:11 -0600 Subject: [PATCH 02/16] Validate ERP readouts against the existing state parameters --- control_plane/src/physical/erp.rs | 109 ++++++++++++++++++++++++++++-- 1 file changed, 103 insertions(+), 6 deletions(-) diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index 646162e21..a638e38a8 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -273,11 +273,11 @@ impl asap_aware_mapping::AccuracyModel for ErpAccuracyModel<'_> { _ => theoretical, }; }; - match policy.select_readout( + match policy.evidence_for_readout( kind.algorithm().clone(), readout, self.max_error, - kind.params().clone(), + kind.params(), ) { ErpParameterDecision::ExactFallback { .. } => None, ErpParameterDecision::TheoreticalFallback { .. } => theoretical, @@ -471,7 +471,7 @@ impl ErpPlanningInput { max_error: f64, theoretical: SketchParams, ) -> ErpParameterDecision { - self.select_metric(algorithm, &self.error_metric, max_error, theoretical) + self.select_metric(algorithm, &self.error_metric, max_error, theoretical, None) } pub(crate) fn select_readout( @@ -481,7 +481,31 @@ impl ErpPlanningInput { max_error: f64, theoretical: SketchParams, ) -> ErpParameterDecision { - self.select_metric(algorithm, readout.metric_key(), max_error, theoretical) + self.select_metric( + algorithm, + readout.metric_key(), + max_error, + theoretical, + None, + ) + } + + /// Validate an existing state's readout independently of which new state + /// would be cheapest to allocate for this one consumer. + fn evidence_for_readout( + &self, + algorithm: SketchAlgorithm, + readout: ReadoutEvidence, + max_error: f64, + actual: &SketchParams, + ) -> ErpParameterDecision { + self.select_metric( + algorithm, + readout.metric_key(), + max_error, + actual.clone(), + Some(actual), + ) } fn select_metric( @@ -490,6 +514,7 @@ impl ErpPlanningInput { error_metric: &str, max_error: f64, theoretical: SketchParams, + required_params: Option<&SketchParams>, ) -> ErpParameterDecision { // Runtime admissibility belongs before ranking: an unusable cheap // profile must not hide a more expensive executable alternative. @@ -497,8 +522,12 @@ impl ErpPlanningInput { artifact.records.retain(|row| { sketch_name_matches(&row.sketch, &algorithm) && parse_params(&algorithm, &row.parameters).is_some_and(|params| { - self.runtime - .supports(&algorithm, ¶ms, Some(row.resources.memory_bytes)) + required_params.is_none_or(|required| required == ¶ms) + && self.runtime.supports( + &algorithm, + ¶ms, + Some(row.resources.memory_bytes), + ) }) }); let allowed_sketches = artifact @@ -811,6 +840,74 @@ mod tests { } } + /// Structural evidence fixture, not measured calibration data. + #[test] + fn existing_univmon_readout_uses_its_own_evidence_despite_cheaper_sibling() { + use asap_aware_mapping::AccuracyModel; + use planner_types::post_asap::*; + let mut policy = input(ErpAccuracyMode::Empirical); + let small = &mut policy.artifact.records[0]; + small.sketch = "univmon".into(); + small.parameters = + serde_json::json!({"heap_size":32,"sketch_rows":5,"sketch_cols":128,"layers":4}); + small.error_metrics = + BTreeMap::from([("max_frequency_entropy_absolute_bits_error".into(), 0.1)]); + let mut large = small.clone(); + large.id = "larger-shared-state".into(); + large.parameters["heap_size"] = 128.into(); + large.resources.memory_bytes *= 2.0; + large + .error_metrics + .insert("max_frequency_entropy_absolute_bits_error".into(), 0.02); + policy.artifact.records.push(large); + let small_params = SketchParams::UnivMon { + heap_size: 32, + sketch_rows: 5, + sketch_cols: 128, + layers: 4, + }; + let actual = SketchParams::UnivMon { + heap_size: 128, + sketch_rows: 5, + sketch_cols: 128, + layers: 4, + }; + assert_eq!( + policy + .select_readout( + SketchAlgorithm::UnivMon, + ReadoutEvidence::EntropyAbsoluteBits, + 0.2, + actual.clone() + ) + .params(), + Some(&small_params) + ); + let family = SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::UnivMon, actual), + GroupingStrategy::PerSubpopulationInstance, + ); + let model = ErpAccuracyModel { + policy: Some(&policy), + max_error: 0.2, + }; + let guarantee = model + .local_guarantee(&family, &SketchQuery::FrequencyEntropy) + .unwrap(); + assert_eq!(guarantee.bound, BoundExpr::Constant { value: 0.02 }); + assert!(matches!( + guarantee.failure_probability, + ProbabilityExpr::Unknown { .. } + )); + policy.artifact.records[1].error_metrics.clear(); + assert!(ErpAccuracyModel { + policy: Some(&policy), + max_error: 0.2 + } + .local_guarantee(&family, &SketchQuery::FrequencyEntropy) + .is_none()); + } + #[test] fn hll_cardinality_uses_measured_relative_error_not_rse() { use asap_aware_mapping::AccuracyModel; From 1a7cdbcedc22c75edb1f0ac54100659a8511c43c Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:16:16 -0600 Subject: [PATCH 03/16] feat(erp): require common configuration evidence across populations --- control_plane/src/physical/compiler.rs | 2 + control_plane/src/physical/erp.rs | 73 ++++++++++++++++++++++++++ 2 files changed, 75 insertions(+) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index e4eaa85c6..8816d4842 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -4350,6 +4350,7 @@ mod tests { byte_second_weight: 1e-9, mode: super::super::erp::ErpAccuracyMode::Hybrid, observed_shape: None, + observed_populations: None, observed_shape_source: None, shape_match: None, runtime: super::super::erp::ErpRuntimeCapabilities { @@ -4399,6 +4400,7 @@ mod tests { byte_second_weight: 1e-9, mode: ErpAccuracyMode::Hybrid, observed_shape: None, + observed_populations: None, observed_shape_source: None, shape_match: None, runtime: ErpRuntimeCapabilities { diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index a638e38a8..973cf3d89 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -362,6 +362,9 @@ pub struct ErpPlanningInput { /// replaced by bounded nearest-profile matching. #[serde(default)] pub observed_shape: Option, + #[serde(default)] + pub observed_populations: + Option>, /// Runtime-samples ring key from which the backend resolves the freshest /// `erp_observed_shape` payload before compiling a plan. #[serde(default)] @@ -516,6 +519,75 @@ impl ErpPlanningInput { theoretical: SketchParams, required_params: Option<&SketchParams>, ) -> ErpParameterDecision { + if let Some(populations) = &self.observed_populations { + // Every installed partition must support the same configuration. + // Never aggregate their frequency maps into a fictitious population. + let mut best: Option = None; + if populations.invalid_reason.is_none() && !populations.populations.is_empty() { + for row in &self.artifact.records { + if !sketch_name_matches(&row.sketch, &algorithm) { + continue; + } + let Some(params) = parse_params(&algorithm, &row.parameters) else { + continue; + }; + if required_params.is_some_and(|required| required != ¶ms) { + continue; + } + let mut policy = self.clone(); + policy.observed_populations = None; + policy.mode = ErpAccuracyMode::Empirical; + let mut errors: f64 = 0.0; + let mut cost = 0.0; + let mut evidence = Vec::new(); + let valid = populations.populations.iter().all(|population| { + policy.observed_shape = Some(population.shape.observation.clone()); + match policy.select_metric( + algorithm.clone(), + error_metric, + max_error, + theoretical.clone(), + Some(¶ms), + ) { + ErpParameterDecision::Empirical { + record_id, + observed_error, + estimated_cost, + .. + } => { + errors = errors.max(observed_error); + cost += estimated_cost; + evidence.push(record_id); + true + } + _ => false, + } + }); + if valid && cost.is_finite() && best.as_ref().is_none_or(|previous| { + matches!(previous, ErpParameterDecision::Empirical { estimated_cost, .. } if cost < *estimated_cost) + }) { + best = Some(ErpParameterDecision::Empirical { + params, record_id: serde_json::to_string(&evidence).expect("string IDs serialize"), + observed_error: errors, estimated_cost: cost, + }); + } + } + } + return best.unwrap_or_else(|| { + let reason = + "ERP has no configuration valid for every observed population".to_owned(); + if self.mode == ErpAccuracyMode::Hybrid + && self.runtime.supports(&algorithm, &theoretical, None) + { + ErpParameterDecision::TheoreticalFallback { + params: theoretical, + reason, + } + } else { + ErpParameterDecision::ExactFallback { reason } + } + }); + } // Runtime admissibility belongs before ranking: an unusable cheap // profile must not hide a more expensive executable alternative. let mut artifact = self.artifact.clone(); @@ -834,6 +906,7 @@ mod tests { byte_second_weight: 1e-9, mode, observed_shape: None, + observed_populations: None, observed_shape_source: None, shape_match: None, runtime: ErpRuntimeCapabilities::default(), From 218715c86284d65efede7bb92d91d68c808074d9 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:16:54 -0600 Subject: [PATCH 04/16] fix(erp): preserve comparable population observation contracts --- crates/asap_types/src/erp_observation.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/asap_types/src/erp_observation.rs b/crates/asap_types/src/erp_observation.rs index 9bdde9b77..9f543102d 100644 --- a/crates/asap_types/src/erp_observation.rs +++ b/crates/asap_types/src/erp_observation.rs @@ -10,14 +10,14 @@ pub enum ErpObservationInputSemantics { UnitSampleFrequency, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ErpPopulationObservation { pub population_id: SummaryInstanceId, pub shape: Shape, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ErpPopulationObservations { pub schema_version: u32, From 38066e569d8fd285be79c627bfd235414521490a Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:16:54 -0600 Subject: [PATCH 05/16] fix(erp): preserve comparable population observation contracts --- crates/asap_types/src/erp_observation.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/asap_types/src/erp_observation.rs b/crates/asap_types/src/erp_observation.rs index 9bdde9b77..9f543102d 100644 --- a/crates/asap_types/src/erp_observation.rs +++ b/crates/asap_types/src/erp_observation.rs @@ -10,14 +10,14 @@ pub enum ErpObservationInputSemantics { UnitSampleFrequency, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ErpPopulationObservation { pub population_id: SummaryInstanceId, pub shape: Shape, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ErpPopulationObservations { pub schema_version: u32, From 8bd9d233341d75c7cd96cbe45bed2193f26542a9 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:24:23 -0600 Subject: [PATCH 06/16] feat(erp): retain authoritative observed data descriptor --- control_plane/src/physical/compiler.rs | 2 ++ control_plane/src/physical/erp.rs | 3 +++ 2 files changed, 5 insertions(+) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 8816d4842..1bc6ab9fd 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -4351,6 +4351,7 @@ mod tests { mode: super::super::erp::ErpAccuracyMode::Hybrid, observed_shape: None, observed_populations: None, + resolved_data_descriptor: None, observed_shape_source: None, shape_match: None, runtime: super::super::erp::ErpRuntimeCapabilities { @@ -4401,6 +4402,7 @@ mod tests { mode: ErpAccuracyMode::Hybrid, observed_shape: None, observed_populations: None, + resolved_data_descriptor: None, observed_shape_source: None, shape_match: None, runtime: ErpRuntimeCapabilities { diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index 973cf3d89..8d2b1dd8d 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -365,6 +365,8 @@ pub struct ErpPlanningInput { #[serde(default)] pub observed_populations: Option>, + #[serde(skip)] + pub resolved_data_descriptor: Option>, /// Runtime-samples ring key from which the backend resolves the freshest /// `erp_observed_shape` payload before compiling a plan. #[serde(default)] @@ -907,6 +909,7 @@ mod tests { mode, observed_shape: None, observed_populations: None, + resolved_data_descriptor: None, observed_shape_source: None, shape_match: None, runtime: ErpRuntimeCapabilities::default(), From 2d6999006f934884bcd88662e019f0ee462fe286 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:26:13 -0600 Subject: [PATCH 07/16] feat(erp): scope live evidence to catalog input semantics --- control_plane/src/physical/compiler.rs | 86 ++++++++++++++++--- crates/asap_types/src/sds.rs | 14 +++ .../storage_engines/sketch_db/index/mod.rs | 16 ++-- 3 files changed, 96 insertions(+), 20 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 1bc6ab9fd..9aa011bdc 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -2464,6 +2464,52 @@ pub fn select_workload_roots( select_workload_roots_with_erp(queries, roots, evidence, exact_costs, None) } +fn observed_population_matches_root( + policy: &super::erp::ErpPlanningInput, + root: &QueryExpr, +) -> bool { + use asap_types::sds::{PopulationPartitioning, ValueProjectionIdentity}; + use planner_types::pre_asap::{AggIntent, Reduction}; + let (Some(data), Some(observed)) = ( + &policy.resolved_data_descriptor, + &policy.observed_populations, + ) else { + return false; + }; + let QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures, + having: None, + child, + .. + } = root + else { + return false; + }; + if measures.is_empty() + || !measures.iter().all(|intent| { + matches!( + intent, + AggIntent::Cardinality { col: None, .. } + | AggIntent::FrequencyL2 { col: None, .. } + | AggIntent::FrequencyEntropy { col: None, .. } + ) + }) + { + return false; + } + let Ok((metric, Some(window), filter)) = raw_time_series_input_contract(child, false) else { + return false; + }; + data.time_series_metric() == Some(metric.as_str()) + && data.population_filter_canonical == filter + && data.value_projection == ValueProjectionIdentity::SampleValue + && data.partitioning == Some(PopulationPartitioning::PerEntity) + && data.group_by_keys.is_empty() + && observed.window_end_ms.checked_sub(observed.window_start_ms) + == i64::try_from(window.saturating_mul(1000)).ok() +} + pub fn select_workload_roots_with_erp( queries: &mut [PlanningQuery], roots: Vec>, @@ -2501,6 +2547,17 @@ pub fn select_workload_roots_with_erp( if !matches!(accuracy, AccuracyTarget::Epsilon(_)) { policy.artifact.records.clear(); } + if policy.observed_populations.is_some() + && !roots + .iter() + .all(|(_, root)| observed_population_matches_root(&policy, root)) + { + if let Some(observed) = &mut policy.observed_populations { + observed.invalid_reason = + Some("candidate input differs from observed catalog data semantics".into()); + observed.populations.clear(); + } + } // A benchmark of a different KLL implementation is not evidence // for the collector's sketchlib KLL, even with the same k. policy.artifact.records.retain(|row| { @@ -3109,22 +3166,31 @@ pub(crate) fn materialization_leaf_contract( let SummaryExpr::KeepPreAsap(expr) = &child.expr else { return Err("materialization input is not a raw source".into()); }; - let (source, window_secs) = match expr.as_ref() { + raw_time_series_input_contract( + expr, + matches!( + &node.expr, + SummaryExpr::SummaryAgg { + family: SummaryFamilyType::ExactAggregate(..), + .. + } + ), + ) +} + +fn raw_time_series_input_contract( + expr: &QueryExpr, + exact: bool, +) -> Result<(String, Option, String), String> { + use planner_types::pre_asap::{CompareOpKind, ScalarValue}; + let (source, window_secs) = match expr { QueryExpr::TimeRange { child, range } => { if range.as_millis() == 0 || range.as_millis() % 1000 != 0 { return Err("warm producer requires a positive whole-second range".into()); } (child.as_ref(), Some(range.as_secs())) } - QueryExpr::Scan { .. } - if matches!( - &node.expr, - SummaryExpr::SummaryAgg { - family: SummaryFamilyType::ExactAggregate(..), - .. - } - ) => - { + QueryExpr::Scan { .. } if exact => { return Err( "instantaneous sample selection is not a temporal accumulator readout".into(), ); diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index 24fe2a6d9..7720cf60a 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -115,6 +115,20 @@ pub struct SummaryInstanceCoordinates { pub group_values: BTreeMap, } +impl SummaryInstanceCoordinates { + pub fn instance_id(&self) -> Result { + let bytes = + serde_json::to_vec(&self.group_values).map_err(|error| SdsError(error.to_string()))?; + SummaryInstanceId::new(format!( + "summary-instance:v1:{}:{}:{}:{}", + self.summary_definition_id.as_u64(), + self.time_range.start_ms, + self.time_range.end_ms, + xxhash_rust::xxh64::xxh64(&bytes, 0) + )) + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct SummaryWindowCompletion { 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 6e383d520..60411e9bd 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -1062,16 +1062,12 @@ impl SketchStore { .map_err(|_| "summary instance start exceeds signed timestamp range")?; let end_ms = i64::try_from(window.1) .map_err(|_| "summary instance end exceeds signed timestamp range")?; - let group_bytes = - serde_json::to_vec(&group_values).map_err(|error| error.to_string())?; - let group_fingerprint = xxhash_rust::xxh64::xxh64(&group_bytes, 0); - let instance_id = SummaryInstanceId::new(format!( - "summary-instance:v1:{}:{}:{}:{}", - summary_definition_id.as_u64(), - window.0, - window.1, - group_fingerprint - )) + let instance_id = asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id, + time_range: asap_types::sds::HalfOpenTimeRange { start_ms, end_ms }, + group_values: group_values.clone(), + } + .instance_id() .map_err(|error| error.to_string())?; let instance = SummaryInstance { instance_id: instance_id.clone(), From a503d10c93ad181bb3efa7e8404b2f3e7ec8d7a6 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:31:51 -0600 Subject: [PATCH 08/16] feat(erp): observe bounded precompute inputs and publish after finite drain --- control_plane/src/physical/erp.rs | 4 + .../drivers/ingest/prometheus_remote_write.rs | 7 + data_plane/src/main.rs | 10 + data_plane/src/precompute_engine/engine.rs | 3 +- .../src/precompute_engine/erp_observer.rs | 223 ++++++++++++++++++ data_plane/src/precompute_engine/mod.rs | 1 + .../src/precompute_engine/series_router.rs | 11 + data_plane/src/precompute_engine/worker.rs | 43 ++++ 8 files changed, 301 insertions(+), 1 deletion(-) create mode 100644 data_plane/src/precompute_engine/erp_observer.rs diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index 8d2b1dd8d..646a7f010 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -31,6 +31,10 @@ pub struct ErpShapeObserver { } impl ErpShapeObserver { + pub fn observed_key_count(&self) -> usize { + self.frequencies.len() + } + pub fn new(max_observed_keys: usize) -> Result { Self::with_limits(max_observed_keys, 256) } diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 2841bf626..7deb61e22 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -201,6 +201,13 @@ impl PrometheusRemoteWriteReceiver { .ingest .sketch_index .seal_finite_summary_input(&generation)?; + if let Some(observer) = self.inner.ingest.router.erp_observer() { + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|error| error.to_string())? + .as_millis() as u64; + observer.publish_finite(&generation, now_ms).await?; + } trim_process_allocator(); Ok(()) } diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 345479b29..208b7db50 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -44,6 +44,9 @@ fn unix_time_ms() -> u64 { #[derive(Parser, Debug)] #[command(author, version, about, long_about = None)] struct Args { + /// Publish bounded materialization-input ERP observations after a verified finite-source drain. + #[arg(long)] + erp_runtime_samples_endpoint: Option, /// Runtime component profile. `asapquery` enables backend ingest-time /// materialization and rejects Collector/OTLP-only components. #[arg(long, value_enum, default_value = "distributed")] @@ -874,6 +877,13 @@ async fn main() -> Result<()> { series_resolver.clone(), sketch_index.clone(), ); + if let Some(endpoint) = args.erp_runtime_samples_endpoint.clone() { + engine + .ingest_state() + .router + .enable_erp_observation(endpoint) + .map_err(std::io::Error::other)?; + } let worker_diagnostics = engine.diagnostics(); let ingest_state = engine.ingest_state(); info!("Starting precompute engine (ingest adapters share its bounded worker queues)"); diff --git a/data_plane/src/precompute_engine/engine.rs b/data_plane/src/precompute_engine/engine.rs index 15f44c40f..8d59efe94 100644 --- a/data_plane/src/precompute_engine/engine.rs +++ b/data_plane/src/precompute_engine/engine.rs @@ -129,7 +129,7 @@ impl PrecomputeEngine { // ConfigReload messages needed. let mut worker_handles = Vec::with_capacity(num_workers); for (id, rx) in receivers.into_iter().enumerate() { - let worker = Worker::new( + let mut worker = Worker::new( id, rx, output_sink.clone(), @@ -148,6 +148,7 @@ impl PrecomputeEngine { self.diagnostics.worker_group_counts[id].clone(), self.diagnostics.worker_watermarks[id].clone(), ); + worker.set_erp_observer(self.ingest_state.router.erp_observer()); let handle = tokio::spawn(async move { worker.run().await; }); diff --git a/data_plane/src/precompute_engine/erp_observer.rs b/data_plane/src/precompute_engine/erp_observer.rs new file mode 100644 index 000000000..4c05aa617 --- /dev/null +++ b/data_plane/src/precompute_engine/erp_observer.rs @@ -0,0 +1,223 @@ +//! Bounded observations of actual materialization inputs, published only after +//! an explicit finite-source completion barrier. No worker timestamp is a seal. +use asap_types::erp_observation::*; +use asap_types::sds::*; +use control_plane::physical::erp::{ErpObservedShape, ErpShapeObserver}; +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +const MAX_POPULATIONS: usize = 128; +const MAX_KEYS: usize = 65_536; + +struct Population { + coordinates: SummaryInstanceCoordinates, + semantics: ErpObservationInputSemantics, + source: String, + sketch: String, + implementation: String, + observer: ErpShapeObserver, +} +#[derive(Default)] +struct Observations { + generation: Option, + populations: BTreeMap, + extent: BTreeMap, + total_keys: usize, + invalid: Option, +} + +pub struct RuntimeErpObserver { + endpoint: String, + observations: Mutex, +} +impl RuntimeErpObserver { + pub fn new(endpoint: String) -> Arc { + Arc::new(Self { + endpoint, + observations: Mutex::new(Observations::default()), + }) + } + pub fn observe( + &self, + generation: &CatalogGeneration, + coordinates: SummaryInstanceCoordinates, + config: &asap_types::AggregationConfig, + timestamp_ms: i64, + value: f64, + ) { + use asap_types::{AggregationType, SampleUpdateRule}; + let (semantics, sketch, implementation) = match config.aggregation_type { + AggregationType::HLL => ( + ErpObservationInputSemantics::ScalarSampleValue, + "hll", + "asap-sketchlib-hll-regular-v1", + ), + AggregationType::UnivMon => ( + ErpObservationInputSemantics::UnitSampleFrequency, + "univmon", + "asap-sketchlib-univmon-standard-v1", + ), + _ => return, + }; + let mut state = self.observations.lock().unwrap(); + if state.generation.as_ref() != Some(generation) { + *state = Observations { + generation: Some(generation.clone()), + ..Default::default() + }; + } + if state.invalid.is_some() { + return; + } + if !value.is_finite() + || !matches!( + config.sample_update_rule(), + SampleUpdateRule::Value { scale: 1.0 } + ) + { + state.invalid = Some("unsupported non-finite or transformed summary input".into()); + for population in state.populations.values_mut() { + population.observer = ErpShapeObserver::new(1).unwrap(); + } + return; + } + let extent = state + .extent + .entry(coordinates.summary_definition_id) + .or_insert((timestamp_ms, timestamp_ms)); + extent.0 = extent.0.min(timestamp_ms); + extent.1 = extent.1.max(timestamp_ms); + let Ok(id) = coordinates.instance_id() else { + return; + }; + if !state.populations.contains_key(&id) && state.populations.len() >= MAX_POPULATIONS { + state.invalid = Some("ERP population observation budget exceeded".into()); + for population in state.populations.values_mut() { + population.observer = ErpShapeObserver::new(1).unwrap(); + } + return; + } + let population = state.populations.entry(id).or_insert_with(|| Population { + source: format!( + "summary-definition:{}", + coordinates.summary_definition_id.as_u64() + ), + coordinates, + semantics, + sketch: sketch.into(), + implementation: implementation.into(), + observer: ErpShapeObserver::new(MAX_KEYS).unwrap(), + }); + let before = population.observer.observed_key_count(); + // Canonical fixed-width numeric identity; no raw sample or arbitrary label copy. + let key = format!("{:016x}", if value == 0.0 { 0 } else { value.to_bits() }); + let result = population.observer.observe(&key, 0); + let added = population + .observer + .observed_key_count() + .saturating_sub(before); + state.total_keys = state.total_keys.saturating_add(added); + if result.is_err() || state.total_keys > MAX_KEYS { + state.invalid = Some("ERP key observation budget exceeded".into()); + for population in state.populations.values_mut() { + population.observer = ErpShapeObserver::new(1).unwrap(); + } + } + } + + pub async fn publish_finite( + &self, + generation: &CatalogGeneration, + now_ms: u64, + ) -> Result<(), String> { + use control_plane::runtime_samples::feedback::{ + runtime_samples_client::RuntimeSamplesClient, PushBatch, RuntimeRecord, + }; + let records = { + let state = self.observations.lock().unwrap(); + if state.generation.as_ref() != Some(generation) { + return Ok(()); + } + let mut groups: BTreeMap< + (SummaryDefinitionId, i64, i64), + ( + String, + String, + String, + ErpPopulationObservations, + ), + > = BTreeMap::new(); + for (id, population) in &state.populations { + let c = &population.coordinates; + let Some((first, last)) = state.extent.get(&c.summary_definition_id) else { + continue; + }; + if c.time_range.start_ms < *first || c.time_range.end_ms > *last { + continue; + } + let key = ( + c.summary_definition_id, + c.time_range.start_ms, + c.time_range.end_ms, + ); + let entry = groups.entry(key).or_insert_with(|| { + ( + population.source.clone(), + population.sketch.clone(), + population.implementation.clone(), + ErpPopulationObservations { + schema_version: 1, + catalog_generation: generation.clone(), + summary_definition_id: c.summary_definition_id, + observed_at_unix_ms: now_ms, + window_start_ms: c.time_range.start_ms, + window_end_ms: c.time_range.end_ms, + input_semantics: population.semantics, + invalid_reason: state.invalid.clone(), + populations: Vec::new(), + }, + ) + }); + match population.observer.snapshot() { + Some(shape) => entry.3.populations.push(ErpPopulationObservation { + population_id: id.clone(), + shape, + }), + None => entry.3.invalid_reason = Some("ERP population fit unavailable".into()), + } + } + groups + .into_values() + .map(|(source, sketch, impl_name, mut envelope)| { + if envelope.invalid_reason.is_some() { + envelope.populations.clear(); + } + RuntimeRecord { + source, + sketch, + impl_name, + schema_version: 1, + payload_json: serde_json::json!({"erp_population_observations":envelope}) + .to_string(), + } + }) + .collect::>() + }; + if records.is_empty() { + return Ok(()); + } + let expected = records.len() as u64; + let mut client = RuntimeSamplesClient::connect(self.endpoint.clone()) + .await + .map_err(|e| e.to_string())?; + let ack = client + .push(PushBatch { records }) + .await + .map_err(|e| e.to_string())? + .into_inner(); + if ack.accepted != expected { + return Err("control plane rejected ERP observation records".into()); + } + Ok(()) + } +} diff --git a/data_plane/src/precompute_engine/mod.rs b/data_plane/src/precompute_engine/mod.rs index c10b4bb1f..068ac13b7 100644 --- a/data_plane/src/precompute_engine/mod.rs +++ b/data_plane/src/precompute_engine/mod.rs @@ -2,6 +2,7 @@ pub mod accumulator_factory; pub mod config; pub mod coordination_checkpoint; mod engine; +pub mod erp_observer; pub mod frame_lineage; pub mod group_key; pub mod ingest_handler; diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs index b0de9df62..3a90705d8 100644 --- a/data_plane/src/precompute_engine/series_router.rs +++ b/data_plane/src/precompute_engine/series_router.rs @@ -148,6 +148,7 @@ impl fmt::Debug for WorkerMessage { /// Routes incoming samples to one of N workers based on a consistent hash. pub struct SeriesRouter { + erp_observer: std::sync::OnceLock>, senders: Vec>, num_workers: usize, } @@ -156,11 +157,21 @@ impl SeriesRouter { pub fn new(senders: Vec>) -> Self { let num_workers = senders.len(); Self { + erp_observer: std::sync::OnceLock::new(), senders, num_workers, } } + pub fn enable_erp_observation(&self, endpoint: String) -> Result<(), String> { + self.erp_observer + .set(super::erp_observer::RuntimeErpObserver::new(endpoint)) + .map_err(|_| "ERP observer already configured".into()) + } + pub fn erp_observer(&self) -> Option> { + self.erp_observer.get().cloned() + } + /// Route a pre-grouped batch of group messages to workers concurrently. /// /// Each `GroupSamples` / `AccumulatorInput` message is routed by diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 6fdbeed50..845c72d71 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -148,6 +148,7 @@ pub struct WorkerRuntimeConfig { /// `(metric, attrs_fingerprint, agg_kind_canonical)` identity contract on /// `SeriesIdResolver`, so one sid uniquely names one bucket. pub struct Worker { + erp_observer: Option>, current_input_revision: Option>, id: usize, receiver: mpsc::Receiver, @@ -184,6 +185,12 @@ pub struct Worker { } impl Worker { + pub fn set_erp_observer( + &mut self, + observer: Option>, + ) { + self.erp_observer = observer; + } #[allow(clippy::too_many_arguments)] pub fn new( id: usize, @@ -204,6 +211,7 @@ impl Worker { wall_clock_max_open_grace_period_ms, } = runtime_config; Self { + erp_observer: None, current_input_revision: None, id, receiver, @@ -574,6 +582,24 @@ impl Worker { record_late_input("append_correction", "raw_sample"); let mut updater = create_accumulator_updater(&state.config); apply_sample(&mut *updater, series_key, *val, *ts, &state.config); + if let (Some(observer), Some(revision)) = + (&self.erp_observer, &input_revision) + { + observer.observe( + &revision.generation, + asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: policy_fp.into(), + time_range: asap_types::sds::HalfOpenTimeRange { + start_ms: bucket_start, + end_ms: bucket_end, + }, + group_values: group_key.as_population_labels(), + }, + &state.config, + *ts, + *val, + ); + } let key = build_group_key_label_values(group_key); let output = precomputed_output_for_group( window_start as u64, @@ -604,6 +630,23 @@ impl Worker { .or_insert_with(|| create_accumulator_updater(&state.config)); if let Some(value) = value { apply_sample(&mut **updater, series_key, value, *ts, &state.config); + if let (Some(observer), Some(revision)) = (&self.erp_observer, &input_revision) + { + observer.observe( + &revision.generation, + asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: policy_fp.into(), + time_range: asap_types::sds::HalfOpenTimeRange { + start_ms: bucket_start, + end_ms: bucket_end, + }, + group_values: group_key.as_population_labels(), + }, + &state.config, + *ts, + value, + ); + } } } } From 1826bc8d29e822fe3ed2628bb843c8af6cbd79f2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:32:28 -0600 Subject: [PATCH 09/16] feat(erp): validate live evidence against the activated catalog --- control_plane/src/main.rs | 87 ++++- control_plane/src/physical/erp.rs | 299 +++++++++++++++++- crates/asap_types/src/erp_observation.rs | 2 +- .../summary-catalog-sds-architecture.md | 32 ++ 4 files changed, 411 insertions(+), 9 deletions(-) diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index a79236381..dbf63519d 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -69,6 +69,10 @@ struct AppState { /// agents' `sketch-runtime::PushExporter`. Read by decision /// loops in the replanner. runtime_samples: Arc, + /// The successfully activated typed catalog is authoritative for live ERP + /// input identity; incoming telemetry cannot supply its own descriptors. + active_summary_catalog: + Arc>>>, /// Phase C (MVP v6): shared `BackendClient` for posting /// `StreamingConfig` JSON / YAML to the ASAPQuery-backend's /// `POST /api/v1/streaming-config` endpoint. Phase B had this @@ -487,6 +491,7 @@ async fn main() { opamp_endpoint: opamp_ep, workload_registry: Arc::clone(&workload_registry), runtime_samples: Arc::clone(&runtime_samples_store), + active_summary_catalog: Arc::new(tokio::sync::Mutex::new(None)), backend_client: backend_client_shared, backend_routing_cache: Arc::clone(&backend_routing_cache), }; @@ -611,6 +616,8 @@ struct PhysicalPlanQueryRequest { #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct CompileAndPublishPhysicalPlanRequest { + #[serde(default = "default_physical_deployment_target")] + target: physical::compiler::PhysicalDeploymentTarget, #[serde(default)] workload_cost_evidence: Option, queries: Vec, @@ -635,6 +642,10 @@ struct CompileAndPublishPhysicalPlanRequest { apply_timeout_ms: u64, } +fn default_physical_deployment_target() -> physical::compiler::PhysicalDeploymentTarget { + physical::compiler::PhysicalDeploymentTarget::DistributedCollectors +} + fn default_physical_plan_timeout_ms() -> u64 { 10_000 } @@ -704,10 +715,15 @@ async fn compile_and_publish_physical_plan( mut request: CompileAndPublishPhysicalPlanRequest, frontend: PhysicalQueryFrontend, ) -> Response { + // Serialize typed activations so an older response cannot overwrite the + // catalog recorded after a newer backend activation. + let mut active_catalog = st.active_summary_catalog.lock().await; if let Some(erp) = &mut request.erp { if let Err(error) = erp.hydrate_observed_shape(&st.runtime_samples) { return (StatusCode::UNPROCESSABLE_ENTITY, error).into_response(); } + let catalog = active_catalog.clone(); + erp.resolve_population_data_descriptor(catalog.as_deref()); } let (bundle, collector_ids, apply_timeout, adaptation_evidence, _) = match compile_physical_plan_request(request, false, frontend) { @@ -802,6 +818,8 @@ async fn compile_and_publish_physical_plan( .into_response(); } + *active_catalog = Some(Arc::new(bundle.summary_catalog)); + Json(CompileAndPublishPhysicalPlanResponse { cost_comparison: bundle.cost_comparison, plan_id: bundle.envelope.plan_id, @@ -842,6 +860,7 @@ async fn publish_clickhouse_plan( publication: physical::publication::PhysicalPlanPublication, selection_trace: Option, ) -> axum::response::Response { + let mut active_catalog = state.active_summary_catalog.lock().await; let plan_id = publication.summary_catalog.plan_id; let plan_version = publication.summary_catalog.plan_version; let Some(client) = state.backend_client.as_ref() else { @@ -863,6 +882,7 @@ async fn publish_clickhouse_plan( .await; return (StatusCode::BAD_GATEWAY, error.to_string()).into_response(); } + *active_catalog = Some(Arc::new(publication.summary_catalog)); Json(serde_json::json!({ "plan_id": plan_id, "plan_version": plan_version, @@ -888,10 +908,15 @@ fn compile_physical_plan_request( ), (StatusCode, String), > { - if request.queries.is_empty() || request.collector_ids.is_empty() { + if request.queries.is_empty() + || (request.target == physical::compiler::PhysicalDeploymentTarget::DistributedCollectors + && request.collector_ids.is_empty()) + || (request.target == physical::compiler::PhysicalDeploymentTarget::BackendLocalRemoteWrite + && !request.collector_ids.is_empty()) + { return Err(( StatusCode::UNPROCESSABLE_ENTITY, - "queries and collector_ids must both be non-empty".to_string(), + "queries must be non-empty; distributed deployment requires collectors and backend-local deployment requires none".to_string(), )); } if request.max_evidence_age_ms == 0 || request.apply_timeout_ms == 0 { @@ -967,7 +992,8 @@ fn compile_physical_plan_request( let planning_request = physical::compiler::PlanningRequest { query_workload: None, queries, - hybrid_execution: false, + hybrid_execution: request.target + == physical::compiler::PhysicalDeploymentTarget::BackendLocalRemoteWrite, materialization_policy: None, evidence: request.evidence, exact_composition_costs: request.exact_composition_costs, @@ -978,7 +1004,7 @@ fn compile_physical_plan_request( retained_summary_memory_budget_bytes: None, }; let environment = physical::compiler::DeploymentEnvironment { - target: physical::compiler::PhysicalDeploymentTarget::DistributedCollectors, + target: request.target, collector_ids: request.collector_ids.clone(), capability_snapshot_id: request.capability_snapshot_id, observed_at_unix_ms: now, @@ -2242,6 +2268,7 @@ fn test_app_with_backend(backend_url: Option) -> (AppState, axum::Router opamp_endpoint: "ws://ctrl:4320/v1/opamp".into(), workload_registry: Arc::new(WorkloadRegistry::empty()), runtime_samples: runtime_samples::RuntimeSamplesStore::new(64), + active_summary_catalog: Arc::new(tokio::sync::Mutex::new(None)), backend_client, backend_routing_cache: Arc::new(Mutex::new(HashMap::new())), }; @@ -2309,6 +2336,58 @@ mod api_tests { assert_eq!(manifests.as_array().unwrap().len(), 1); } + #[test] + fn backend_local_typed_request_compiles_without_collectors() { + let snapshot: physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_str( + include_str!("../../docs/examples/asapquery-compatibility-demo-snapshot.json"), + ) + .unwrap(); + let (planning, _) = snapshot.planning_request().unwrap(); + let mut query = planning.queries[0].clone(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + query.lifecycle.evidence_observed_at_unix_ms = now; + for implementation in &mut query.window_implementations { + implementation.cost.observed_at_unix_ms = now; + } + let planner_types::pre_asap::Source::TimeSeries { metric } = &query.source else { + panic!("expected time series fixture"); + }; + let value = serde_json::json!({ + "target": "backend_local_remote_write", + "queries": [{ + "query_id": query.query_id, "query_string": query.query_string, + "metric": metric, "window_secs": query.window_secs, "accuracy": query.accuracy, + "lifecycle": query.lifecycle, "window_implementations": query.window_implementations + }], + "collector_ids": [], "capability_snapshot_id": "test", + "planner_revision": physical::compiler::PLANNER_REVISION, + "max_evidence_age_ms": 60000, "plan_version": 1, + "activation_unix_ms": now, "backend_compat": physical::compiler::BACKEND_COMPAT + }); + let request = serde_json::from_value(value.clone()).unwrap(); + let (plan, collectors, _, _, _) = + compile_physical_plan_request(request, false, PhysicalQueryFrontend::PromQl).unwrap(); + let plan = plan.unwrap(); + assert!(collectors.is_empty()); + assert!(plan.collector_plans.is_empty()); + assert_eq!( + plan.precompute_plan.ingest.protocol, + physical::compiler::IngestProtocol::PrometheusRemoteWriteV1 + ); + assert!(!plan.precompute_plan.materializations.is_empty()); + let mut distributed = value; + distributed["target"] = serde_json::json!("distributed_collectors"); + assert!(compile_physical_plan_request( + serde_json::from_value(distributed).unwrap(), + false, + PhysicalQueryFrontend::PromQl + ) + .is_err()); + } + async fn body_json(resp: axum::response::Response) -> serde_json::Value { let bytes = resp.into_body().collect().await.unwrap().to_bytes(); serde_json::from_slice(&bytes).unwrap() diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index 8d2b1dd8d..a57ddbc05 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -383,25 +383,134 @@ pub struct ErpObservedShapeSource { pub source: String, pub sketch: String, pub implementation: String, + /// Required for catalog-scoped online evidence; absent only for legacy + /// offline runtime records that contain a single shape. + #[serde(default)] + pub population_scope: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct ErpPopulationObservationScope { + pub catalog_generation: asap_types::sds::CatalogGeneration, + pub summary_definition_id: asap_types::sds::SummaryDefinitionId, + pub input_semantics: asap_types::erp_observation::ErpObservationInputSemantics, + pub freshness: asap_types::erp_observation::ErpObservationFreshness, } impl ErpPlanningInput { + /// Resolve evidence against the control plane's accepted catalog, never + /// descriptors supplied by the observation producer. + pub fn resolve_population_data_descriptor( + &mut self, + catalog: Option<&asap_types::summary_catalog::SummaryCatalog>, + ) { + self.resolved_data_descriptor = None; + let Some(populations) = self.observed_populations.as_mut() else { + return; + }; + if populations.invalid_reason.is_some() { + return; + } + let resolved = (|| -> Result<_, String> { + let catalog = catalog.ok_or("no active authoritative catalog for ERP evidence")?; + let generation = catalog.reference().map_err(|error| error.to_string())?; + if generation != populations.catalog_generation { + return Err("ERP evidence catalog differs from the active catalog".into()); + } + let materialization = catalog + .materializations + .get(&populations.summary_definition_id) + .ok_or("ERP evidence summary is absent from the active catalog")?; + let data = catalog + .data_descriptors + .get(&materialization.data_descriptor_id) + .ok_or("ERP evidence data descriptor is absent from the active catalog")?; + Ok(std::sync::Arc::new(data.clone())) + })(); + match resolved { + Ok(data) => self.resolved_data_descriptor = Some(data), + Err(reason) => { + populations.invalid_reason = Some(reason); + populations.populations.clear(); + } + } + } + pub fn hydrate_observed_shape( &mut self, samples: &crate::runtime_samples::RuntimeSamplesStore, ) -> Result<(), String> { - if self.observed_shape.is_some() { - return Ok(()); - } + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|error| format!("cannot timestamp ERP observation: {error}"))? + .as_millis() + .try_into() + .map_err(|_| "ERP observation timestamp overflows milliseconds")?; + self.hydrate_observed_shape_at(samples, now_ms) + } + + pub fn hydrate_observed_shape_at( + &mut self, + samples: &crate::runtime_samples::RuntimeSamplesStore, + now_ms: u64, + ) -> Result<(), String> { let Some(source) = &self.observed_shape_source else { return Ok(()); }; + // A reused planning request must resolve new evidence on every compile. + // Never keep an old inline or previously hydrated fit after drift. + self.observed_shape = None; + self.observed_populations = None; let key = crate::runtime_samples::SampleKey { source: source.source.clone(), sketch: source.sketch.clone(), impl_name: source.implementation.clone(), }; - let record = samples.latest(&key).ok_or_else(|| { + let record = samples.latest(&key); + if let Some(scope) = &source.population_scope { + let resolved = (|| -> Result<_, String> { + let record = record + .as_ref() + .ok_or("no live ERP population observation")?; + let value = record + .payload + .get("erp_population_observations") + .ok_or("latest runtime sample has no erp_population_observations")?; + let observed: asap_types::erp_observation::ErpPopulationObservations< + ErpObservedShape, + > = serde_json::from_value(value.clone()) + .map_err(|error| format!("invalid ERP population observations: {error}"))?; + observed.validate_identity_and_freshness( + &scope.catalog_generation, + scope.summary_definition_id, + now_ms, + scope.freshness, + )?; + if observed.input_semantics != scope.input_semantics { + return Err("ERP observation input semantics do not match the summary".into()); + } + Ok(observed) + })(); + // An explicit invalid envelope prevents selection from borrowing an + // older fit or falling back to the artifact's distribution identity. + // The normal ERP -> theoretical -> exact policy handles this miss. + self.observed_populations = Some(resolved.unwrap_or_else(|reason| { + asap_types::erp_observation::ErpPopulationObservations { + schema_version: 1, + catalog_generation: scope.catalog_generation.clone(), + summary_definition_id: scope.summary_definition_id, + observed_at_unix_ms: now_ms, + window_start_ms: 0, + window_end_ms: 0, + input_semantics: scope.input_semantics, + invalid_reason: Some(reason), + populations: Vec::new(), + } + })); + return Ok(()); + } + let record = record.ok_or_else(|| { format!( "no runtime shape sample for {}/{}/{}", source.source, source.sketch, source.implementation @@ -1211,11 +1320,193 @@ mod tests { source: "edge-a".into(), sketch: "cms".into(), implementation: "oxide".into(), + population_scope: None, }); policy.hydrate_observed_shape(&samples).unwrap(); assert_eq!(policy.observed_shape.unwrap().cardinality, 1000); } + fn online_population_fixture() -> ( + ErpPlanningInput, + asap_types::erp_observation::ErpPopulationObservations, + ) { + use asap_types::erp_observation::*; + let generation = asap_types::sds::CatalogGeneration { + schema_version: 1, + plan_id: 7, + plan_version: 3, + snapshot_sha256: "test-catalog".into(), + }; + let definition = asap_types::PolicyFingerprint(7).into(); + let mut observer = ErpShapeObserver::new(4).unwrap(); + for key in ["a", "a", "b", "b"] { + observer.observe(key, 0).unwrap(); + } + let observed = ErpPopulationObservations { + schema_version: 1, + catalog_generation: generation.clone(), + summary_definition_id: definition, + observed_at_unix_ms: 1_000, + window_start_ms: 0, + window_end_ms: 1_000, + input_semantics: ErpObservationInputSemantics::UnitSampleFrequency, + invalid_reason: None, + populations: vec![ErpPopulationObservation { + population_id: asap_types::sds::SummaryInstanceId::new("partition-a").unwrap(), + shape: observer.snapshot().unwrap(), + }], + }; + let mut policy = input(ErpAccuracyMode::Hybrid); + policy.observed_shape_source = Some(ErpObservedShapeSource { + source: "backend-a".into(), + sketch: "univmon".into(), + implementation: "asap_sketchlib".into(), + population_scope: Some(ErpPopulationObservationScope { + catalog_generation: generation, + summary_definition_id: definition, + input_semantics: ErpObservationInputSemantics::UnitSampleFrequency, + freshness: ErpObservationFreshness { + max_age_ms: 100, + max_future_skew_ms: 5, + }, + }), + }); + (policy, observed) + } + + fn publish_population_fixture( + samples: &crate::runtime_samples::RuntimeSamplesStore, + payload: serde_json::Value, + ) { + samples.append_for_test(crate::runtime_samples::RuntimeRecord { + source: "backend-a".into(), + sketch: "univmon".into(), + impl_name: "asap_sketchlib".into(), + schema_version: 1, + payload, + }); + } + + /// New live evidence replaces hydrated fits; expiry invalidates them even + /// when a caller reuses the same planning request. + #[test] + fn online_hydration_refreshes_and_expires_population_evidence() { + let (mut policy, mut observed) = online_population_fixture(); + let samples = crate::runtime_samples::RuntimeSamplesStore::new(4); + publish_population_fixture( + &samples, + serde_json::json!({"erp_population_observations": observed}), + ); + policy.hydrate_observed_shape_at(&samples, 1_000).unwrap(); + assert_eq!( + policy.observed_populations.as_ref().unwrap().populations[0] + .shape + .observation + .cardinality, + 2 + ); + observed.populations[0].shape.observation.cardinality = 3; + observed.observed_at_unix_ms = 1_010; + publish_population_fixture( + &samples, + serde_json::json!({"erp_population_observations": observed}), + ); + policy.hydrate_observed_shape_at(&samples, 1_010).unwrap(); + assert_eq!( + policy.observed_populations.as_ref().unwrap().populations[0] + .shape + .observation + .cardinality, + 3 + ); + policy.hydrate_observed_shape_at(&samples, 1_111).unwrap(); + let invalid = policy.observed_populations.as_ref().unwrap(); + assert!(invalid.populations.is_empty()); + assert!(invalid.invalid_reason.as_deref().unwrap().contains("stale")); + assert!(policy.observed_shape.is_none()); + } + + /// Only the activated catalog may supply a candidate's data contract. + #[test] + fn online_population_descriptor_requires_authoritative_catalog() { + let snapshot: crate::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let plan = snapshot.compile().unwrap(); + let (mut policy, mut observed) = online_population_fixture(); + observed.catalog_generation = plan.summary_catalog.reference().unwrap(); + observed.summary_definition_id = + *plan.summary_catalog.materializations.keys().next().unwrap(); + policy.observed_populations = Some(observed.clone()); + policy.resolve_population_data_descriptor(Some(&plan.summary_catalog)); + let expected = &plan.summary_catalog.materializations[&observed.summary_definition_id] + .data_descriptor_id; + assert_eq!( + &policy.resolved_data_descriptor.as_ref().unwrap().id, + expected + ); + policy.resolve_population_data_descriptor(None); + assert!(policy.resolved_data_descriptor.is_none()); + assert!(policy + .observed_populations + .as_ref() + .unwrap() + .invalid_reason + .is_some()); + assert!(policy + .observed_populations + .as_ref() + .unwrap() + .populations + .is_empty()); + } + + /// Missing, corrupt, foreign and overflow evidence must never recover a + /// previously accepted fit or the artifact's legacy distribution match. + #[test] + fn online_hydration_invalidates_unusable_latest_evidence() { + let (template, observed) = online_population_fixture(); + let mut foreign = observed.clone(); + foreign.catalog_generation.plan_version += 1; + let mut wrong_input = observed.clone(); + wrong_input.input_semantics = + asap_types::erp_observation::ErpObservationInputSemantics::ScalarSampleValue; + let mut overflow = observed.clone(); + overflow.invalid_reason = Some("population limit".into()); + let mut future = observed.clone(); + future.observed_at_unix_ms = 1_006; + for payload in [ + None, + Some(serde_json::json!({})), + Some(serde_json::json!({"erp_population_observations": foreign})), + Some(serde_json::json!({"erp_population_observations": wrong_input})), + Some(serde_json::json!({"erp_population_observations": overflow})), + Some(serde_json::json!({"erp_population_observations": future})), + ] { + let mut policy = template.clone(); + policy.observed_shape = Some(observed.populations[0].shape.observation.clone()); + let samples = crate::runtime_samples::RuntimeSamplesStore::new(4); + if let Some(payload) = payload { + publish_population_fixture(&samples, payload); + } + policy.hydrate_observed_shape_at(&samples, 1_000).unwrap(); + let invalid = policy.observed_populations.as_ref().unwrap(); + assert!(invalid.invalid_reason.is_some()); + assert!(invalid.populations.is_empty()); + assert!(policy.observed_shape.is_none()); + assert!(matches!( + policy.select( + SketchAlgorithm::Hll, + 0.05, + SketchParams::Hll { precision: 12 } + ), + ErpParameterDecision::TheoreticalFallback { .. } + )); + } + } + #[test] fn observer_fails_loud_at_cardinality_cap() { let mut observer = ErpShapeObserver::new(1).unwrap(); diff --git a/crates/asap_types/src/erp_observation.rs b/crates/asap_types/src/erp_observation.rs index 9f543102d..efb693c05 100644 --- a/crates/asap_types/src/erp_observation.rs +++ b/crates/asap_types/src/erp_observation.rs @@ -33,7 +33,7 @@ pub struct ErpPopulationObservations { pub populations: Vec>, } -#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ErpObservationFreshness { pub max_age_ms: u64, diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index baf28e179..4cf38d01c 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -375,3 +375,35 @@ creates a new Data Descriptor. Advancing the time range creates a new Summary Instance. Merge compatibility additionally requires the operator's merge rules, compatible data scopes and valid instance coverage; sharing descriptors alone does not authorize merging overlapping observations. + +### Catalog-scoped runtime ERP evidence + +A runtime observation describes the input of one allocated summary, not an +entire deployment. `ErpPopulationObservations` identifies its catalog generation, +summary definition, observation time, input window and separate summary-instance +populations. The control plane resolves the `DataDescriptor` from its successfully +activated catalog; a telemetry payload cannot provide replacement descriptors. +Alternative sketch parameters may use this evidence only when the compiler +verifies the same data and update semantics. + +The typed physical-plan HTTP endpoints accept `target: backend_local_remote_write` +with an empty `collector_ids` list. Omitting `target` preserves the distributed +collector deployment. Both paths use catalog publication and activation. Typed +activations are serialized, and the accepted catalog is retained only after the +backend acknowledges activation, including ClickHouse publications. + +An ERP `observed_shape_source.population_scope` supplies the expected catalog +and definition, input semantics, and explicit `max_age_ms` / +`max_future_skew_ms` bounds. Each compilation reads the latest runtime record +again. Missing, stale, malformed, foreign or incomplete observations invalidate +all population fits. This is an ERP miss handled by theoretical sizing or exact +execution; it must not restore an older fit or match the artifact's legacy +distribution descriptor. Offline single-shape inputs remain a separate path. + +The initial eligibility is deliberately limited to verified raw per-series +frequency/cardinality readouts over a complete matching window. A 30-second pane +observation does not certify a one-hour input distribution. These checks do not +implement an autonomous drift-triggered replan scheduler, continuous source +completion, or durable restoration of the control plane's active catalog. After +a control-plane restart, live evidence remains ineligible until an authoritative +catalog has been activated again. From c13ba4f00c93f0eb8e32040d593acec8fc4c15da Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:33:06 -0600 Subject: [PATCH 10/16] test(erp): cover bounded partition observations and catalog reset --- .../src/precompute_engine/erp_observer.rs | 89 ++++++++++++++++++- 1 file changed, 88 insertions(+), 1 deletion(-) diff --git a/data_plane/src/precompute_engine/erp_observer.rs b/data_plane/src/precompute_engine/erp_observer.rs index 4c05aa617..e8bc37485 100644 --- a/data_plane/src/precompute_engine/erp_observer.rs +++ b/data_plane/src/precompute_engine/erp_observer.rs @@ -111,7 +111,13 @@ impl RuntimeErpObserver { let before = population.observer.observed_key_count(); // Canonical fixed-width numeric identity; no raw sample or arbitrary label copy. let key = format!("{:016x}", if value == 0.0 { 0 } else { value.to_bits() }); - let result = population.observer.observe(&key, 0); + let range = population.coordinates.time_range; + let duration = range.end_ms.saturating_sub(range.start_ms).max(1); + let offset = timestamp_ms + .saturating_sub(range.start_ms) + .clamp(0, duration); + let interval = ((offset as u128 * 64) / duration as u128).min(63) as usize; + let result = population.observer.observe(&key, interval); let added = population .observer .observed_key_count() @@ -221,3 +227,84 @@ impl RuntimeErpObserver { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + fn fixture() -> (CatalogGeneration, asap_types::AggregationConfig) { + let config = asap_types::AggregationConfig::new( + asap_types::AggregationType::HLL, + String::new(), + Default::default(), + asap_types::KeyByLabelNames::new(vec![]), + asap_types::KeyByLabelNames::new(vec![]), + asap_types::KeyByLabelNames::new(vec![]), + String::new(), + 60, + 60, + planner_types::pre_asap::WindowKind::Tumbling, + String::new(), + "m".into(), + None, + None, + None, + ); + ( + CatalogGeneration { + schema_version: 1, + plan_id: 1, + plan_version: 1, + snapshot_sha256: "a".repeat(64), + }, + config, + ) + } + fn coordinate(group: usize) -> SummaryInstanceCoordinates { + SummaryInstanceCoordinates { + summary_definition_id: asap_types::PolicyFingerprint(1).into(), + time_range: HalfOpenTimeRange { + start_ms: 0, + end_ms: 60_000, + }, + group_values: BTreeMap::from([("job".into(), group.to_string())]), + } + } + #[test] + fn observations_keep_partitions_separate_and_invalidate_on_overflow() { + let observer = RuntimeErpObserver::new("http://127.0.0.1:1".into()); + let (generation, config) = fixture(); + observer.observe(&generation, coordinate(0), &config, 1, 1.0); + observer.observe(&generation, coordinate(0), &config, 2, 1.0); + observer.observe(&generation, coordinate(1), &config, 3, 2.0); + { + let state = observer.observations.lock().unwrap(); + assert_eq!(state.populations.len(), 2); + assert_eq!(state.total_keys, 2); + assert!(state + .populations + .values() + .all(|p| p.observer.observed_key_count() == 1)); + } + for group in 2..=MAX_POPULATIONS { + observer.observe(&generation, coordinate(group), &config, 4, 3.0); + } + let state = observer.observations.lock().unwrap(); + assert!(state.invalid.is_some()); + assert!(state + .populations + .values() + .all(|p| p.observer.snapshot().is_none())); + } + #[test] + fn changed_catalog_resets_invalid_observation_without_cross_generation_counts() { + let observer = RuntimeErpObserver::new("http://127.0.0.1:1".into()); + let (mut generation, config) = fixture(); + observer.observe(&generation, coordinate(0), &config, 1, f64::NAN); + generation.plan_version = 2; + observer.observe(&generation, coordinate(0), &config, 2, 9.0); + let state = observer.observations.lock().unwrap(); + assert!(state.invalid.is_none()); + assert_eq!(state.total_keys, 1); + assert_eq!(state.generation.as_ref(), Some(&generation)); + } +} From f033f5b24bccf810dee3dc72d4196df5d765a927 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:40:21 -0600 Subject: [PATCH 11/16] test(erp): exercise live worker observations through control-plane selection --- control_plane/src/physical/compiler.rs | 1 - .../src/precompute_engine/erp_observer.rs | 2 +- .../storage_engines/sketch_db/index/mod.rs | 4 + .../tests/support/univmon_erp_process.rs | 82 ++++++++++++++++++- 4 files changed, 86 insertions(+), 3 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 9aa011bdc..ab977d1ab 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -3159,7 +3159,6 @@ fn select_lifecycle( pub(crate) fn materialization_leaf_contract( node: &SummaryNode, ) -> Result<(String, Option, String), String> { - use planner_types::pre_asap::{CompareOpKind, QueryExpr, ScalarValue}; let SummaryExpr::SummaryAgg { child, .. } = &node.expr else { return Err("materialization requires a SummaryAgg leaf".into()); }; diff --git a/data_plane/src/precompute_engine/erp_observer.rs b/data_plane/src/precompute_engine/erp_observer.rs index e8bc37485..e31dca828 100644 --- a/data_plane/src/precompute_engine/erp_observer.rs +++ b/data_plane/src/precompute_engine/erp_observer.rs @@ -242,7 +242,7 @@ mod tests { String::new(), 60, 60, - planner_types::pre_asap::WindowKind::Tumbling, + asap_types::enums::WindowKind::Tumbling, String::new(), "m".into(), None, 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 60411e9bd..ee127988f 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -1062,6 +1062,10 @@ impl SketchStore { .map_err(|_| "summary instance start exceeds signed timestamp range")?; let end_ms = i64::try_from(window.1) .map_err(|_| "summary instance end exceeds signed timestamp range")?; + let group_fingerprint = xxhash_rust::xxh64::xxh64( + &serde_json::to_vec(&group_values).map_err(|error| error.to_string())?, + 0, + ); let instance_id = asap_types::sds::SummaryInstanceCoordinates { summary_definition_id, time_range: asap_types::sds::HalfOpenTimeRange { start_ms, end_ms }, diff --git a/data_plane/tests/support/univmon_erp_process.rs b/data_plane/tests/support/univmon_erp_process.rs index a37bc601e..4737db8f1 100644 --- a/data_plane/tests/support/univmon_erp_process.rs +++ b/data_plane/tests/support/univmon_erp_process.rs @@ -195,12 +195,26 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { .await .unwrap(); }); + let runtime_samples = control_plane::runtime_samples::RuntimeSamplesStore::new(8); + let runtime_port = unused_port(); + let runtime_endpoint = format!("http://127.0.0.1:{runtime_port}"); + let runtime_service = + control_plane::runtime_samples::RuntimeSamplesService::new(runtime_samples.clone()) + .into_server(); + let runtime_task = tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(runtime_service) + .serve(([127, 0, 0, 1], runtime_port).into()) + .await + .unwrap(); + }); let output = tempfile::tempdir().unwrap(); let path = output.path().join("planning.json"); std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); let port = unused_port(); let mut child = ChildGuard( Command::new(env!("CARGO_BIN_EXE_data_plane")) + .args(["--erp-runtime-samples-endpoint", &runtime_endpoint]) .args(["--profile", "asapquery", "--planning-snapshot"]) .arg(&path) .args([ @@ -231,11 +245,14 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { .unwrap() .as_millis() as i64; let base = now - now.rem_euclid(5000) - 20_000; - let samples: Vec<_> = raw + let mut samples: Vec<_> = raw .iter() .enumerate() .map(|(i, v)| (base + 1 + i as i64, *v)) .collect(); + // Declared finite source includes the preceding boundary; this sample is + // outside the query's left-open range and does not alter its truth. + samples.insert(0, (base, 0.0)); assert_eq!( remote_write( &client, @@ -259,6 +276,69 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { 204 ); drain_precompute(&client, &backend).await; + let keys = runtime_samples.keys(); + assert!( + !keys.is_empty(), + "real worker inputs must reach RuntimeSamples after finite drain" + ); + for key in keys { + let record = runtime_samples.latest(&key).unwrap(); + let observed: asap_types::erp_observation::ErpPopulationObservations< + control_plane::physical::erp::ErpObservedShape, + > = serde_json::from_value(record.payload["erp_population_observations"].clone()).unwrap(); + assert!(observed.invalid_reason.is_none(), "{observed:?}"); + assert!(!observed.populations.is_empty()); + assert_eq!(observed.window_end_ms - observed.window_start_ms, 5000); + assert!(plan + .summary_catalog + .materializations + .contains_key(&observed.summary_definition_id)); + assert_eq!( + observed.catalog_generation, + plan.summary_catalog.reference().unwrap() + ); + if key.sketch == "univmon" { + let mut live_snapshot: BackendLocalPlanningSnapshot = + serde_json::from_value(fixture.clone()).unwrap(); + let policy = live_snapshot.implementation.erp.as_mut().unwrap(); + policy.observed_shape_source = + Some(control_plane::physical::erp::ErpObservedShapeSource { + source: key.source.clone(), + sketch: key.sketch.clone(), + implementation: key.impl_name.clone(), + population_scope: Some( + control_plane::physical::erp::ErpPopulationObservationScope { + catalog_generation: observed.catalog_generation.clone(), + summary_definition_id: observed.summary_definition_id, + input_semantics: observed.input_semantics, + freshness: asap_types::erp_observation::ErpObservationFreshness { + max_age_ms: 60_000, + max_future_skew_ms: 1000, + }, + }, + ), + }); + policy.hydrate_observed_shape(&runtime_samples).unwrap(); + policy.resolve_population_data_descriptor(Some(&plan.summary_catalog)); + assert!(policy + .observed_populations + .as_ref() + .unwrap() + .invalid_reason + .is_none()); + let replanned = live_snapshot.compile().unwrap(); + assert!( + replanned + .precompute_plan + .materializations + .iter() + .any(|m| m.aggregation_type == asap_types::AggregationType::UnivMon), + "actual producer evidence should reach normal Planner selection" + ); + } + } + runtime_task.abort(); + for (i, query) in queries.iter().enumerate() { let result = wait_for_warm_instant( &client, From 190ee925181045b608827395ca3809a73e52d0df Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:42:04 -0600 Subject: [PATCH 12/16] fix(erp): bound retained population metadata bytes --- .../src/precompute_engine/erp_observer.rs | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/data_plane/src/precompute_engine/erp_observer.rs b/data_plane/src/precompute_engine/erp_observer.rs index e31dca828..766bf4103 100644 --- a/data_plane/src/precompute_engine/erp_observer.rs +++ b/data_plane/src/precompute_engine/erp_observer.rs @@ -8,6 +8,7 @@ use std::sync::{Arc, Mutex}; const MAX_POPULATIONS: usize = 128; const MAX_KEYS: usize = 65_536; +const MAX_POPULATION_METADATA_BYTES: usize = 1_048_576; struct Population { coordinates: SummaryInstanceCoordinates, @@ -23,6 +24,7 @@ struct Observations { populations: BTreeMap, extent: BTreeMap, total_keys: usize, + metadata_bytes: usize, invalid: Option, } @@ -97,6 +99,25 @@ impl RuntimeErpObserver { } return; } + if !state.populations.contains_key(&id) { + let bytes = coordinates + .group_values + .iter() + .try_fold(0usize, |total, (key, value)| { + total + .checked_add(key.len()) + .and_then(|n| n.checked_add(value.len())) + }); + let total = bytes.and_then(|bytes| state.metadata_bytes.checked_add(bytes)); + if total.is_none_or(|bytes| bytes > MAX_POPULATION_METADATA_BYTES) { + state.invalid = Some("ERP population metadata budget exceeded".into()); + for population in state.populations.values_mut() { + population.observer = ErpShapeObserver::new(1).unwrap(); + } + return; + } + state.metadata_bytes = total.unwrap(); + } let population = state.populations.entry(id).or_insert_with(|| Population { source: format!( "summary-definition:{}", From 559f6ca5a777216f42fe69f3099bac94f24ce03c Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:45:16 -0600 Subject: [PATCH 13/16] fix(erp): validate observation semantics against catalog operators --- control_plane/src/physical/compiler.rs | 1 + control_plane/src/physical/erp.rs | 24 ++++++++++++++++++++++++ crates/asap_types/src/sds.rs | 2 ++ crates/asap_types/src/summary_catalog.rs | 2 +- 4 files changed, 28 insertions(+), 1 deletion(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index ab977d1ab..b3482731d 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -2506,6 +2506,7 @@ fn observed_population_matches_root( && data.value_projection == ValueProjectionIdentity::SampleValue && data.partitioning == Some(PopulationPartitioning::PerEntity) && data.group_by_keys.is_empty() + && data.observation_semantics == asap_types::sds::TIMESTAMPED_OBSERVATION_SEMANTICS && observed.window_end_ms.checked_sub(observed.window_start_ms) == i64::try_from(window.saturating_mul(1000)).ok() } diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index ec4e035fb..90ad0d069 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -426,6 +426,30 @@ impl ErpPlanningInput { .materializations .get(&populations.summary_definition_id) .ok_or("ERP evidence summary is absent from the active catalog")?; + let summary = catalog + .summary_descriptors + .get(&materialization.summary_descriptor_id) + .ok_or("ERP summary descriptor is absent from the active catalog")?; + let actual_semantics = match &summary.operator { + asap_types::sds::SummaryOperator::Configured { + aggregation_type: asap_types::AggregationType::HLL, + .. + } => Some( + asap_types::erp_observation::ErpObservationInputSemantics::ScalarSampleValue, + ), + asap_types::sds::SummaryOperator::Configured { + aggregation_type: asap_types::AggregationType::UnivMon, + .. + } => Some( + asap_types::erp_observation::ErpObservationInputSemantics::UnitSampleFrequency, + ), + _ => None, + }; + if actual_semantics != Some(populations.input_semantics) { + return Err( + "ERP observation semantics differ from the installed summary operator".into(), + ); + } let data = catalog .data_descriptors .get(&materialization.data_descriptor_id) diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index 7720cf60a..94a7924fd 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -1,5 +1,7 @@ //! Shared SDS metadata contracts. Summary payload bytes remain storage-engine //! owned; catalogs and inventories contain identities and state references only. +pub const TIMESTAMPED_OBSERVATION_SEMANTICS: &str = "asap.timestamped-observations.v2"; + use crate::{AggregationType, PrecomputeMaterialization}; use planner_types::post_asap::{SketchAlgorithm, SketchParams, SummaryFamilyType}; use serde::{Deserialize, Serialize}; diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 4febef66b..c7bc75c1e 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -124,7 +124,7 @@ impl SummaryCatalog { .population_filter_canonical() .map_err(SummaryCatalogError::Descriptor)?, config.grouping_labels.labels.clone(), - "asap.timestamped-observations.v2", + crate::sds::TIMESTAMPED_OBSERVATION_SEMANTICS, ) .with_partitioning(config.partitioning) .with_timestamp_column(config.table_timestamp_column.clone()); From 6168be5f313f926e9e587aac036cb6c33720e5f2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:50:45 -0600 Subject: [PATCH 14/16] fix(erp): apportion update demand across observed populations --- control_plane/src/physical/erp.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index 90ad0d069..9ca65e0e2 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -676,11 +676,22 @@ impl ErpPlanningInput { let mut policy = self.clone(); policy.observed_populations = None; policy.mode = ErpAccuracyMode::Empirical; + let total_events: f64 = populations + .populations + .iter() + .map(|p| p.shape.observation.observed_events as f64) + .sum(); let mut errors: f64 = 0.0; let mut cost = 0.0; let mut evidence = Vec::new(); let valid = populations.populations.iter().all(|population| { policy.observed_shape = Some(population.shape.observation.clone()); + // expected_updates is the definition-wide demand; distribute + // it across partitions instead of charging it once per series. + policy.expected_updates = self.expected_updates + * population.shape.observation.observed_events as f64 + / total_events; + match policy.select_metric( algorithm.clone(), error_metric, From d87d30a2d8589035e5f15e342c231c72d753be84 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 21:03:12 -0600 Subject: [PATCH 15/16] fix(erp): bind observations to catalog generations and fit only in control plane --- control_plane/src/physical/erp.rs | 140 ++++++++-------- crates/asap_types/src/erp_observation.rs | 150 ++++++++++++++++++ data_plane/src/main.rs | 9 +- .../src/precompute_engine/erp_observer.rs | 64 +++++--- .../src/precompute_engine/series_router.rs | 10 +- .../tests/support/univmon_erp_process.rs | 2 +- 6 files changed, 271 insertions(+), 104 deletions(-) diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index 9ca65e0e2..a679a607a 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -7,7 +7,7 @@ use asap_aware_mapping::erp::{ use planner_types::post_asap::{SketchAlgorithm, SketchParams}; use serde::{Deserialize, Serialize}; use serde_json::Value; -use std::collections::{BTreeMap, HashMap}; +use std::collections::BTreeMap; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] @@ -22,74 +22,40 @@ pub struct ErpObservedShape { /// instead of silently under-reporting cardinality. #[derive(Debug)] pub struct ErpShapeObserver { - frequencies: HashMap, - interval_updates: HashMap, - max_observed_keys: usize, - max_observed_intervals: usize, - updates: u64, - invalid: bool, + inner: asap_types::erp_observation::BoundedFrequencyObserver, } - impl ErpShapeObserver { pub fn observed_key_count(&self) -> usize { - self.frequencies.len() + self.inner.observed_key_count() } - - pub fn new(max_observed_keys: usize) -> Result { - Self::with_limits(max_observed_keys, 256) + pub fn new(max_keys: usize) -> Result { + Self::with_limits(max_keys, 256) } - - pub fn with_limits( - max_observed_keys: usize, - max_observed_intervals: usize, - ) -> Result { - if max_observed_keys == 0 || max_observed_intervals == 0 { - return Err("observation limits must be positive"); - } + pub fn with_limits(max_keys: usize, max_intervals: usize) -> Result { Ok(Self { - frequencies: HashMap::new(), - interval_updates: HashMap::new(), - max_observed_keys, - max_observed_intervals, - updates: 0, - invalid: false, + inner: asap_types::erp_observation::BoundedFrequencyObserver::new( + max_keys, + max_intervals, + )?, }) } - pub fn observe(&mut self, key: &str, interval: usize) -> Result<(), &'static str> { - if self.invalid { - return Err("ERP observation was invalidated; start a fresh observation window"); - } if key.len() > 4096 { - self.invalid = true; + self.inner.invalidate(); return Err("ERP shape observer key size exceeded"); } - if !self.frequencies.contains_key(key) && self.frequencies.len() == self.max_observed_keys { - self.invalid = true; - return Err("ERP shape observer cardinality cap exceeded"); - } - if !self.interval_updates.contains_key(&interval) - && self.interval_updates.len() == self.max_observed_intervals - { - self.invalid = true; - return Err("ERP shape observer interval cap exceeded"); - } - let Some(updates) = self.updates.checked_add(1) else { - self.invalid = true; - return Err("ERP shape observer count overflow"); - }; - *self.frequencies.entry(key.to_owned()).or_default() += 1; - *self.interval_updates.entry(interval).or_default() += 1; - self.updates = updates; - Ok(()) + self.inner.observe(key.to_owned(), interval) } - pub fn snapshot(&self) -> Option { - if self.frequencies.is_empty() || self.invalid { - return None; - } - let mut counts: Vec<_> = self.frequencies.values().copied().collect(); - counts.sort_unstable_by(|left, right| right.cmp(left)); + ErpObservedShape::from_empirical(&self.inner.snapshot()?) + } +} +impl ErpObservedShape { + pub fn from_empirical( + empirical: &asap_types::erp_observation::EmpiricalFrequencyObservation, + ) -> Option { + let events = empirical.event_count()?; + let counts = &empirical.sorted_counts; let exponent = fit_zipf_exponent(&counts); let fits = [("uniform", 0.0), ("zipf", exponent)] .into_iter() @@ -104,7 +70,7 @@ impl ErpShapeObserver { .iter() .zip(expected) .map(|(count, expected)| { - (*count as f64 / self.updates as f64 - expected / total).abs() + (*count as f64 / events as f64 - expected / total).abs() }) .sum::() / 2.0; @@ -117,13 +83,13 @@ impl ErpShapeObserver { }, goodness_of_fit: distance, // A fit-quality score, not an estimator tail-probability claim. - confidence: (1.0 - distance) * (1.0 - 1.0 / (self.updates as f64).sqrt()), + confidence: (1.0 - distance) * (1.0 - 1.0 / (events as f64).sqrt()), } }) .collect(); - let mut nonzero: Vec<_> = self - .interval_updates - .values() + let mut nonzero: Vec<_> = empirical + .interval_counts + .iter() .copied() .filter(|count| *count > 0) .collect(); @@ -132,8 +98,8 @@ impl ErpShapeObserver { let peak = nonzero.last().copied().unwrap_or(median); Some(ErpObservedShape { observation: ErpShapeObservation { - cardinality: self.frequencies.len() as u64, - observed_events: self.updates, + cardinality: counts.len() as u64, + observed_events: events, fits, empirical_fingerprint: None, }, @@ -506,7 +472,7 @@ impl ErpPlanningInput { .get("erp_population_observations") .ok_or("latest runtime sample has no erp_population_observations")?; let observed: asap_types::erp_observation::ErpPopulationObservations< - ErpObservedShape, + asap_types::erp_observation::EmpiricalFrequencyObservation, > = serde_json::from_value(value.clone()) .map_err(|error| format!("invalid ERP population observations: {error}"))?; observed.validate_identity_and_freshness( @@ -518,7 +484,9 @@ impl ErpPlanningInput { if observed.input_semantics != scope.input_semantics { return Err("ERP observation input semantics do not match the summary".into()); } - Ok(observed) + observed + .try_map_shapes(|empirical| ErpObservedShape::from_empirical(&empirical)) + .ok_or_else(|| "invalid bounded empirical frequency observations".to_owned()) })(); // An explicit invalid envelope prevents selection from borrowing an // older fit or falling back to the artifact's distribution identity. @@ -1415,8 +1383,21 @@ mod tests { fn publish_population_fixture( samples: &crate::runtime_samples::RuntimeSamplesStore, - payload: serde_json::Value, + mut payload: serde_json::Value, ) { + if let Some(populations) = payload + .pointer_mut("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/erp_population_observations/populations") + .and_then(Value::as_array_mut) + { + for population in populations { + if let Some(cardinality) = population + .pointer("/shape/observation/cardinality") + .and_then(Value::as_u64) + { + population["shape"] = serde_json::json!({"sorted_counts":vec![2;cardinality as usize],"interval_counts":[cardinality*2]}); + } + } + } samples.append_for_test(crate::runtime_samples::RuntimeRecord { source: "backend-a".into(), sketch: "univmon".into(), @@ -1468,16 +1449,23 @@ mod tests { /// Only the activated catalog may supply a candidate's data contract. #[test] fn online_population_descriptor_requires_authoritative_catalog() { + let mut fixture: Value = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let mut query = fixture["query_workload"]["repeating_queries"][3].clone(); + query["query"] = "distinct_over_time(asap_demo_latency_ms[5s])".into(); + query["requirements"]["accuracy"] = serde_json::json!({"explicit":{"Epsilon":0.05}}); + fixture["query_workload"]["repeating_queries"] = serde_json::json!([query]); let snapshot: crate::physical::compiler::BackendLocalPlanningSnapshot = - serde_json::from_str(include_str!( - "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" - )) - .unwrap(); + serde_json::from_value(fixture).unwrap(); let plan = snapshot.compile().unwrap(); let (mut policy, mut observed) = online_population_fixture(); observed.catalog_generation = plan.summary_catalog.reference().unwrap(); observed.summary_definition_id = *plan.summary_catalog.materializations.keys().next().unwrap(); + observed.input_semantics = + asap_types::erp_observation::ErpObservationInputSemantics::ScalarSampleValue; policy.observed_populations = Some(observed.clone()); policy.resolve_population_data_descriptor(Some(&plan.summary_catalog)); let expected = &plan.summary_catalog.materializations[&observed.summary_definition_id] @@ -1563,14 +1551,16 @@ mod tests { let mut observer = ErpShapeObserver::with_limits(2, 2).unwrap(); observer.observe("a", usize::MAX).unwrap(); observer.observe("a", 0).unwrap(); - assert_eq!(observer.interval_updates.len(), 2); + assert_eq!(observer.inner.snapshot().unwrap().interval_counts.len(), 2); assert!(observer.observe("a", 1).is_err()); assert!(observer.snapshot().is_none()); - let mut observer = ErpShapeObserver::new(2).unwrap(); - observer.observe("a", 0).unwrap(); - observer.updates = u64::MAX; - assert!(observer.observe("a", 0).is_err()); - assert!(observer.snapshot().is_none()); + assert!(ErpObservedShape::from_empirical( + &asap_types::erp_observation::EmpiricalFrequencyObservation { + sorted_counts: vec![u64::MAX, 1], + interval_counts: vec![u64::MAX, 1], + } + ) + .is_none()); } #[test] diff --git a/crates/asap_types/src/erp_observation.rs b/crates/asap_types/src/erp_observation.rs index efb693c05..7693a3c72 100644 --- a/crates/asap_types/src/erp_observation.rs +++ b/crates/asap_types/src/erp_observation.rs @@ -77,3 +77,153 @@ impl ErpPopulationObservations { Ok(()) } } + +/// Empirical sufficient statistics for frequency-shape fitting. No raw keys, +/// raw samples, distribution-family assertion, or planner configuration crosses the wire. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct EmpiricalFrequencyObservation { + pub sorted_counts: Vec, + pub interval_counts: Vec, +} +impl EmpiricalFrequencyObservation { + pub fn event_count(&self) -> Option { + if self.sorted_counts.len() > 65_536 + || self.interval_counts.len() > 256 + || self.sorted_counts.is_empty() + || self.sorted_counts.iter().any(|n| *n == 0) + || self.sorted_counts.windows(2).any(|w| w[0] < w[1]) + { + return None; + } + let total = self + .sorted_counts + .iter() + .try_fold(0u64, |sum, n| sum.checked_add(*n))?; + let intervals = self + .interval_counts + .iter() + .try_fold(0u64, |sum, n| sum.checked_add(*n))?; + (total == intervals).then_some(total) + } +} + +/// Bounded exact frequency counting. Fitting and ERP selection remain control-plane work. +#[derive(Debug)] +pub struct BoundedFrequencyObserver { + frequencies: std::collections::HashMap, + intervals: std::collections::HashMap, + max_keys: usize, + max_intervals: usize, + events: u64, + invalid: bool, +} +impl BoundedFrequencyObserver { + pub fn new(max_keys: usize, max_intervals: usize) -> Result { + if max_keys == 0 || max_intervals == 0 { + return Err("observation limits must be positive"); + } + Ok(Self { + frequencies: Default::default(), + intervals: Default::default(), + max_keys, + max_intervals, + events: 0, + invalid: false, + }) + } + pub fn observed_key_count(&self) -> usize { + self.frequencies.len() + } + pub fn invalidate(&mut self) { + self.invalid = true; + } + pub fn observe(&mut self, key: K, interval: usize) -> Result<(), &'static str> { + if self.invalid { + return Err("ERP observation was invalidated"); + } + if (!self.frequencies.contains_key(&key) && self.frequencies.len() >= self.max_keys) + || (!self.intervals.contains_key(&interval) + && self.intervals.len() >= self.max_intervals) + { + self.invalid = true; + return Err("ERP observation key or interval budget exceeded"); + } + let Some(events) = self.events.checked_add(1) else { + self.invalid = true; + return Err("ERP observation count overflow"); + }; + *self.frequencies.entry(key).or_default() += 1; + *self.intervals.entry(interval).or_default() += 1; + self.events = events; + Ok(()) + } + pub fn snapshot(&self) -> Option { + if self.invalid || self.frequencies.is_empty() { + return None; + } + let mut sorted_counts: Vec<_> = self.frequencies.values().copied().collect(); + sorted_counts.sort_unstable_by(|a, b| b.cmp(a)); + Some(EmpiricalFrequencyObservation { + sorted_counts, + interval_counts: self.intervals.values().copied().collect(), + }) + } +} + +impl ErpPopulationObservations { + pub fn try_map_shapes( + self, + mut convert: impl FnMut(S) -> Option, + ) -> Option> { + let populations = self + .populations + .into_iter() + .map(|population| { + Some(ErpPopulationObservation { + population_id: population.population_id, + shape: convert(population.shape)?, + }) + }) + .collect::>>()?; + Some(ErpPopulationObservations { + schema_version: self.schema_version, + catalog_generation: self.catalog_generation, + summary_definition_id: self.summary_definition_id, + observed_at_unix_ms: self.observed_at_unix_ms, + window_start_ms: self.window_start_ms, + window_end_ms: self.window_end_ms, + input_semantics: self.input_semantics, + invalid_reason: self.invalid_reason, + populations, + }) + } +} + +#[cfg(test)] +mod empirical_tests { + use super::*; + #[test] + fn count_overflow_invalidates_instead_of_publishing_prefix() { + let mut observer = BoundedFrequencyObserver::new(2, 2).unwrap(); + observer.observe(1u64, 0).unwrap(); + observer.events = u64::MAX; + assert!(observer.observe(1, 0).is_err()); + assert!(observer.snapshot().is_none()); + } + #[test] + fn malformed_frequency_counts_fail_closed() { + assert!(EmpiricalFrequencyObservation { + sorted_counts: vec![2, 1], + interval_counts: vec![2] + } + .event_count() + .is_none()); + assert!(EmpiricalFrequencyObservation { + sorted_counts: vec![1, 2], + interval_counts: vec![3] + } + .event_count() + .is_none()); + } +} diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 208b7db50..4c8204104 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -878,10 +878,17 @@ async fn main() -> Result<()> { sketch_index.clone(), ); if let Some(endpoint) = args.erp_runtime_samples_endpoint.clone() { + let generation = engine + .ingest_state() + .physical_plan_snapshot() + .and_then(|plan| plan.precompute_plan.summary_catalog.clone()) + .ok_or_else(|| { + std::io::Error::other("ERP observation requires an installed catalog") + })?; engine .ingest_state() .router - .enable_erp_observation(endpoint) + .enable_erp_observation(endpoint, generation) .map_err(std::io::Error::other)?; } let worker_diagnostics = engine.diagnostics(); diff --git a/data_plane/src/precompute_engine/erp_observer.rs b/data_plane/src/precompute_engine/erp_observer.rs index 766bf4103..25fb6ffcf 100644 --- a/data_plane/src/precompute_engine/erp_observer.rs +++ b/data_plane/src/precompute_engine/erp_observer.rs @@ -2,7 +2,6 @@ //! an explicit finite-source completion barrier. No worker timestamp is a seal. use asap_types::erp_observation::*; use asap_types::sds::*; -use control_plane::physical::erp::{ErpObservedShape, ErpShapeObserver}; use std::collections::BTreeMap; use std::sync::{Arc, Mutex}; @@ -16,7 +15,7 @@ struct Population { source: String, sketch: String, implementation: String, - observer: ErpShapeObserver, + observer: BoundedFrequencyObserver, } #[derive(Default)] struct Observations { @@ -33,10 +32,13 @@ pub struct RuntimeErpObserver { observations: Mutex, } impl RuntimeErpObserver { - pub fn new(endpoint: String) -> Arc { + pub fn new(endpoint: String, generation: CatalogGeneration) -> Arc { Arc::new(Self { endpoint, - observations: Mutex::new(Observations::default()), + observations: Mutex::new(Observations { + generation: Some(generation), + ..Default::default() + }), }) } pub fn observe( @@ -62,11 +64,10 @@ impl RuntimeErpObserver { _ => return, }; let mut state = self.observations.lock().unwrap(); + // Catalog installation establishes identity. Delayed old worker input + // cannot reset the current generation or create a partial replacement fit. if state.generation.as_ref() != Some(generation) { - *state = Observations { - generation: Some(generation.clone()), - ..Default::default() - }; + return; } if state.invalid.is_some() { return; @@ -79,7 +80,7 @@ impl RuntimeErpObserver { { state.invalid = Some("unsupported non-finite or transformed summary input".into()); for population in state.populations.values_mut() { - population.observer = ErpShapeObserver::new(1).unwrap(); + population.observer = BoundedFrequencyObserver::new(1, 64).unwrap(); } return; } @@ -95,7 +96,7 @@ impl RuntimeErpObserver { if !state.populations.contains_key(&id) && state.populations.len() >= MAX_POPULATIONS { state.invalid = Some("ERP population observation budget exceeded".into()); for population in state.populations.values_mut() { - population.observer = ErpShapeObserver::new(1).unwrap(); + population.observer = BoundedFrequencyObserver::new(1, 64).unwrap(); } return; } @@ -112,7 +113,7 @@ impl RuntimeErpObserver { if total.is_none_or(|bytes| bytes > MAX_POPULATION_METADATA_BYTES) { state.invalid = Some("ERP population metadata budget exceeded".into()); for population in state.populations.values_mut() { - population.observer = ErpShapeObserver::new(1).unwrap(); + population.observer = BoundedFrequencyObserver::new(1, 64).unwrap(); } return; } @@ -127,18 +128,18 @@ impl RuntimeErpObserver { semantics, sketch: sketch.into(), implementation: implementation.into(), - observer: ErpShapeObserver::new(MAX_KEYS).unwrap(), + observer: BoundedFrequencyObserver::new(MAX_KEYS, 64).unwrap(), }); let before = population.observer.observed_key_count(); // Canonical fixed-width numeric identity; no raw sample or arbitrary label copy. - let key = format!("{:016x}", if value == 0.0 { 0 } else { value.to_bits() }); + let key = if value == 0.0 { 0 } else { value.to_bits() }; let range = population.coordinates.time_range; let duration = range.end_ms.saturating_sub(range.start_ms).max(1); let offset = timestamp_ms .saturating_sub(range.start_ms) .clamp(0, duration); let interval = ((offset as u128 * 64) / duration as u128).min(63) as usize; - let result = population.observer.observe(&key, interval); + let result = population.observer.observe(key, interval); let added = population .observer .observed_key_count() @@ -147,7 +148,7 @@ impl RuntimeErpObserver { if result.is_err() || state.total_keys > MAX_KEYS { state.invalid = Some("ERP key observation budget exceeded".into()); for population in state.populations.values_mut() { - population.observer = ErpShapeObserver::new(1).unwrap(); + population.observer = BoundedFrequencyObserver::new(1, 64).unwrap(); } } } @@ -171,7 +172,7 @@ impl RuntimeErpObserver { String, String, String, - ErpPopulationObservations, + ErpPopulationObservations, ), > = BTreeMap::new(); for (id, population) in &state.populations { @@ -292,8 +293,8 @@ mod tests { } #[test] fn observations_keep_partitions_separate_and_invalidate_on_overflow() { - let observer = RuntimeErpObserver::new("http://127.0.0.1:1".into()); let (generation, config) = fixture(); + let observer = RuntimeErpObserver::new("http://127.0.0.1:1".into(), generation.clone()); observer.observe(&generation, coordinate(0), &config, 1, 1.0); observer.observe(&generation, coordinate(0), &config, 2, 1.0); observer.observe(&generation, coordinate(1), &config, 3, 2.0); @@ -317,15 +318,28 @@ mod tests { .all(|p| p.observer.snapshot().is_none())); } #[test] - fn changed_catalog_resets_invalid_observation_without_cross_generation_counts() { - let observer = RuntimeErpObserver::new("http://127.0.0.1:1".into()); - let (mut generation, config) = fixture(); - observer.observe(&generation, coordinate(0), &config, 1, f64::NAN); - generation.plan_version = 2; - observer.observe(&generation, coordinate(0), &config, 2, 9.0); + fn delayed_generation_cannot_reset_current_population_counts() { + let (generation, config) = fixture(); + let observer = RuntimeErpObserver::new("http://127.0.0.1:1".into(), generation.clone()); + observer.observe(&generation, coordinate(0), &config, 1, 1.0); + let mut stale = generation.clone(); + stale.plan_version = 0; + observer.observe(&stale, coordinate(0), &config, 2, 99.0); + observer.observe(&generation, coordinate(0), &config, 3, 2.0); let state = observer.observations.lock().unwrap(); - assert!(state.invalid.is_none()); - assert_eq!(state.total_keys, 1); + assert_eq!(state.total_keys, 2); assert_eq!(state.generation.as_ref(), Some(&generation)); + assert_eq!( + state + .populations + .values() + .next() + .unwrap() + .observer + .snapshot() + .unwrap() + .sorted_counts, + vec![1, 1] + ); } } diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs index 3a90705d8..b709d9e6f 100644 --- a/data_plane/src/precompute_engine/series_router.rs +++ b/data_plane/src/precompute_engine/series_router.rs @@ -163,9 +163,15 @@ impl SeriesRouter { } } - pub fn enable_erp_observation(&self, endpoint: String) -> Result<(), String> { + pub fn enable_erp_observation( + &self, + endpoint: String, + generation: asap_types::sds::CatalogGeneration, + ) -> Result<(), String> { self.erp_observer - .set(super::erp_observer::RuntimeErpObserver::new(endpoint)) + .set(super::erp_observer::RuntimeErpObserver::new( + endpoint, generation, + )) .map_err(|_| "ERP observer already configured".into()) } pub fn erp_observer(&self) -> Option> { diff --git a/data_plane/tests/support/univmon_erp_process.rs b/data_plane/tests/support/univmon_erp_process.rs index 4737db8f1..9c933b8cb 100644 --- a/data_plane/tests/support/univmon_erp_process.rs +++ b/data_plane/tests/support/univmon_erp_process.rs @@ -284,7 +284,7 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { for key in keys { let record = runtime_samples.latest(&key).unwrap(); let observed: asap_types::erp_observation::ErpPopulationObservations< - control_plane::physical::erp::ErpObservedShape, + asap_types::erp_observation::EmpiricalFrequencyObservation, > = serde_json::from_value(record.payload["erp_population_observations"].clone()).unwrap(); assert!(observed.invalid_reason.is_none(), "{observed:?}"); assert!(!observed.populations.is_empty()); From 9d98460347226ea1187dc8a45adb8cea4530c017 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 21:10:51 -0600 Subject: [PATCH 16/16] fix(erp): reset observation epoch only on catalog activation --- control_plane/src/physical/erp.rs | 2 +- .../drivers/ingest/prometheus_remote_write.rs | 9 +++++++++ data_plane/src/drivers/query/servers/http.rs | 15 ++++++++++++--- data_plane/src/precompute_engine/erp_observer.rs | 16 ++++++++++++++++ .../univmon-erp-process-validation.md | 7 +++++++ 5 files changed, 45 insertions(+), 4 deletions(-) diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index a679a607a..c41fa2fe2 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -1540,7 +1540,7 @@ mod tests { observer.observe("a", 0).unwrap(); assert_eq!( observer.observe("b", 0), - Err("ERP shape observer cardinality cap exceeded") + Err("ERP observation key or interval budget exceeded") ); assert!(observer.snapshot().is_none()); assert!(observer.observe("a", 0).is_err()); diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 7deb61e22..bb8181d29 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -177,6 +177,15 @@ impl PrometheusRemoteWriteReceiver { self.inner.stats.clone() } + pub(crate) fn install_erp_observation_generation( + &self, + generation: asap_types::sds::CatalogGeneration, + ) { + if let Some(observer) = self.inner.ingest.router.erp_observer() { + observer.install_generation(generation); + } + } + /// Permanently seal this finite source before queuing worker barriers. pub async fn drain(&self) -> Result<(), String> { { diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 087b3d586..ff66bc12f 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -6263,14 +6263,23 @@ async fn handle_activate_physical_plan( }; let _guard = state.physical_plan_lock.lock().await; let store = Arc::clone(&state.sketch_index); + let remote_write = state.remote_write.clone(); let old = match lifecycle.activate_with_prepare( request.plan_id, request.plan_version, unix_time_ms(), move |plan| match plan.summary_catalog.as_ref() { - Some(catalog) => store - .install_summary_catalog(Arc::clone(catalog)) - .map_err(|error| format!("SummaryCatalog install error: {error}")), + Some(catalog) => { + store + .install_summary_catalog(Arc::clone(catalog)) + .map_err(|error| format!("SummaryCatalog install error: {error}"))?; + if let Some(receiver) = &remote_write { + receiver.install_erp_observation_generation( + catalog.reference().map_err(|error| error.to_string())?, + ); + } + Ok(()) + } None => Err("authoritative SummaryCatalog is unavailable".to_string()), }, ) { diff --git a/data_plane/src/precompute_engine/erp_observer.rs b/data_plane/src/precompute_engine/erp_observer.rs index 25fb6ffcf..47593276b 100644 --- a/data_plane/src/precompute_engine/erp_observer.rs +++ b/data_plane/src/precompute_engine/erp_observer.rs @@ -41,6 +41,17 @@ impl RuntimeErpObserver { }), }) } + /// Only the accepted catalog activation path may reset an observation epoch. + pub(crate) fn install_generation(&self, generation: CatalogGeneration) { + let mut state = self.observations.lock().unwrap(); + if state.generation.as_ref() != Some(&generation) { + *state = Observations { + generation: Some(generation), + ..Default::default() + }; + } + } + pub fn observe( &self, generation: &CatalogGeneration, @@ -321,6 +332,11 @@ mod tests { fn delayed_generation_cannot_reset_current_population_counts() { let (generation, config) = fixture(); let observer = RuntimeErpObserver::new("http://127.0.0.1:1".into(), generation.clone()); + let mut previous = generation.clone(); + previous.plan_version = 0; + observer.install_generation(previous.clone()); + observer.observe(&previous, coordinate(0), &config, 0, 77.0); + observer.install_generation(generation.clone()); observer.observe(&generation, coordinate(0), &config, 1, 1.0); let mut stale = generation.clone(); stale.plan_version = 0; diff --git a/docs/developer_docs/univmon-erp-process-validation.md b/docs/developer_docs/univmon-erp-process-validation.md index b3e5849ef..51e65f914 100644 --- a/docs/developer_docs/univmon-erp-process-validation.md +++ b/docs/developer_docs/univmon-erp-process-validation.md @@ -69,3 +69,10 @@ The three queries need different physical configurations in this fixture; they do not demonstrate a single shared state. Planner may share compatible equal- parameter states. The finite replay barrier also does not establish continuous multi-worker population completeness, which is a separate publication contract. + +Finite runtime observations carry bounded empirical frequency counts and temporal +interval counts through the shared metadata contract. The data plane does not +classify distributions or depend on Planner fitting types; the control plane +fits these counts before profile matching. Raw keys are absent from feedback. +An accepted catalog activation establishes the observer generation. Delayed +samples from another generation are ignored and cannot reset current counts.