From 02555b3e26258fb3320b8ee12500ace5b2ca06ef Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 2 Sep 2026 21:48:40 -0600 Subject: [PATCH 1/3] feat(precompute): install runtime contracts from physical plan --- control_plane/src/emit/backend_push.rs | 61 ++-- control_plane/src/emit/stage_config.rs | 33 +- control_plane/src/opamp/mod.rs | 1 + control_plane/src/physical/compiler.rs | 298 +++++++++++++++++- data_plane/src/drivers/query/servers/http.rs | 41 ++- data_plane/src/main.rs | 10 + .../types/hot_reload_config.rs | 12 + .../control-plane/physical-compiler.md | 17 +- .../control-plane/plan-publication.md | 40 +-- 9 files changed, 432 insertions(+), 81 deletions(-) diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index 7167cdf9b..e3f9985ce 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -442,41 +442,52 @@ async fn push_cumulative_entries( // succeeds only when all three are accepted. let plan_id = PLAN_ID_COUNTER.fetch_add(1, Ordering::Relaxed); let generated_at_unix_ms = now_unix_ms(); - let plan_bytes = match crate::backend_plan::from_stage_config( + let backend_plan = match crate::backend_plan::from_stage_config( &cumulative_be, monitors, plan_id, generated_at_unix_ms, ) { - Ok(plan) => plan.encode_to_vec(), + Ok(plan) => plan, Err(e) => { warn!(error = %e, "backend_plan::from_stage_config failed; refusing partial publication"); return PushOutcome::EmitFailed; } }; - let precompute_plan = PrecomputePlan { - envelope: PlanEnvelope { - plan_id, - plan_version: 1, - generated_at_unix_ms, - activation_unix_ms: generated_at_unix_ms, - expiry_unix_ms: None, - backend_compat: "asap-query-backend.v1".into(), - planner_revision: crate::physical::compiler::PLANNER_REVISION.into(), - capability_snapshot_id: "replanner".into(), - }, - materializations: match cumulative_be - .aggregations - .iter() - .map(crate::backend_plan::aggregation_config_for_materialization) - .collect::>>() - { - Ok(materializations) => materializations, - Err(error) => { - warn!(%error, "failed to build typed PrecomputePlan"); - return PushOutcome::EmitFailed; - } - }, + let plan_bytes = backend_plan.encode_to_vec(); + let precompute_envelope = PlanEnvelope { + plan_id, + plan_version: 1, + generated_at_unix_ms, + activation_unix_ms: generated_at_unix_ms, + expiry_unix_ms: None, + backend_compat: "asap-query-backend.v1".into(), + planner_revision: crate::physical::compiler::PLANNER_REVISION.into(), + capability_snapshot_id: "replanner".into(), + }; + let materializations = match cumulative_be + .aggregations + .iter() + .map(crate::backend_plan::aggregation_config_for_materialization) + .collect::>>() + { + Ok(materializations) => materializations, + Err(error) => { + warn!(%error, "failed to build typed PrecomputePlan"); + return PushOutcome::EmitFailed; + } + }; + let precompute_plan = match PrecomputePlan::build( + precompute_envelope, + materializations, + &backend_plan, + &["legacy-replanner".into()], + ) { + Ok(plan) => plan, + Err(error) => { + warn!(%error, "failed to validate typed PrecomputePlan"); + return PushOutcome::EmitFailed; + } }; // Storage-routing: the routing classifier (`build_routing_entry` in diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index b9ff01811..8155a20eb 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -3142,24 +3142,33 @@ fn sketch_params_to_json(p: &SketchParams) -> JsonValue { SketchParams::Kll { k } => json!({ "k": k }), SketchParams::DDSketch { alpha } => json!({ "alpha": alpha }), SketchParams::Hll { precision } => json!({ "precision": precision }), - // Cms/CmsWithHeap: matches the pre-split shape exactly — the old - // `SketchParams::Cms` arm never emitted `with_heap` in JSON even - // though `CmsParams.with_heap` existed as a field; heap-bearing - // and bare CMS produced identical wire JSON. `heap_size` is a - // new field with no wire representation here (nothing on the - // real backend wire path reads it — see `bind_cms_topk.rs`). - SketchParams::Cms { width, depth } | SketchParams::CmsWithHeap { width, depth, .. } => { - json!({ "w": width, "d": depth }) - } + SketchParams::Cms { width, depth } => json!({ "w": width, "d": depth }), + SketchParams::CmsWithHeap { + width, + depth, + heap_size, + } => json!({ + "w": width, + "d": depth, + "with_heap": true, + "heap_size": heap_size, + }), // CountSketch/CountSketchWithHeap: the old arm always emitted // `with_heap` (from `CountSketchParams.with_heap: bool`); // that boolean is now the kind identity itself. SketchParams::CountSketch { width, depth } => { json!({ "w": width, "d": depth, "with_heap": false }) } - SketchParams::CountSketchWithHeap { width, depth, .. } => { - json!({ "w": width, "d": depth, "with_heap": true }) - } + SketchParams::CountSketchWithHeap { + width, + depth, + heap_size, + } => json!({ + "w": width, + "d": depth, + "with_heap": true, + "heap_size": heap_size, + }), // Exact accumulators never reach here -- see // `sketch_kind_to_backend_type`'s doc. SketchParams::Kmv { .. } | SketchParams::Theta { .. } => unreachable!( diff --git a/control_plane/src/opamp/mod.rs b/control_plane/src/opamp/mod.rs index 4c08647f6..1c8a41329 100644 --- a/control_plane/src/opamp/mod.rs +++ b/control_plane/src/opamp/mod.rs @@ -1016,6 +1016,7 @@ mod tests { }, materializations: vec![crate::physical::compiler::CollectorMaterialization { query_id: "q".into(), + materialization: asap_types::PolicyFingerprint(1), metric: "requests".into(), algorithm: "hll".into(), parameters: serde_json::json!({"precision": 14}), diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index ff5217bb6..21ae7cdab 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -159,6 +159,7 @@ pub struct PlanEnvelope { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct CollectorMaterialization { pub query_id: String, + pub materialization: asap_types::PolicyFingerprint, pub metric: String, pub algorithm: String, pub parameters: Value, @@ -196,9 +197,223 @@ pub struct CollectorPlan { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PrecomputePlan { pub envelope: PlanEnvelope, + pub ingest: IngestContract, + pub schemas: Vec, + pub producers: Vec, pub materializations: Vec, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum IngestProtocol { + ModifiedOtlpMetricsV1, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TimestampUnit { + UnixNanoseconds, +} + +/// Backend ingress semantics installed with the precompute projection. This +/// replaces implicit knowledge formerly hidden in the streaming-config path. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct IngestContract { + pub protocol: IngestProtocol, + pub endpoint_path: String, + pub timestamp_unit: TimestampUnit, + pub require_plan_identity: bool, + pub require_materialization_identity: bool, + pub require_registered_producer: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum StateEncoding { + SketchlibProtobufV1, + SketchCoreMsgpackV1, + ExactAccumulatorV1, +} + +/// Decoder/schema contract for one content-addressed materialization. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct StateSchemaContract { + pub schema_id: String, + pub schema_version: u32, + pub materialization: asap_types::PolicyFingerprint, + pub family: SummaryFamilyType, + pub encodings: Vec, +} + +/// A collector authorized to produce state for one materialization. Runtime +/// producer epochs and frame sequences belong to TransmissionPlan. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(deny_unknown_fields)] +pub struct ProducerContract { + pub producer_id: String, + pub collector_id: String, + pub materialization: asap_types::PolicyFingerprint, + pub schema_id: String, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum PrecomputePlanError { + #[error("PrecomputePlan envelope does not match BackendPlan identity/lifecycle")] + PlanIdentityMismatch, + #[error("precompute ingest endpoint must be /v1/metrics")] + UnsupportedIngestEndpoint, + #[error("duplicate materialization {0}")] + DuplicateMaterialization(u64), + #[error("schema set does not exactly match the materialization set")] + SchemaSetMismatch, + #[error("schema {schema_id} has invalid version or no encoding")] + InvalidSchema { schema_id: String }, + #[error("producer {producer_id} references an unknown materialization or schema")] + InvalidProducer { producer_id: String }, + #[error("materialization {0} has no registered producer")] + MissingProducer(u64), + #[error("duplicate producer binding {0}")] + DuplicateProducer(String), +} + +impl PrecomputePlan { + pub fn build( + envelope: PlanEnvelope, + materializations: Vec, + backend_plan: &BackendPlan, + producer_ids: &[String], + ) -> Result { + if envelope.plan_id != backend_plan.plan_id + || envelope.plan_version != backend_plan.plan_version + || envelope.generated_at_unix_ms != backend_plan.generated_at_unix_ms + || envelope.activation_unix_ms != backend_plan.activation_unix_ms + || envelope.expiry_unix_ms != backend_plan.expiry_unix_ms + || envelope.backend_compat != backend_plan.backend_compat + { + return Err(PrecomputePlanError::PlanIdentityMismatch); + } + let schemas = backend_plan + .materializations + .iter() + .map(|(fingerprint, materialization)| StateSchemaContract { + schema_id: state_schema_id(*fingerprint), + schema_version: 1, + materialization: *fingerprint, + family: materialization.family.clone(), + encodings: state_encodings(&materialization.family), + }) + .collect::>(); + let producers = producer_ids + .iter() + .flat_map(|producer_id| { + schemas.iter().map(move |schema| ProducerContract { + producer_id: producer_id.clone(), + collector_id: producer_id.clone(), + materialization: schema.materialization, + schema_id: schema.schema_id.clone(), + }) + }) + .collect(); + let plan = Self { + envelope, + ingest: IngestContract { + protocol: IngestProtocol::ModifiedOtlpMetricsV1, + endpoint_path: "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/v1/metrics".into(), + timestamp_unit: TimestampUnit::UnixNanoseconds, + require_plan_identity: true, + require_materialization_identity: true, + require_registered_producer: true, + }, + schemas, + producers, + materializations, + }; + plan.validate()?; + Ok(plan) + } + + pub fn runtime_materializations( + &self, + ) -> Result, PrecomputePlanError> { + self.validate()?; + Ok(self + .materializations + .iter() + .cloned() + .map(|materialization| (materialization.policy_fp_u64(), materialization)) + .collect()) + } + + pub fn validate(&self) -> Result<(), PrecomputePlanError> { + if self.ingest.endpoint_path != "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/v1/metrics" + || !self.ingest.require_plan_identity + || !self.ingest.require_materialization_identity + || !self.ingest.require_registered_producer + { + return Err(PrecomputePlanError::UnsupportedIngestEndpoint); + } + let mut materializations = BTreeSet::new(); + for materialization in &self.materializations { + if !materializations.insert(materialization.policy_fingerprint()) { + return Err(PrecomputePlanError::DuplicateMaterialization( + materialization.policy_fp_u64(), + )); + } + } + let schemas: BTreeSet<_> = self + .schemas + .iter() + .map(|schema| schema.materialization) + .collect(); + if schemas != materializations || schemas.len() != self.schemas.len() { + return Err(PrecomputePlanError::SchemaSetMismatch); + } + for schema in &self.schemas { + if schema.schema_version == 0 || schema.encodings.is_empty() { + return Err(PrecomputePlanError::InvalidSchema { + schema_id: schema.schema_id.clone(), + }); + } + } + let schema_by_materialization: BTreeMap<_, _> = self + .schemas + .iter() + .map(|schema| (schema.materialization, schema.schema_id.as_str())) + .collect(); + let mut producers = BTreeSet::new(); + let mut produced = BTreeSet::new(); + for producer in &self.producers { + if !materializations.contains(&producer.materialization) + || schema_by_materialization + .get(&producer.materialization) + .copied() + != Some(producer.schema_id.as_str()) + { + return Err(PrecomputePlanError::InvalidProducer { + producer_id: producer.producer_id.clone(), + }); + } + let key = ( + producer.producer_id.as_str(), + producer.materialization, + producer.schema_id.as_str(), + ); + if !producers.insert(key) { + return Err(PrecomputePlanError::DuplicateProducer( + producer.producer_id.clone(), + )); + } + produced.insert(producer.materialization); + } + if let Some(missing) = materializations.difference(&produced).next() { + return Err(PrecomputePlanError::MissingProducer(missing.0)); + } + Ok(()) + } +} + /// Complete physical projection of one post-ASAP planning decision. /// All three child plans share the same envelope and are compiled together. #[derive(Debug, Clone)] @@ -321,7 +536,7 @@ impl PhysicalCompiler { } }; let aggregation_id = format!("{}:{}", query.query_id, metric); - aggregations.push(BackendAggregation { + let aggregation = BackendAggregation { aggregation_id: aggregation_id.clone(), metric_name: metric.clone(), family: SummaryFamilyType::Sketch( @@ -333,13 +548,18 @@ impl PhysicalCompiler { grouping: query.group_by.clone(), item_label: None, aggregation_input: AggregationInput::SketchEnvelope, - }); + }; + let precompute_materialization = + backend_plan::aggregation_config_for_materialization(&aggregation)?; + let materialization = precompute_materialization.policy_fingerprint(); + aggregations.push(aggregation); readouts.push(BackendReadout { aggregation_id, op: selected.readout.clone(), }); collector_materializations.push(CollectorMaterialization { query_id: query.query_id.clone(), + materialization, metric, algorithm: format!("{:?}", selected.kind.algorithm()).to_ascii_lowercase(), parameters: sketch_params_json(&selected.params), @@ -401,14 +621,25 @@ impl PhysicalCompiler { envelope: envelope.clone(), materializations: collector_materializations.clone(), }) - .collect(); - let precompute_plan = PrecomputePlan { - envelope: envelope.clone(), - materializations: aggregations - .iter() - .map(backend_plan::aggregation_config_for_materialization) - .collect::, _>>()?, - }; + .collect::>(); + let materializations = aggregations + .iter() + .map(backend_plan::aggregation_config_for_materialization) + .collect::, _>>()?; + let producer_ids = collector_plans + .iter() + .map(|plan| plan.collector_id.clone()) + .collect::>(); + let precompute_plan = PrecomputePlan::build( + envelope.clone(), + materializations, + &backend_plan, + &producer_ids, + ) + .map_err(|error| CompileError::Query { + query_id: "precompute-plan".into(), + reason: error.to_string(), + })?; let materialization_fingerprints: BTreeSet<_> = backend_plan.materializations.keys().copied().collect(); let mut query_entries = BTreeMap::new(); @@ -581,6 +812,25 @@ pub fn select_post_asap( ) } +fn state_schema_id(fingerprint: asap_types::PolicyFingerprint) -> String { + format!( + "{}:summary-state:v1:{}", + backend_plan::BACKEND_COMPAT, + fingerprint.0 + ) +} + +fn state_encodings(family: &SummaryFamilyType) -> Vec { + match family { + SummaryFamilyType::ExactAggregate(..) => vec![StateEncoding::ExactAccumulatorV1], + SummaryFamilyType::Sketch(..) => vec![ + StateEncoding::SketchlibProtobufV1, + StateEncoding::SketchCoreMsgpackV1, + ], + _ => Vec::new(), + } +} + fn validate_evidence( query_id: &str, evidence: &TopKMembershipEvidence, @@ -954,6 +1204,10 @@ mod tests { assert_eq!(bundle.collector_plans.len(), 2); assert_eq!(bundle.precompute_plan.envelope, bundle.envelope); assert_eq!(bundle.precompute_plan.materializations.len(), 1); + assert_eq!(bundle.precompute_plan.schemas.len(), 1); + assert_eq!(bundle.precompute_plan.producers.len(), 2); + assert_eq!(bundle.precompute_plan.ingest.endpoint_path, "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/v1/metrics"); + bundle.precompute_plan.validate().expect("runtime contract"); assert_eq!(bundle.backend_plan.materializations.len(), 1); assert_eq!( bundle @@ -1038,6 +1292,30 @@ mod tests { assert!(matches!(error, CompileError::Lifecycle { .. })); } + #[test] + fn precompute_plan_rejects_schema_or_producer_drift() { + let bundle = PhysicalCompiler + .compile( + request("q-quantile", "quantile_over_time(0.99, m[1m])"), + environment(10_000), + ) + .expect("compile"); + + let mut schema_drift = bundle.precompute_plan.clone(); + schema_drift.schemas[0].schema_version = 0; + assert!(matches!( + schema_drift.validate(), + Err(PrecomputePlanError::InvalidSchema { .. }) + )); + + let mut producer_drift = bundle.precompute_plan; + producer_drift.producers[0].schema_id = "wrong".into(); + assert!(matches!( + producer_drift.validate(), + Err(PrecomputePlanError::InvalidProducer { .. }) + )); + } + #[test] fn topk_fails_closed_without_membership_evidence() { assert!(request_with_evidence("q-topk", "topk(5, m)", None).is_err()); diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 66c896e08..382022b9d 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -5330,15 +5330,18 @@ async fn handle_post_physical_plan( ) .into_response(); }; - let new_config = crate::storage_engines::types::StreamingConfig::new( - request - .precompute_plan - .materializations - .iter() - .cloned() - .map(|config| (config.policy_fp_u64(), config)) - .collect(), - ); + let runtime_materializations = + match request.precompute_plan.runtime_materializations() { + Ok(materializations) => materializations, + Err(error) => return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", "error": format!("PrecomputePlan validation error: {error}") + })), + ) + .into_response(), + }; + let new_config = crate::storage_engines::types::StreamingConfig::new(runtime_materializations); let new_plan = match control_plane::backend_plan::BackendPlan::decode(&request.backend_plan) { Ok(plan) => plan, Err(error) => { @@ -5388,6 +5391,26 @@ async fn handle_post_physical_plan( ) .into_response(); } + for schema in &request.precompute_plan.schemas { + let Some(materialization) = new_plan.materializations.get(&schema.materialization) else { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", "error": "PrecomputePlan schema is absent from BackendPlan" + })), + ) + .into_response(); + }; + if materialization.kind != schema.family || materialization.params != schema.parameters { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", "error": "PrecomputePlan and BackendPlan schema semantics differ" + })), + ) + .into_response(); + } + } let typed_plan_fps: BTreeSet<_> = new_plan.materializations.keys().copied().collect(); if let Err(error) = request.query_plan.validate(&typed_plan_fps) { return ( diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 554c2a071..24937765d 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -435,6 +435,16 @@ async fn main() -> Result<()> { planner_revision: control_plane::physical::compiler::PLANNER_REVISION.into(), capability_snapshot_id: "bootstrap".into(), }, + ingest: control_plane::physical::compiler::IngestContract { + protocol: control_plane::physical::compiler::IngestProtocol::ModifiedOtlpMetricsV1, + endpoint_path: "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/v1/metrics".into(), + timestamp_unit: control_plane::physical::compiler::TimestampUnit::UnixNanoseconds, + require_plan_identity: false, + require_materialization_identity: false, + require_registered_producer: false, + }, + schemas: Vec::new(), + producers: Vec::new(), materializations: streaming_config .aggregation_configs .values() diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index abc4065bd..8e368b8ff 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -618,6 +618,18 @@ mod tests { ActivePhysicalPlan { precompute_plan: control_plane::physical::compiler::PrecomputePlan { envelope, + ingest: control_plane::physical::compiler::IngestContract { + protocol: + control_plane::physical::compiler::IngestProtocol::ModifiedOtlpMetricsV1, + endpoint_path: "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/v1/metrics".into(), + timestamp_unit: + control_plane::physical::compiler::TimestampUnit::UnixNanoseconds, + require_plan_identity: true, + require_materialization_identity: true, + require_registered_producer: true, + }, + schemas: Vec::new(), + producers: Vec::new(), materializations: Vec::new(), }, runtime_config: Arc::new(StreamingConfig::new(HashMap::new())), diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index 40129850a..948297261 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -120,6 +120,10 @@ pub struct DeploymentEnvironment { pub capability_snapshot_id: String, pub observed_at_unix_ms: u64, pub max_evidence_age_ms: u64, + pub plan_version: u64, + pub activation_unix_ms: u64, + pub expiry_unix_ms: Option, + pub backend_compat: String, } pub struct PhysicalPlan { @@ -138,13 +142,14 @@ Supporting public types: | `DeploymentEnvironment` | Target collector IDs, capability snapshot identity, planning time, and evidence freshness policy. | | `PlanEnvelope` | Shared deterministic `plan_id`, generation time, capability snapshot, and Planner revision. | | `CollectorPlan` | Serializable execution projection consumed by ASAPCollector. | -| `PrecomputePlan` | Config-driven aggregation/window projection consumed by the backend streaming precompute engine. | +| `PrecomputePlan` | Authoritative materialization, ingest, state-schema, and producer contract consumed directly by the backend runtime. | | `BackendPlan` | Versioned public data-plane materialization/routing contract defined in this repository. | +| `QueryPlan` | Canonical-query keyed executable DAG with exact materialization bindings and fallback policy. | -Current MVP limits are explicit: time-series sources and sketch -materializations are supported; table sources and non-sketch selected families -return `CompileError`. Runtime activation/expiry and richer topology placement -remain publication-layer work and are not claimed by this compiler API. +Current MVP limits are explicit: time-series sources and supported summary +materializations are accepted; table sources and unexecutable selected families +return `CompileError`. Versioned stage/activate/drain/retire lifecycle is part +of publication; richer topology placement remains follow-up work. Output definitions: @@ -152,7 +157,7 @@ Output definitions: | --- | --- | | `envelope` | Shared plan/version/activation/compatibility identity. | | `collector_plans` | One plan per targeted collector, following ASAPCollector's public CollectorPlan schema. | -| `precompute_plan` | Aggregation definitions emitted to `/api/v1/streaming-config`; contains no query-string jobs. | +| `precompute_plan` | Materializations plus `/v1/metrics` ingest semantics, typed state schemas/encodings, and allowed producers; it is installed directly and contains no query-string jobs. | | `backend_plan` | Matching data-plane materialization and routing contract. | | `query_plan` | Canonical query identity, explicit fallback policy, node-ID DAG, and exact per-node materialization bindings. | diff --git a/docs/developer_docs/control-plane/plan-publication.md b/docs/developer_docs/control-plane/plan-publication.md index 322f16f7d..712ab4456 100644 --- a/docs/developer_docs/control-plane/plan-publication.md +++ b/docs/developer_docs/control-plane/plan-publication.md @@ -13,12 +13,15 @@ typed `BackendPlan` plus one target-specific `CollectorPlan` per Collector. ingestion rate/freshness, optimization horizon, and primitive state costs); - `collector_ids`: the required OpAMP agent IDs; - `capability_snapshot_id` and the exact `planner_revision`; +- `plan_version`, activation/optional expiry timestamps, and the exact backend + compatibility identity; - optional per-query TopK evidence, with `max_evidence_age_ms`; and - `apply_timeout_ms` (default 10000). Unknown JSON fields, empty target/query sets, zero windows/timeouts, stale evidence, and a Planner revision mismatch are rejected. The response is only -successful after every target has applied the same generated `plan_id`. +successful only after every target has applied the exact generated +`(plan_id, plan_version)` and the backend activates that generation. The physical compiler passes normalized recurrence, data-arrival evidence, runtime capabilities, and costs to ASAPPlanner's summary-maintenance lifecycle @@ -36,25 +39,24 @@ validate request and compile one bundle preflight every Collector capability | v -POST typed BackendPlan protobuf; require 2xx +POST one atomic PrecomputePlan + BackendPlan + QueryPlan bundle (stage) | v publish target-specific CollectorPlans over OpAMP | v -require exact (agent_id, plan_id, APPLIED) from every target +require exact (agent_id, plan_id, plan_version, APPLIED) from every target + | + v +wait until activation and atomically activate backend snapshot ``` -Preflight happens before backend mutation. Publication repeats capability -validation to close disconnect races. A missing backend endpoint, backend -non-2xx, Collector disconnect, timeout, malformed report, wrong plan ID, or -`FAILED` status fails the request. This path intentionally does not inherit the -legacy replanner's best-effort behavior. - -The MVP endpoint installs the backend before enabling new producers. It does -not claim distributed atomic activation or rollback; those remain post-MVP -work. If a Collector fails after backend installation, the backend has a -superset accepting view but the request fails and no success is reported. +Preflight happens before staging. Publication repeats capability validation to +close disconnect races. A missing backend endpoint, backend non-2xx, Collector +disconnect, timeout, malformed report, wrong identity/version, incompatible +schema, or `FAILED` status fails the request. A failed rollout leaves the +previous active generation untouched; a staged generation never serves before +explicit activation. ## OpAMP custom capability @@ -71,7 +73,7 @@ The exact capability and message types shared with ASAPCollector are: status payload is strict JSON: ```json -{"plan_id": 42, "status": "APPLIED", "error": null} +{"plan_id": 42, "plan_version": 3, "status": "APPLIED", "error": null} ``` `status` is exactly `APPLIED` or `FAILED`. Transport/config acknowledgements @@ -81,8 +83,8 @@ another Collector's plan. ## Backend wire contract -The matching `BackendPlan` uses the protobuf contract documented in -`control_plane/docs/design-backend-plan-wire-format.md` and is sent to -`POST /api/v1/backend-plan` with `application/x-protobuf`. Both projections -carry the same numeric `plan_id`; the compiler, not either transport, owns the -materialization choice. +The matching BackendPlan protobuf is embedded with the typed PrecomputePlan +and QueryPlan in `POST /api/v1/physical-plan`. The backend validates all shared +identities, fingerprints, schemas, parameters, producers, and lifecycle fields +before returning `staged`; `POST /api/v1/physical-plan/activate` performs the +single immutable-snapshot swap. From e4f52b4a46458ee2aa74aa94cdda8739717b6d15 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 13:58:23 -0600 Subject: [PATCH 2/3] fix(precompute): validate and deduplicate runtime contracts --- control_plane/src/physical/compiler.rs | 100 ++++++++++++++++++- data_plane/src/drivers/query/servers/http.rs | 9 ++ 2 files changed, 104 insertions(+), 5 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 21ae7cdab..c2fdde956 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -244,9 +244,21 @@ pub struct StateSchemaContract { pub schema_version: u32, pub materialization: asap_types::PolicyFingerprint, pub family: SummaryFamilyType, + pub source: Source, + pub value_column: planner_types::pre_asap::ColumnRef, + pub group_by: Vec, + pub window: StateWindowContract, pub encodings: Vec, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct StateWindowContract { + pub kind: asap_types::WindowKind, + pub size_ms: u64, + pub slide_ms: Option, +} + /// A collector authorized to produce state for one materialization. Runtime /// producer epochs and frame sequences belong to TransmissionPlan. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] @@ -302,6 +314,14 @@ impl PrecomputePlan { schema_version: 1, materialization: *fingerprint, family: materialization.family.clone(), + source: materialization.source.clone(), + value_column: materialization.col.clone(), + group_by: materialization.group_by.clone(), + window: StateWindowContract { + kind: materialization.window.kind, + size_ms: materialization.window.size_ms, + slide_ms: materialization.window.slide_ms, + }, encodings: state_encodings(&materialization.family), }) .collect::>(); @@ -330,7 +350,7 @@ impl PrecomputePlan { producers, materializations, }; - plan.validate()?; + plan.validate_against_backend(backend_plan)?; Ok(plan) } @@ -412,6 +432,33 @@ impl PrecomputePlan { } Ok(()) } + + pub fn validate_against_backend( + &self, + backend_plan: &BackendPlan, + ) -> Result<(), PrecomputePlanError> { + self.validate()?; + for schema in &self.schemas { + let Some(materialization) = backend_plan.materializations.get(&schema.materialization) + else { + return Err(PrecomputePlanError::SchemaSetMismatch); + }; + if schema.family != materialization.family + || schema.schema_id != state_schema_id(schema.materialization) + || schema.source != materialization.source + || schema.value_column != materialization.col + || schema.group_by != materialization.group_by + || schema.window.kind != materialization.window.kind + || schema.window.size_ms != materialization.window.size_ms + || schema.window.slide_ms != materialization.window.slide_ms + { + return Err(PrecomputePlanError::InvalidSchema { + schema_id: schema.schema_id.clone(), + }); + } + } + Ok(()) + } } /// Complete physical projection of one post-ASAP planning decision. @@ -622,10 +669,17 @@ impl PhysicalCompiler { materializations: collector_materializations.clone(), }) .collect::>(); - let materializations = aggregations - .iter() - .map(backend_plan::aggregation_config_for_materialization) - .collect::, _>>()?; + // Several queries/readouts may intentionally share one maintained + // summary. PrecomputePlan is keyed by physical identity, not query ID. + let mut materializations_by_fingerprint = BTreeMap::new(); + for aggregation in &aggregations { + let materialization = + backend_plan::aggregation_config_for_materialization(aggregation)?; + materializations_by_fingerprint + .entry(materialization.policy_fingerprint()) + .or_insert(materialization); + } + let materializations = materializations_by_fingerprint.into_values().collect(); let producer_ids = collector_plans .iter() .map(|plan| plan.collector_id.clone()) @@ -1282,6 +1336,42 @@ mod tests { } } + #[test] + fn multiple_readouts_share_one_precompute_materialization() { + let mut planning_request = request("q-p90", "quantile_over_time(0.90, m[1m])"); + let second = request("q-p99", "quantile_over_time(0.99, m[1m])") + .queries + .into_iter() + .next() + .unwrap(); + planning_request.queries.push(second); + let bundle = PhysicalCompiler + .compile(planning_request, environment(10_000)) + .unwrap(); + + assert_eq!(bundle.query_plan.entries.len(), 2); + assert_eq!(bundle.backend_plan.materializations.len(), 1); + assert_eq!(bundle.precompute_plan.materializations.len(), 1); + assert_eq!(bundle.precompute_plan.schemas.len(), 1); + assert_eq!(bundle.precompute_plan.producers.len(), 2); + } + + #[test] + fn precompute_schema_must_match_backend_semantics() { + let bundle = PhysicalCompiler + .compile( + request("q-quantile", "quantile_over_time(0.99, m[1m])"), + environment(10_000), + ) + .unwrap(); + let mut plan = bundle.precompute_plan.clone(); + plan.schemas[0].group_by.push("invented".into()); + assert!(matches!( + plan.validate_against_backend(&bundle.backend_plan), + Err(PrecomputePlanError::InvalidSchema { .. }) + )); + } + #[test] fn missing_window_implementation_evidence_fails_closed() { let mut request = request("q-window", "quantile_over_time(0.99, m[1m])"); diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 382022b9d..03bb8ac82 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -5363,6 +5363,15 @@ async fn handle_post_physical_plan( ) .into_response(); } + if let Err(error) = request.precompute_plan.validate_against_backend(&new_plan) { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", "error": format!("PrecomputePlan validation error: {error}") + })), + ) + .into_response(); + } if request.query_plan.plan_id != new_plan.plan_id || request.query_plan.plan_version != new_plan.plan_version || request.precompute_plan.envelope.plan_id != new_plan.plan_id From 11306d0214828f75a9a74f57040251b5f5ca0e1b Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 20:56:32 -0600 Subject: [PATCH 3/3] fix(stack): serialize precompute state identity as a wire DTO --- control_plane/src/physical/compiler.rs | 97 ++++++++++++++++---- data_plane/src/drivers/query/servers/http.rs | 4 +- 2 files changed, 80 insertions(+), 21 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index c2fdde956..ca2c2c2bb 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -14,9 +14,10 @@ use asap_aware_mapping::{ SummaryMaintenanceLifecycleCapabilities, SummaryMaintenanceLifecycleCostInputs, WorkloadDemand, }; use planner_types::post_asap::{ - CompositionOperator, EvaluationSchedule, OutputRepresentation, SketchQuery, SummaryExpr, - SummaryFamilyType, SummaryMaintenanceLifecycle, SummaryMaintenanceLifecycleGuarantee, - SummaryMaintenanceMode, SummaryNode, SummaryWindowFramework, + CompositionOperator, EvaluationSchedule, OutputRepresentation, SketchAlgorithm, SketchParams, + SketchQuery, SummaryExpr, SummaryFamilyType, SummaryMaintenanceLifecycle, + SummaryMaintenanceLifecycleGuarantee, SummaryMaintenanceMode, SummaryNode, + SummaryWindowFramework, }; use planner_types::pre_asap::QueryExpr; use planner_types::workload::{ @@ -236,6 +237,54 @@ pub enum StateEncoding { ExactAccumulatorV1, } +/// Serializable physical state identity derived from Planner's canonical +/// summary family. This is a wire DTO, not a second planning algebra. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "family", rename_all = "snake_case", deny_unknown_fields)] +pub enum StateFamilyContract { + Exact { + kind: ExactStateKind, + }, + Sketch { + algorithm: SketchAlgorithm, + parameters: SketchParams, + }, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ExactStateKind { + Sum, + Count, + MinMax, + Increase, + Rate, +} + +impl TryFrom<&SummaryFamilyType> for StateFamilyContract { + type Error = (); + + fn try_from(family: &SummaryFamilyType) -> Result { + use planner_types::post_asap::ExactKind; + Ok(match family { + SummaryFamilyType::ExactAggregate(kind, _) => Self::Exact { + kind: match kind { + ExactKind::Sum => ExactStateKind::Sum, + ExactKind::Count => ExactStateKind::Count, + ExactKind::MinMax => ExactStateKind::MinMax, + ExactKind::Increase => ExactStateKind::Increase, + ExactKind::Rate => ExactStateKind::Rate, + }, + }, + SummaryFamilyType::Sketch(kind, _) => Self::Sketch { + algorithm: kind.algorithm().clone(), + parameters: kind.params().clone(), + }, + _ => return Err(()), + }) + } +} + /// Decoder/schema contract for one content-addressed materialization. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] @@ -243,7 +292,7 @@ pub struct StateSchemaContract { pub schema_id: String, pub schema_version: u32, pub materialization: asap_types::PolicyFingerprint, - pub family: SummaryFamilyType, + pub family: StateFamilyContract, pub source: Source, pub value_column: planner_types::pre_asap::ColumnRef, pub group_by: Vec, @@ -282,6 +331,8 @@ pub enum PrecomputePlanError { SchemaSetMismatch, #[error("schema {schema_id} has invalid version or no encoding")] InvalidSchema { schema_id: String }, + #[error("materialization {0} uses a summary family unsupported by the runtime schema")] + UnsupportedFamily(u64), #[error("producer {producer_id} references an unknown materialization or schema")] InvalidProducer { producer_id: String }, #[error("materialization {0} has no registered producer")] @@ -309,22 +360,26 @@ impl PrecomputePlan { let schemas = backend_plan .materializations .iter() - .map(|(fingerprint, materialization)| StateSchemaContract { - schema_id: state_schema_id(*fingerprint), - schema_version: 1, - materialization: *fingerprint, - family: materialization.family.clone(), - source: materialization.source.clone(), - value_column: materialization.col.clone(), - group_by: materialization.group_by.clone(), - window: StateWindowContract { - kind: materialization.window.kind, - size_ms: materialization.window.size_ms, - slide_ms: materialization.window.slide_ms, - }, - encodings: state_encodings(&materialization.family), + .map(|(fingerprint, materialization)| { + let family = StateFamilyContract::try_from(&materialization.family) + .map_err(|_| PrecomputePlanError::UnsupportedFamily(fingerprint.0))?; + Ok(StateSchemaContract { + schema_id: state_schema_id(*fingerprint), + schema_version: 1, + materialization: *fingerprint, + family, + source: materialization.source.clone(), + value_column: materialization.col.clone(), + group_by: materialization.group_by.clone(), + window: StateWindowContract { + kind: materialization.window.kind, + size_ms: materialization.window.size_ms, + slide_ms: materialization.window.slide_ms, + }, + encodings: state_encodings(&materialization.family), + }) }) - .collect::>(); + .collect::, PrecomputePlanError>>()?; let producers = producer_ids .iter() .flat_map(|producer_id| { @@ -443,7 +498,9 @@ impl PrecomputePlan { else { return Err(PrecomputePlanError::SchemaSetMismatch); }; - if schema.family != materialization.family + let expected_family = StateFamilyContract::try_from(&materialization.family) + .map_err(|_| PrecomputePlanError::UnsupportedFamily(schema.materialization.0))?; + if schema.family != expected_family || schema.schema_id != state_schema_id(schema.materialization) || schema.source != materialization.source || schema.value_column != materialization.col diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 03bb8ac82..8030f65f7 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -5410,7 +5410,9 @@ async fn handle_post_physical_plan( ) .into_response(); }; - if materialization.kind != schema.family || materialization.params != schema.parameters { + if control_plane::physical::compiler::StateFamilyContract::try_from(&materialization.family) + != Ok(schema.family.clone()) + { return ( StatusCode::UNPROCESSABLE_ENTITY, axum::Json(serde_json::json!({