diff --git a/control_plane/examples/audit_clickhouse_corpus.rs b/control_plane/examples/audit_clickhouse_corpus.rs index 71fedf79e..e7497b8d1 100644 --- a/control_plane/examples/audit_clickhouse_corpus.rs +++ b/control_plane/examples/audit_clickhouse_corpus.rs @@ -4,7 +4,7 @@ use asap_frontend_sql::SqlCatalog; use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; use control_plane::clickhouse::{ClickHouseSqlWorkload, ClickHouseSqlWorkloadEntry}; use control_plane::physical::compiler::{ - PlanEnvelope, PrecomputePlan, TransmissionPlan, BACKEND_COMPAT, PLANNER_REVISION, + PlanEnvelope, PrecomputePlan, BACKEND_COMPAT, PLANNER_REVISION, }; use planner_types::pre_asap::{Column, DataType, Schema}; use planner_types::types::AccuracyTarget; @@ -56,8 +56,12 @@ fn publication_inputs(schema: &Schema, sql: String) -> ClickHouseSqlWorkload { let mut precompute_plan = PrecomputePlan::build_backend_local(envelope.clone(), vec![materialization.clone()]) .unwrap(); - let mut transmission_plan = - TransmissionPlan::build(envelope, &precompute_plan, &Default::default()).unwrap(); + let mut transmission_plan = control_plane::physical::compiler::compile_transmission_plan( + envelope, + &precompute_plan, + &Default::default(), + ) + .unwrap(); let sds = asap_types::summary_catalog::SummaryCatalog::from_materializations( 27, 1, diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 533219378..f1334574f 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -183,7 +183,7 @@ pub async fn compile_automatic_clickhouse_workload( .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?, ); precompute.executable_dags = installed_dags; - let mut transmission = TransmissionPlan::build( + let mut transmission = crate::physical::compiler::compile_transmission_plan( request.envelope.clone(), &precompute, &std::collections::BTreeMap::new(), @@ -1010,9 +1010,12 @@ mod tests { let mut precompute = PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap(); precompute.summary_catalog = Some(sds.reference().unwrap()); - let mut transmission = - TransmissionPlan::build(envelope, &precompute, &std::collections::BTreeMap::new()) - .unwrap(); + let mut transmission = crate::physical::compiler::compile_transmission_plan( + envelope, + &precompute, + &std::collections::BTreeMap::new(), + ) + .unwrap(); transmission.summary_catalog = Some(sds.reference().unwrap()); let timestamped = |time_name: &str, value_name: &str| { Schema::with_time_index( @@ -1204,7 +1207,7 @@ mod tests { request.precompute_plan = PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap(); request.precompute_plan.summary_catalog = Some(request.sds.reference().unwrap()); - request.transmission_plan = TransmissionPlan::build( + request.transmission_plan = crate::physical::compiler::compile_transmission_plan( envelope, &request.precompute_plan, &std::collections::BTreeMap::new(), diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index d3f4861de..9be93c761 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -260,7 +260,7 @@ async fn push_documents_coupled( warn!(%error, "failed to bind compatibility PrecomputePlan to SummaryCatalog"); return (false, false, 0); } - let transmission_plan = match crate::physical::compiler::TransmissionPlan::build( + let transmission_plan = match crate::physical::compiler::compile_transmission_plan( precompute_plan.envelope.clone(), &precompute_plan, &Default::default(), diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 6026ac86a..26482cc23 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -258,47 +258,136 @@ pub use asap_types::precompute_plan::{ StateWindowContract, TimestampUnit, }; -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -pub struct CollectorMaterialization { - pub query_id: String, - pub materialization: asap_types::sds::SummaryDefinitionId, - pub metric: String, - pub algorithm: String, - pub parameters: Value, - pub group_by: Vec, - pub window_secs: u64, - pub abstract_window_framework: SummaryWindowFramework, - pub window_implementation_id: String, - pub slide_secs: u64, - #[serde( - default, - alias = "paneOriginMs", - skip_serializing_if = "Option::is_none" - )] - pub pane_origin_ms: Option, - pub window_layout: asap_types::WindowMaterializationLayout, - pub evidence_source: Option, - pub lifecycle: CollectorLifecycle, +pub use asap_types::producer_plan::{ + AdaptiveF64Bounds, AdaptiveU64Bounds, CollectorLifecycle, CollectorMaterialization, + CollectorPlan, DeltaPolicy, FrameIdentityContract, GosPolicy, GosThresholdMode, + RuntimeAdaptationEvidence, RuntimeAdaptationPolicy, RuntimeRulePolicy, SamplingEstimator, + SamplingPolicy, SequenceScope, SummaryFrameIdentity, SummaryFrameKind, TransmissionMode, + TransmissionPlan, TransmissionPlanError, TransmissionRule, +}; + +/// Build the fixed physical knob from the controller's canonical +/// epsilon-floor allocator. Degenerate budgets/rates disable sampling. +pub fn sampling_policy_from_accuracy_budget( + epsilon_sampling: f64, + updates_per_window: f64, + estimator: SamplingEstimator, +) -> SamplingPolicy { + if !epsilon_sampling.is_finite() + || epsilon_sampling <= 0.0 + || !updates_per_window.is_finite() + || updates_per_window <= 0.0 + { + return SamplingPolicy::Disabled; + } + let probability = crate::epsilon_alloc::derive_sample_p(epsilon_sampling, updates_per_window); + if probability >= 1.0 { + SamplingPolicy::Disabled + } else { + SamplingPolicy::Fixed { + probability, + estimator, + } + } } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct CollectorLifecycle { - pub kind: String, - pub maintenance_mode: String, - pub evaluation_schedule: String, - pub output_representation: String, +/// Allocate the deterministic staleness share with the same linear-peel +/// composition used by `epsilon_alloc`. `None` means the selected sketch +/// already consumes the budget or communication has no allocated weight. +pub fn gos_policy_from_accuracy_budget( + epsilon_total: f64, + epsilon_sketch: f64, + sites: u32, + edge_cpu_weight: f64, + communication_weight: f64, + threshold_mode: GosThresholdMode, +) -> Option { + if !epsilon_total.is_finite() + || !epsilon_sketch.is_finite() + || !(0.0..=1.0).contains(&epsilon_total) + || epsilon_sketch < 0.0 + || !edge_cpu_weight.is_finite() + || edge_cpu_weight < 0.0 + || !communication_weight.is_finite() + || communication_weight <= 0.0 + { + return None; + } + let (_, epsilon_staleness) = crate::epsilon_alloc::split_budget( + epsilon_total, + epsilon_sketch, + edge_cpu_weight, + communication_weight, + ); + (epsilon_staleness.is_finite() && epsilon_staleness > 0.0).then_some(GosPolicy { + epsilon_staleness, + sites: sites.max(1), + threshold_mode, + }) } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -pub struct CollectorPlan { - /// Absent only in legacy artifacts; catalog-aware validation requires it. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub summary_catalog: Option, - pub collector_id: String, - pub envelope: PlanEnvelope, - pub materializations: Vec, - pub transmission_rules: Vec, +pub fn compile_transmission_plan( + envelope: PlanEnvelope, + precompute: &PrecomputePlan, + runtime_policies: &BTreeMap, +) -> Result { + if envelope != precompute.envelope { + return Err(TransmissionPlanError::EnvelopeMismatch); + } + let schemas: BTreeMap<_, _> = precompute + .schemas + .iter() + .map(|schema| (schema.materialization, schema)) + .collect(); + let rules = precompute + .producers + .iter() + .map(|producer| { + let schema = schemas + .get(&producer.materialization) + .expect("validated PrecomputePlan schema binding"); + let materialization = precompute + .materializations + .iter() + .find(|m| m.policy_fingerprint() == producer.materialization.fingerprint()) + .expect("validated PrecomputePlan materialization binding"); + let runtime_policy = runtime_policies + .get(&producer.materialization.fingerprint()) + .cloned() + .unwrap_or_default(); + let mode = if runtime_policy.delta.is_some() { + TransmissionMode::Delta + } else { + TransmissionMode::Full + }; + let emit_every_ms = materialization.window_size.saturating_mul(1_000); + TransmissionRule { + materialization: producer.materialization, + producer_id: producer.producer_id.clone(), + schema_id: producer.schema_id.clone(), + mode, + encoding: schema.encodings[0].clone(), + emit_every_ms, + full_checkpoint_every_ms: (mode == TransmissionMode::Delta) + .then(|| emit_every_ms.saturating_mul(10)), + destination_ref: "asapquery-backend".into(), + runtime_policy, + } + }) + .collect(); + let plan = TransmissionPlan { + summary_catalog: precompute.summary_catalog.clone(), + envelope, + frame_identity: FrameIdentityContract { + identity_version: 1, + sequence_scope: SequenceScope::MaterializationSeriesProducerEpoch, + require_checkpoint_for_full: true, + require_base_checkpoint_for_delta: true, + }, + rules, + }; + plan.validate(precompute)?; + Ok(plan) } /// Complete physical projection of one post-ASAP planning decision. @@ -327,905 +416,6 @@ pub struct MaterializationLifecycleEstimate { pub lifecycle_cost: f64, } -#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] -#[serde(rename_all = "snake_case")] -pub enum TransmissionMode { - Full, - Delta, -} - -#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum SequenceScope { - MaterializationSeriesProducerEpoch, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct FrameIdentityContract { - pub identity_version: u32, - pub sequence_scope: SequenceScope, - pub require_checkpoint_for_full: bool, - pub require_base_checkpoint_for_delta: bool, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(deny_unknown_fields)] -pub struct TransmissionRule { - pub materialization: asap_types::sds::SummaryDefinitionId, - pub producer_id: String, - pub schema_id: String, - pub mode: TransmissionMode, - pub encoding: StateEncoding, - pub emit_every_ms: u64, - pub full_checkpoint_every_ms: Option, - pub destination_ref: String, - /// Plan-owned runtime knobs. These values are part of the immutable plan - /// generation; live feedback may only change them by publishing a - /// successor generation accepted by [`TransmissionPlan::authorize_successor`]. - #[serde(default)] - pub runtime_policy: RuntimeRulePolicy, -} - -/// How the collector admits updates before sketch maintenance. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(tag = "mode", rename_all = "snake_case", deny_unknown_fields)] -pub enum SamplingPolicy { - Disabled, - Fixed { - /// Probability in `(0, 1]`; `1` is valid but should normally be - /// represented by `Disabled`. - probability: f64, - estimator: SamplingEstimator, - }, -} - -impl Default for SamplingPolicy { - fn default() -> Self { - Self::Disabled - } -} - -impl SamplingPolicy { - /// Build the fixed physical knob from the controller's canonical - /// epsilon-floor allocator. Degenerate budgets/rates disable sampling. - pub fn from_accuracy_budget( - epsilon_sampling: f64, - updates_per_window: f64, - estimator: SamplingEstimator, - ) -> Self { - if !epsilon_sampling.is_finite() - || epsilon_sampling <= 0.0 - || !updates_per_window.is_finite() - || updates_per_window <= 0.0 - { - return Self::Disabled; - } - let probability = - crate::epsilon_alloc::derive_sample_p(epsilon_sampling, updates_per_window); - if probability >= 1.0 { - Self::Disabled - } else { - Self::Fixed { - probability, - estimator, - } - } - } -} - -#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum SamplingEstimator { - /// Hash-threshold element sampling, used by cardinality summaries. - HashThreshold, - /// Geometric admission/Nitro-style update sampling, used by frequency - /// summaries. The sketch readout carries the corresponding correction. - GeometricAdmission, -} - -/// Norm-adaptive Group-of-Sketches delta gating. GOS is meaningful only for -/// CountSketch families and only when the transmission rule is in delta mode. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(deny_unknown_fields)] -pub struct GosPolicy { - pub epsilon_staleness: f64, - pub sites: u32, - pub threshold_mode: GosThresholdMode, -} - -impl GosPolicy { - /// Allocate the deterministic staleness share with the same linear-peel - /// composition used by `epsilon_alloc`. `None` means the selected sketch - /// already consumes the budget or communication has no allocated weight. - pub fn from_accuracy_budget( - epsilon_total: f64, - epsilon_sketch: f64, - sites: u32, - edge_cpu_weight: f64, - communication_weight: f64, - threshold_mode: GosThresholdMode, - ) -> Option { - if !epsilon_total.is_finite() - || !epsilon_sketch.is_finite() - || !(0.0..=1.0).contains(&epsilon_total) - || epsilon_sketch < 0.0 - || !edge_cpu_weight.is_finite() - || edge_cpu_weight < 0.0 - || !communication_weight.is_finite() - || communication_weight <= 0.0 - { - return None; - } - let (_, epsilon_staleness) = crate::epsilon_alloc::split_budget( - epsilon_total, - epsilon_sketch, - edge_cpu_weight, - communication_weight, - ); - (epsilon_staleness.is_finite() && epsilon_staleness > 0.0).then_some(Self { - epsilon_staleness, - sites: sites.max(1), - threshold_mode, - }) - } -} - -#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum GosThresholdMode { - Isotropic, - Anisotropic, -} - -/// Sparse-delta semantics within a delta transmission rule. An absolute -/// threshold of zero sends every changed cell. When `gos` is present it -/// replaces the fixed threshold with the GOS norm-adaptive threshold. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(deny_unknown_fields)] -pub struct DeltaPolicy { - pub absolute_threshold: f64, - pub gos: Option, -} - -/// Inclusive bounds for one floating-point adaptation knob. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(deny_unknown_fields)] -pub struct AdaptiveF64Bounds { - pub min: f64, - pub max: f64, - pub max_step: f64, -} - -/// Inclusive bounds for one integer adaptation knob. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct AdaptiveU64Bounds { - pub min: u64, - pub max: u64, - pub max_step: u64, -} - -/// Guardrails for telemetry-driven runtime adaptation. This is an -/// authorization contract, not an instruction to mutate the active plan. -/// Every accepted change becomes a staged successor PhysicalPlan. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(deny_unknown_fields)] -pub struct RuntimeAdaptationPolicy { - pub enabled: bool, - pub not_before_unix_ms: u64, - pub max_evidence_age_ms: u64, - pub min_evidence_samples: u64, - pub sample_probability: Option, - pub emit_every_ms: Option, - pub delta_threshold: Option, - pub gos_epsilon_staleness: Option, -} - -impl Default for RuntimeAdaptationPolicy { - fn default() -> Self { - Self { - enabled: false, - not_before_unix_ms: 0, - max_evidence_age_ms: 0, - min_evidence_samples: 0, - sample_probability: None, - emit_every_ms: None, - delta_threshold: None, - gos_epsilon_staleness: None, - } - } -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] -#[serde(deny_unknown_fields)] -pub struct RuntimeRulePolicy { - #[serde(default)] - pub sampling: SamplingPolicy, - pub delta: Option, - #[serde(default)] - pub adaptation: RuntimeAdaptationPolicy, -} - -/// Identity and sufficiency information for evidence authorizing one rule's -/// successor knobs. Raw measurements remain in the runtime-samples store; the -/// authorization boundary needs only their exact provenance and sample count. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct RuntimeAdaptationEvidence { - pub plan_id: u64, - pub plan_version: u64, - pub materialization: asap_types::sds::SummaryDefinitionId, - pub producer_id: String, - pub schema_id: String, - pub producer_version: String, - pub observed_at_unix_ms: u64, - pub sample_count: u64, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(deny_unknown_fields)] -pub struct TransmissionPlan { - /// Absent only in legacy artifacts; catalog-aware validation requires it. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub summary_catalog: Option, - pub envelope: PlanEnvelope, - pub frame_identity: FrameIdentityContract, - pub rules: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum SummaryFrameKind { - Full, - Delta, -} - -/// Identity attached to every summary record. Window bounds come from the -/// data point; the remaining fields are carried as reserved `asap.frame.*` -/// attributes until the modified-OTLP schema gains a dedicated message. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct SummaryFrameIdentity { - pub identity_version: u32, - pub plan_id: u64, - pub plan_version: u64, - pub backend_compat: String, - pub materialization: asap_types::sds::SummaryDefinitionId, - /// Canonical producer-side identity for one concrete retained-label group. - pub series_identity: String, - pub schema_id: String, - pub producer_id: String, - pub producer_epoch: String, - pub window_start_unix_nano: u64, - pub window_end_unix_nano: u64, - pub sequence: u64, - pub kind: SummaryFrameKind, - pub encoding: StateEncoding, - pub checkpoint_id: Option, - pub base_checkpoint_id: Option, -} - -#[derive(Debug, Error, PartialEq, Eq)] -pub enum TransmissionPlanError { - #[error("summary catalog mismatch: {0}")] - Catalog(String), - #[error("TransmissionPlan envelope differs from PrecomputePlan")] - EnvelopeMismatch, - #[error("transmission rules do not exactly match precompute producer bindings")] - ProducerSetMismatch, - #[error("invalid transmission rule for producer {0}")] - InvalidRule(String), - #[error("frame identity is invalid: {0}")] - InvalidFrame(String), - #[error("frame has no matching transmission rule")] - UnknownFrame, - #[error("runtime policy for producer {producer_id} is invalid: {reason}")] - InvalidRuntimePolicy { producer_id: String, reason: String }, - #[error("runtime adaptation successor is invalid: {0}")] - InvalidSuccessor(String), - #[error("runtime adaptation evidence for producer {0} is missing or invalid")] - InvalidAdaptationEvidence(String), - #[error("runtime adaptation for producer {producer_id} exceeds guardrails: {knob}")] - AdaptationOutOfBounds { - producer_id: String, - knob: &'static str, - }, -} - -fn validate_catalog_projection( - reference: Option<&asap_types::sds::CatalogGeneration>, - envelope: &PlanEnvelope, - materializations: impl IntoIterator, - catalog: &super::summary_catalog::SummaryCatalog, -) -> Result<(), TransmissionPlanError> { - let expected = catalog - .reference() - .map_err(|error| TransmissionPlanError::Catalog(error.to_string()))?; - if reference != Some(&expected) - || envelope.plan_id != catalog.plan_id - || envelope.plan_version != catalog.plan_version - { - return Err(TransmissionPlanError::Catalog( - "missing or different snapshot reference".into(), - )); - } - for id in materializations { - if !catalog.materializations.contains_key(&id) { - return Err(TransmissionPlanError::Catalog(format!( - "unknown materialization {}", - id.as_u64() - ))); - } - } - Ok(()) -} - -impl CollectorPlan { - pub fn validate_against_catalog( - &self, - catalog: &super::summary_catalog::SummaryCatalog, - ) -> Result<(), TransmissionPlanError> { - validate_catalog_projection( - self.summary_catalog.as_ref(), - &self.envelope, - self.materializations - .iter() - .map(|m| m.materialization) - .chain(self.transmission_rules.iter().map(|r| r.materialization)), - catalog, - ) - } -} - -impl TransmissionPlan { - pub fn validate_against_catalog( - &self, - catalog: &super::summary_catalog::SummaryCatalog, - ) -> Result<(), TransmissionPlanError> { - validate_catalog_projection( - self.summary_catalog.as_ref(), - &self.envelope, - self.rules.iter().map(|r| r.materialization), - catalog, - ) - } - - pub fn build( - envelope: PlanEnvelope, - precompute: &PrecomputePlan, - runtime_policies: &BTreeMap, - ) -> Result { - if envelope != precompute.envelope { - return Err(TransmissionPlanError::EnvelopeMismatch); - } - let schemas: BTreeMap<_, _> = precompute - .schemas - .iter() - .map(|schema| (schema.materialization, schema)) - .collect(); - let rules = precompute - .producers - .iter() - .map(|producer| { - let schema = schemas - .get(&producer.materialization) - .expect("validated PrecomputePlan schema binding"); - let materialization = precompute - .materializations - .iter() - .find(|m| m.policy_fingerprint() == producer.materialization.fingerprint()) - .expect("validated PrecomputePlan materialization binding"); - let runtime_policy = runtime_policies - .get(&producer.materialization.fingerprint()) - .cloned() - .unwrap_or_default(); - let mode = if runtime_policy.delta.is_some() { - TransmissionMode::Delta - } else { - TransmissionMode::Full - }; - let emit_every_ms = materialization.window_size.saturating_mul(1_000); - TransmissionRule { - materialization: producer.materialization, - producer_id: producer.producer_id.clone(), - schema_id: producer.schema_id.clone(), - mode, - encoding: schema.encodings[0].clone(), - emit_every_ms, - full_checkpoint_every_ms: (mode == TransmissionMode::Delta) - .then(|| emit_every_ms.saturating_mul(10)), - destination_ref: "asapquery-backend".into(), - runtime_policy, - } - }) - .collect(); - let plan = Self { - summary_catalog: precompute.summary_catalog.clone(), - envelope, - frame_identity: FrameIdentityContract { - identity_version: 1, - sequence_scope: SequenceScope::MaterializationSeriesProducerEpoch, - require_checkpoint_for_full: true, - require_base_checkpoint_for_delta: true, - }, - rules, - }; - plan.validate(precompute)?; - Ok(plan) - } - - pub fn validate(&self, precompute: &PrecomputePlan) -> Result<(), TransmissionPlanError> { - if self.summary_catalog != precompute.summary_catalog { - return Err(TransmissionPlanError::Catalog( - "transmission and precompute plans reference different catalog snapshots".into(), - )); - } - if self.envelope != precompute.envelope { - return Err(TransmissionPlanError::EnvelopeMismatch); - } - let expected: BTreeSet<_> = precompute - .producers - .iter() - .map(|producer| { - ( - producer.materialization, - producer.producer_id.as_str(), - producer.schema_id.as_str(), - ) - }) - .collect(); - let actual: BTreeSet<_> = self - .rules - .iter() - .map(|rule| { - ( - rule.materialization, - rule.producer_id.as_str(), - rule.schema_id.as_str(), - ) - }) - .collect(); - if expected != actual || actual.len() != self.rules.len() { - return Err(TransmissionPlanError::ProducerSetMismatch); - } - for rule in &self.rules { - let valid_checkpoint_cadence = match (rule.mode, rule.full_checkpoint_every_ms) { - (TransmissionMode::Full, None) => true, - (TransmissionMode::Delta, Some(full_every)) => { - rule.emit_every_ms > 0 - && full_every >= rule.emit_every_ms - && full_every % rule.emit_every_ms == 0 - } - _ => false, - }; - if rule.emit_every_ms == 0 - || rule.destination_ref.is_empty() - || !valid_checkpoint_cadence - { - return Err(TransmissionPlanError::InvalidRule(rule.producer_id.clone())); - } - let schema = precompute - .schemas - .iter() - .find(|schema| schema.materialization == rule.materialization) - .expect("producer set validation guarantees a matching schema"); - if !schema.encodings.contains(&rule.encoding) { - return Err(TransmissionPlanError::InvalidRule(rule.producer_id.clone())); - } - validate_runtime_rule_policy(rule, &schema.family)?; - } - Ok(()) - } - - /// Authorize a telemetry-driven successor without mutating this active - /// plan. Semantic identity, codecs, destination, transmission mode and - /// checkpoint cadence remain fixed. Only explicitly bounded runtime knobs - /// may move, and each changed rule needs fresh evidence attributed to the - /// exact active generation. - pub fn authorize_successor( - &self, - successor: &TransmissionPlan, - evidence: &[RuntimeAdaptationEvidence], - now_unix_ms: u64, - ) -> Result<(), TransmissionPlanError> { - if successor.envelope.plan_id != self.envelope.plan_id - || successor.envelope.plan_version != self.envelope.plan_version.saturating_add(1) - || successor.envelope.backend_compat != self.envelope.backend_compat - || successor.envelope.planner_revision != self.envelope.planner_revision - || successor.envelope.capability_snapshot_id != self.envelope.capability_snapshot_id - || successor.envelope.generated_at_unix_ms < self.envelope.generated_at_unix_ms - || successor.envelope.activation_unix_ms < successor.envelope.generated_at_unix_ms - || successor.rules.len() != self.rules.len() - || successor.frame_identity != self.frame_identity - { - return Err(TransmissionPlanError::InvalidSuccessor( - "successor must be the next version of the same semantic/capability generation" - .into(), - )); - } - - for current in &self.rules { - let Some(next) = successor.rules.iter().find(|candidate| { - candidate.materialization == current.materialization - && candidate.producer_id == current.producer_id - && candidate.schema_id == current.schema_id - }) else { - return Err(TransmissionPlanError::InvalidSuccessor(format!( - "missing rule for producer {}", - current.producer_id - ))); - }; - if current.mode != next.mode - || current.encoding != next.encoding - || current.full_checkpoint_every_ms != next.full_checkpoint_every_ms - || current.destination_ref != next.destination_ref - || current.runtime_policy.adaptation != next.runtime_policy.adaptation - || sampling_estimator(¤t.runtime_policy.sampling) - != sampling_estimator(&next.runtime_policy.sampling) - || delta_shape(¤t.runtime_policy.delta) - != delta_shape(&next.runtime_policy.delta) - { - return Err(TransmissionPlanError::InvalidSuccessor(format!( - "rule identity/codec/mode/guardrails drifted for producer {}", - current.producer_id - ))); - } - if current.emit_every_ms == next.emit_every_ms - && current.runtime_policy.sampling == next.runtime_policy.sampling - && current.runtime_policy.delta == next.runtime_policy.delta - { - continue; - } - let policy = ¤t.runtime_policy.adaptation; - if !policy.enabled || now_unix_ms < policy.not_before_unix_ms { - return Err(TransmissionPlanError::AdaptationOutOfBounds { - producer_id: current.producer_id.clone(), - knob: "adaptation_disabled_or_in_cooldown", - }); - } - let has_evidence = evidence.iter().any(|item| { - item.plan_id == self.envelope.plan_id - && item.plan_version == self.envelope.plan_version - && item.materialization == current.materialization - && item.producer_id == current.producer_id - && item.schema_id == current.schema_id - && !item.producer_version.trim().is_empty() - && item.sample_count >= policy.min_evidence_samples - && item.observed_at_unix_ms <= now_unix_ms - && now_unix_ms.saturating_sub(item.observed_at_unix_ms) - <= policy.max_evidence_age_ms - }); - if !has_evidence { - return Err(TransmissionPlanError::InvalidAdaptationEvidence( - current.producer_id.clone(), - )); - } - authorize_f64_change( - sampling_probability(¤t.runtime_policy.sampling), - sampling_probability(&next.runtime_policy.sampling), - policy.sample_probability.as_ref(), - ¤t.producer_id, - "sample_probability", - )?; - authorize_u64_change( - current.emit_every_ms, - next.emit_every_ms, - policy.emit_every_ms.as_ref(), - ¤t.producer_id, - "emit_every_ms", - )?; - authorize_f64_change( - delta_threshold(¤t.runtime_policy.delta), - delta_threshold(&next.runtime_policy.delta), - policy.delta_threshold.as_ref(), - ¤t.producer_id, - "delta_threshold", - )?; - authorize_f64_change( - gos_epsilon(¤t.runtime_policy.delta), - gos_epsilon(&next.runtime_policy.delta), - policy.gos_epsilon_staleness.as_ref(), - ¤t.producer_id, - "gos_epsilon_staleness", - )?; - } - Ok(()) - } - - pub fn validate_frame( - &self, - frame: &SummaryFrameIdentity, - ) -> Result<(), TransmissionPlanError> { - if frame.identity_version != self.frame_identity.identity_version - || frame.plan_id != self.envelope.plan_id - || frame.plan_version != self.envelope.plan_version - || frame.backend_compat != self.envelope.backend_compat - || frame.series_identity.is_empty() - || frame.producer_epoch.is_empty() - || frame.sequence == 0 - || frame.window_start_unix_nano >= frame.window_end_unix_nano - || (frame.kind == SummaryFrameKind::Full - && self.frame_identity.require_checkpoint_for_full - && frame.checkpoint_id.is_none()) - || (frame.kind == SummaryFrameKind::Delta - && self.frame_identity.require_base_checkpoint_for_delta - && frame.base_checkpoint_id.is_none()) - { - return Err(TransmissionPlanError::InvalidFrame( - "identity/lifecycle/window/checkpoint fields do not satisfy the active contract" - .into(), - )); - } - if self.rules.iter().any(|rule| { - rule.materialization == frame.materialization - && rule.producer_id == frame.producer_id - && rule.schema_id == frame.schema_id - // A delta rule necessarily emits periodic full checkpoints; - // a full-only rule must never emit deltas. - && (frame.kind == SummaryFrameKind::Full - || rule.mode == TransmissionMode::Delta) - && rule.encoding == frame.encoding - }) { - Ok(()) - } else { - Err(TransmissionPlanError::UnknownFrame) - } - } -} - -fn validate_runtime_rule_policy( - rule: &TransmissionRule, - family: &StateFamilyContract, -) -> Result<(), TransmissionPlanError> { - let invalid = |reason: &str| TransmissionPlanError::InvalidRuntimePolicy { - producer_id: rule.producer_id.clone(), - reason: reason.into(), - }; - if let SamplingPolicy::Fixed { - probability, - estimator, - } = &rule.runtime_policy.sampling - { - if !probability.is_finite() || !(0.0..=1.0).contains(probability) || *probability == 0.0 { - return Err(invalid("sample probability must be finite and in (0, 1]")); - } - let supported = matches!( - (family, *estimator), - ( - StateFamilyContract::Sketch { - algorithm: SketchAlgorithm::Hll, - .. - }, - SamplingEstimator::HashThreshold, - ) | ( - StateFamilyContract::Sketch { - algorithm: SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap, - .. - }, - SamplingEstimator::GeometricAdmission, - ) - ); - if !supported { - return Err(invalid( - "sampling estimator is not implemented for the materialization family", - )); - } - } - - if (rule.mode == TransmissionMode::Delta) != rule.runtime_policy.delta.is_some() { - return Err(invalid( - "delta policy must be present exactly when transmission mode is delta", - )); - } - if let Some(delta) = &rule.runtime_policy.delta { - if !delta.absolute_threshold.is_finite() || delta.absolute_threshold < 0.0 { - return Err(invalid("delta threshold must be finite and non-negative")); - } - if !matches!( - family, - StateFamilyContract::Sketch { - algorithm: SketchAlgorithm::DDSketch - | SketchAlgorithm::Hll - | SketchAlgorithm::Cms - | SketchAlgorithm::CmsWithHeap - | SketchAlgorithm::CountSketch - | SketchAlgorithm::CountSketchWithHeap, - .. - } - ) { - return Err(invalid( - "delta transmission is not implemented for the materialization family", - )); - } - if let Some(gos) = &delta.gos { - if !matches!( - family, - StateFamilyContract::Sketch { - algorithm: SketchAlgorithm::CountSketch | SketchAlgorithm::CountSketchWithHeap, - .. - } - ) || !gos.epsilon_staleness.is_finite() - || !(0.0..=1.0).contains(&gos.epsilon_staleness) - || gos.epsilon_staleness == 0.0 - || gos.sites == 0 - { - return Err(invalid( - "GOS requires a CountSketch family, epsilon in (0, 1], and at least one site", - )); - } - } - } - - let adaptation = &rule.runtime_policy.adaptation; - if adaptation.enabled - && (adaptation.max_evidence_age_ms == 0 || adaptation.min_evidence_samples == 0) - { - return Err(invalid( - "enabled adaptation requires non-zero evidence age and sample-count requirements", - )); - } - validate_f64_bounds(adaptation.sample_probability.as_ref(), 0.0, 1.0) - .map_err(|reason| invalid(reason))?; - validate_f64_bounds(adaptation.delta_threshold.as_ref(), 0.0, f64::MAX) - .map_err(|reason| invalid(reason))?; - validate_f64_bounds(adaptation.gos_epsilon_staleness.as_ref(), 0.0, 1.0) - .map_err(|reason| invalid(reason))?; - if let Some(bounds) = &adaptation.emit_every_ms { - if bounds.min == 0 - || bounds.min > bounds.max - || bounds.max_step == 0 - || !(bounds.min..=bounds.max).contains(&rule.emit_every_ms) - { - return Err(invalid("emit interval guardrails are invalid")); - } - } - if let Some(bounds) = &adaptation.sample_probability { - let current = sampling_probability(&rule.runtime_policy.sampling); - if current < bounds.min || current > bounds.max { - return Err(invalid( - "current sampling probability is outside guardrails", - )); - } - } - if let Some(bounds) = &adaptation.delta_threshold { - let Some(current) = rule - .runtime_policy - .delta - .as_ref() - .map(|policy| policy.absolute_threshold) - else { - return Err(invalid("delta guardrails require an active delta policy")); - }; - if current < bounds.min || current > bounds.max { - return Err(invalid("current delta threshold is outside guardrails")); - } - } - if let Some(bounds) = &adaptation.gos_epsilon_staleness { - let Some(current) = rule - .runtime_policy - .delta - .as_ref() - .and_then(|policy| policy.gos.as_ref()) - .map(|gos| gos.epsilon_staleness) - else { - return Err(invalid("GOS guardrails require an active GOS policy")); - }; - if current < bounds.min || current > bounds.max { - return Err(invalid("current GOS epsilon is outside guardrails")); - } - } - Ok(()) -} - -fn validate_f64_bounds( - bounds: Option<&AdaptiveF64Bounds>, - domain_min: f64, - domain_max: f64, -) -> Result<(), &'static str> { - let Some(bounds) = bounds else { - return Ok(()); - }; - if !bounds.min.is_finite() - || !bounds.max.is_finite() - || !bounds.max_step.is_finite() - || bounds.min < domain_min - || bounds.max > domain_max - || bounds.min > bounds.max - || bounds.max_step <= 0.0 - { - Err("floating-point adaptation guardrails are invalid") - } else { - Ok(()) - } -} - -fn sampling_probability(policy: &SamplingPolicy) -> f64 { - match policy { - SamplingPolicy::Disabled => 1.0, - SamplingPolicy::Fixed { probability, .. } => *probability, - } -} - -fn sampling_estimator(policy: &SamplingPolicy) -> Option { - match policy { - SamplingPolicy::Disabled => None, - SamplingPolicy::Fixed { estimator, .. } => Some(*estimator), - } -} - -fn delta_threshold(policy: &Option) -> f64 { - policy - .as_ref() - .map(|policy| policy.absolute_threshold) - .unwrap_or(0.0) -} - -fn gos_epsilon(policy: &Option) -> f64 { - policy - .as_ref() - .and_then(|policy| policy.gos.as_ref()) - .map(|gos| gos.epsilon_staleness) - .unwrap_or(0.0) -} - -fn delta_shape(policy: &Option) -> Option<(Option<(u32, GosThresholdMode)>,)> { - policy.as_ref().map(|policy| { - (policy - .gos - .as_ref() - .map(|gos| (gos.sites, gos.threshold_mode)),) - }) -} - -fn authorize_f64_change( - current: f64, - next: f64, - bounds: Option<&AdaptiveF64Bounds>, - producer_id: &str, - knob: &'static str, -) -> Result<(), TransmissionPlanError> { - if current == next { - return Ok(()); - } - let allowed = bounds.is_some_and(|bounds| { - next.is_finite() - && (bounds.min..=bounds.max).contains(&next) - && (next - current).abs() <= bounds.max_step - }); - if allowed { - Ok(()) - } else { - Err(TransmissionPlanError::AdaptationOutOfBounds { - producer_id: producer_id.into(), - knob, - }) - } -} - -fn authorize_u64_change( - current: u64, - next: u64, - bounds: Option<&AdaptiveU64Bounds>, - producer_id: &str, - knob: &'static str, -) -> Result<(), TransmissionPlanError> { - if current == next { - return Ok(()); - } - let allowed = bounds.is_some_and(|bounds| { - (bounds.min..=bounds.max).contains(&next) && current.abs_diff(next) <= bounds.max_step - }); - if allowed { - Ok(()) - } else { - Err(TransmissionPlanError::AdaptationOutOfBounds { - producer_id: producer_id.into(), - knob, - }) - } -} - #[derive(Debug, Error)] pub enum CompileError { #[error("invalid backend-local workload snapshot: {0}")] @@ -2089,12 +1279,15 @@ impl PhysicalCompiler { query_id: "precompute-plan".into(), reason: error.to_string(), })?; - let mut transmission_plan = - TransmissionPlan::build(envelope.clone(), &precompute_plan, &runtime_policies) - .map_err(|error| CompileError::Query { - query_id: "transmission-plan".into(), - reason: error.to_string(), - })?; + let mut transmission_plan = crate::physical::compiler::compile_transmission_plan( + envelope.clone(), + &precompute_plan, + &runtime_policies, + ) + .map_err(|error| CompileError::Query { + query_id: "transmission-plan".into(), + reason: error.to_string(), + })?; let mut collector_plans = producer_ids .into_iter() .map(|collector_id| CollectorPlan { @@ -5222,13 +4415,21 @@ mod tests { collector.validate_against_catalog(catalog).unwrap(); collector.envelope.plan_version += 1; assert!(collector.validate_against_catalog(catalog).is_err()); - assert!(validate_catalog_projection( - Some(&catalog.reference().unwrap()), - &bundle.envelope, - [asap_types::PolicyFingerprint(u64::MAX).into()], - catalog - ) - .is_err()); + let mut unknown_materialization = bundle.transmission_plan.clone(); + unknown_materialization.rules.push(TransmissionRule { + materialization: asap_types::PolicyFingerprint(u64::MAX).into(), + producer_id: "foreign".into(), + schema_id: "foreign".into(), + mode: TransmissionMode::Full, + encoding: StateEncoding::ExactAccumulatorV1, + emit_every_ms: 60_000, + full_checkpoint_every_ms: None, + destination_ref: "backend".into(), + runtime_policy: Default::default(), + }); + assert!(unknown_materialization + .validate_against_catalog(catalog) + .is_err()); let encoded = serde_json::to_value(&collector).unwrap(); assert!(encoded["summary_catalog"] .get("summary_descriptors") @@ -5542,10 +4743,14 @@ mod tests { assert_eq!(plan.ingest.timestamp_unit, TimestampUnit::UnixMilliseconds); assert!(plan.producers.is_empty()); plan.validate().expect("valid backend-local projection"); - TransmissionPlan::build(bundle.envelope, &plan, &BTreeMap::new()) - .expect("empty backend-local transmission contract") - .validate(&plan) - .expect("valid empty transmission plan"); + crate::physical::compiler::compile_transmission_plan( + bundle.envelope, + &plan, + &BTreeMap::new(), + ) + .expect("empty backend-local transmission contract") + .validate(&plan) + .expect("valid empty transmission plan"); } #[test] @@ -6005,6 +5210,9 @@ mod tests { ), full_selected ); + // Publication validates stored extent independently of slide cadence. + // In particular the full-window candidate stores 60s at a 10s slide. + bundle.publication().unwrap().validate().unwrap(); assert_eq!(bundle.precompute_plan.materializations[0].window_size, 60); assert_eq!( bundle.precompute_plan.materializations[0].slide_interval, @@ -6213,7 +5421,7 @@ mod tests { #[test] fn runtime_policy_uses_canonical_sampling_and_gos_allocators() { - let sampling = SamplingPolicy::from_accuracy_budget( + let sampling = sampling_policy_from_accuracy_budget( 0.05, 1_000.0, SamplingEstimator::GeometricAdmission, @@ -6224,64 +5432,26 @@ mod tests { assert!((probability - 1.0 / 3.5).abs() < 1e-9); let gos = - GosPolicy::from_accuracy_budget(0.1, 0.03, 4, 1.0, 1.0, GosThresholdMode::Isotropic) + gos_policy_from_accuracy_budget(0.1, 0.03, 4, 1.0, 1.0, GosThresholdMode::Isotropic) .expect("positive communication allocation"); assert!((gos.epsilon_staleness - 0.035).abs() < 1e-12); assert_eq!(gos.sites, 4); } #[test] - fn runtime_policy_is_family_and_mode_checked() { + fn runtime_policy_encoding_is_checked() { let bundle = PhysicalCompiler .compile( request("q", "quantile_over_time(0.99, m[1m])"), environment(10_000), ) .expect("compile"); - let mut rule = bundle.transmission_plan.rules[0].clone(); let mut invalid_encoding = bundle.transmission_plan.clone(); invalid_encoding.rules[0].encoding = StateEncoding::ExactAccumulatorV1; assert!(matches!( invalid_encoding.validate(&bundle.precompute_plan), Err(TransmissionPlanError::InvalidRule(_)) )); - rule.runtime_policy.sampling = SamplingPolicy::Fixed { - probability: 0.5, - estimator: SamplingEstimator::HashThreshold, - }; - let hll = StateFamilyContract::Sketch { - algorithm: SketchAlgorithm::Hll, - parameters: SketchParams::Hll { precision: 14 }, - }; - let count_sketch = StateFamilyContract::Sketch { - algorithm: SketchAlgorithm::CountSketch, - parameters: SketchParams::CountSketch { - width: 128, - depth: 4, - }, - }; - validate_runtime_rule_policy(&rule, &hll).expect("HLL supports hash-threshold sampling"); - assert!(matches!( - validate_runtime_rule_policy(&rule, &count_sketch), - Err(TransmissionPlanError::InvalidRuntimePolicy { .. }) - )); - - rule.runtime_policy.sampling = SamplingPolicy::Disabled; - rule.mode = TransmissionMode::Delta; - rule.full_checkpoint_every_ms = Some(300_000); - rule.runtime_policy.delta = Some(DeltaPolicy { - absolute_threshold: 0.0, - gos: Some(GosPolicy { - epsilon_staleness: 0.02, - sites: 2, - threshold_mode: GosThresholdMode::Isotropic, - }), - }); - validate_runtime_rule_policy(&rule, &count_sketch).expect("CountSketch supports delta GOS"); - assert!(matches!( - validate_runtime_rule_policy(&rule, &hll), - Err(TransmissionPlanError::InvalidRuntimePolicy { .. }) - )); } #[test] diff --git a/control_plane/src/physical/publication.rs b/control_plane/src/physical/publication.rs index 42e28d086..63081fa5b 100644 --- a/control_plane/src/physical/publication.rs +++ b/control_plane/src/physical/publication.rs @@ -1,128 +1,7 @@ -//! Canonical publication document for one catalog generation. -use super::compiler::{CollectorPlan, PhysicalPlan, PrecomputePlan, TransmissionPlan}; -use super::summary_catalog::SummaryCatalog; -use crate::query_plan::QueryPlan; -use serde::{Deserialize, Serialize}; +//! Control-plane construction of the shared catalog publication contract. +use super::compiler::PhysicalPlan; +pub use asap_types::plan_publication::{PhysicalPlanInstallRequest, PhysicalPlanPublication}; -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct PhysicalPlanPublication { - pub summary_catalog: SummaryCatalog, - pub precompute_plan: PrecomputePlan, - pub collector_plans: Vec, - pub transmission_plan: TransmissionPlan, - pub query_plan: QueryPlan, -} - -/// Complete typed envelope accepted by the data-plane install endpoint. -/// -/// Runtime-only routing and adaptation evidence decorate the compiler-owned -/// publication without changing its catalog generation. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct PhysicalPlanInstallRequest { - pub summary_catalog: SummaryCatalog, - #[serde(default)] - pub collector_plans: Vec, - pub precompute_plan: PrecomputePlan, - pub transmission_plan: TransmissionPlan, - pub query_plan: QueryPlan, - pub storage_routing: Option, - #[serde(default)] - pub adaptation_evidence: Vec, -} - -impl PhysicalPlanPublication { - /// Validate every plan against the shared catalog snapshot. - pub fn validate(&self) -> Result<(), String> { - let catalog = &self.summary_catalog; - self.precompute_plan - .validate_against_catalog(catalog) - .map_err(|e| e.to_string())?; - self.transmission_plan - .validate(&self.precompute_plan) - .map_err(|e| e.to_string())?; - self.transmission_plan - .validate_against_catalog(catalog) - .map_err(|e| e.to_string())?; - self.query_plan - .validate_against_catalog(catalog) - .map_err(|e| e.to_string())?; - let materializations = self - .precompute_plan - .materializations - .iter() - .map(|config| (config.policy_fingerprint(), config)) - .collect::>(); - for entry in self.query_plan.entries.values() { - for binding in entry.materialization_bindings() { - let config = materializations - .get(&binding.materialization.fingerprint()) - .copied() - .ok_or("query binding has no precompute materialization")?; - if config.slide_interval.checked_mul(1000) != Some(binding.window_ms) { - return Err("query pane differs from precompute emission interval".into()); - } - if config.pane_origin_ms != binding.pane_origin_ms { - return Err("query pane origin differs from precompute definition".into()); - } - } - } - let mut collectors = std::collections::BTreeSet::new(); - for collector in &self.collector_plans { - if collector.envelope != self.precompute_plan.envelope - || !collectors.insert(&collector.collector_id) - { - return Err("collector envelope mismatch or duplicate collector".into()); - } - collector - .validate_against_catalog(catalog) - .map_err(|e| e.to_string())?; - for rule in &collector.transmission_rules { - if !self.transmission_plan.rules.contains(rule) { - return Err( - "collector transmission rule absent from published transmission plan" - .into(), - ); - } - } - } - for producer in &self.precompute_plan.producers { - let collector = self - .collector_plans - .iter() - .find(|c| c.collector_id == producer.collector_id) - .ok_or("precompute producer has no published CollectorPlan")?; - if !collector - .materializations - .iter() - .any(|m| m.materialization == producer.materialization) - { - return Err( - "collector does not produce referenced precompute materialization".into(), - ); - } - } - Ok(()) - } - - pub fn install_request( - &self, - storage_routing: Option, - adaptation_evidence: Vec, - ) -> Result { - self.validate()?; - Ok(PhysicalPlanInstallRequest { - summary_catalog: self.summary_catalog.clone(), - collector_plans: self.collector_plans.clone(), - precompute_plan: self.precompute_plan.clone(), - transmission_plan: self.transmission_plan.clone(), - query_plan: self.query_plan.clone(), - storage_routing, - adaptation_evidence, - }) - } -} impl PhysicalPlan { pub fn publication(&self) -> Result { let artifact = PhysicalPlanPublication { diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 0b4f40eee..7f4673837 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -7,8 +7,10 @@ pub mod executable_plan; pub mod grouping_projection; pub mod key_by_label_names; pub mod monitor_spec; +pub mod plan_publication; pub mod policy_fingerprint; pub mod policy_registry; +pub mod producer_plan; pub mod query_requirements; pub mod routing_index; pub mod sds; diff --git a/crates/asap_types/src/plan_publication.rs b/crates/asap_types/src/plan_publication.rs new file mode 100644 index 000000000..35ce812f5 --- /dev/null +++ b/crates/asap_types/src/plan_publication.rs @@ -0,0 +1,126 @@ +//! Canonical publication document for one catalog generation. +use crate::precompute_plan::PrecomputePlan; +use crate::producer_plan::{CollectorPlan, TransmissionPlan}; +use crate::query_plan::QueryPlan; +use crate::summary_catalog::SummaryCatalog; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PhysicalPlanPublication { + pub summary_catalog: SummaryCatalog, + pub precompute_plan: PrecomputePlan, + pub collector_plans: Vec, + pub transmission_plan: TransmissionPlan, + pub query_plan: QueryPlan, +} + +/// Complete typed envelope accepted by the data-plane install endpoint. +/// +/// Runtime-only routing and adaptation evidence decorate the shared +/// publication without changing its catalog generation. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PhysicalPlanInstallRequest { + pub summary_catalog: SummaryCatalog, + #[serde(default)] + pub collector_plans: Vec, + pub precompute_plan: PrecomputePlan, + pub transmission_plan: TransmissionPlan, + pub query_plan: QueryPlan, + pub storage_routing: Option, + #[serde(default)] + pub adaptation_evidence: Vec, +} + +impl PhysicalPlanPublication { + /// Validate every plan against the shared catalog snapshot. + pub fn validate(&self) -> Result<(), String> { + let catalog = &self.summary_catalog; + self.precompute_plan + .validate_against_catalog(catalog) + .map_err(|e| e.to_string())?; + self.transmission_plan + .validate(&self.precompute_plan) + .map_err(|e| e.to_string())?; + self.transmission_plan + .validate_against_catalog(catalog) + .map_err(|e| e.to_string())?; + self.query_plan + .validate_against_catalog(catalog) + .map_err(|e| e.to_string())?; + let materializations = self + .precompute_plan + .materializations + .iter() + .map(|config| (config.policy_fingerprint(), config)) + .collect::>(); + for entry in self.query_plan.entries.values() { + for binding in entry.materialization_bindings() { + let config = materializations + .get(&binding.materialization.fingerprint()) + .copied() + .ok_or("query binding has no precompute materialization")?; + if config.stored_window_ms() != binding.window_ms { + return Err("query pane differs from precompute stored window".into()); + } + if config.pane_origin_ms != binding.pane_origin_ms { + return Err("query pane origin differs from precompute definition".into()); + } + } + } + let mut collectors = std::collections::BTreeSet::new(); + for collector in &self.collector_plans { + if collector.envelope != self.precompute_plan.envelope + || !collectors.insert(&collector.collector_id) + { + return Err("collector envelope mismatch or duplicate collector".into()); + } + collector + .validate_against_catalog(catalog) + .map_err(|e| e.to_string())?; + for rule in &collector.transmission_rules { + if !self.transmission_plan.rules.contains(rule) { + return Err( + "collector transmission rule absent from published transmission plan" + .into(), + ); + } + } + } + for producer in &self.precompute_plan.producers { + let collector = self + .collector_plans + .iter() + .find(|c| c.collector_id == producer.collector_id) + .ok_or("precompute producer has no published CollectorPlan")?; + if !collector + .materializations + .iter() + .any(|m| m.materialization == producer.materialization) + { + return Err( + "collector does not produce referenced precompute materialization".into(), + ); + } + } + Ok(()) + } + + pub fn install_request( + &self, + storage_routing: Option, + adaptation_evidence: Vec, + ) -> Result { + self.validate()?; + Ok(PhysicalPlanInstallRequest { + summary_catalog: self.summary_catalog.clone(), + collector_plans: self.collector_plans.clone(), + precompute_plan: self.precompute_plan.clone(), + transmission_plan: self.transmission_plan.clone(), + query_plan: self.query_plan.clone(), + storage_routing, + adaptation_evidence, + }) + } +} diff --git a/crates/asap_types/src/producer_plan.rs b/crates/asap_types/src/producer_plan.rs new file mode 100644 index 000000000..81d4b36e6 --- /dev/null +++ b/crates/asap_types/src/producer_plan.rs @@ -0,0 +1,879 @@ +//! Installed collector and transmission contracts shared across producers and consumers. +//! Deployment choices and accuracy-budget allocation remain in the control plane. +use crate::precompute_plan::{PlanEnvelope, PrecomputePlan, StateEncoding, StateFamilyContract}; +use planner_types::post_asap::{SketchAlgorithm, SummaryWindowFramework}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::BTreeSet; +use thiserror::Error; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct CollectorMaterialization { + pub query_id: String, + pub materialization: crate::sds::SummaryDefinitionId, + pub metric: String, + pub algorithm: String, + pub parameters: Value, + pub group_by: Vec, + pub window_secs: u64, + pub abstract_window_framework: SummaryWindowFramework, + pub window_implementation_id: String, + pub slide_secs: u64, + #[serde( + default, + alias = "paneOriginMs", + skip_serializing_if = "Option::is_none" + )] + pub pane_origin_ms: Option, + pub window_layout: crate::WindowMaterializationLayout, + pub evidence_source: Option, + pub lifecycle: CollectorLifecycle, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct CollectorLifecycle { + pub kind: String, + pub maintenance_mode: String, + pub evaluation_schedule: String, + pub output_representation: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct CollectorPlan { + /// Absent only in legacy artifacts; catalog-aware validation requires it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub summary_catalog: Option, + pub collector_id: String, + pub envelope: PlanEnvelope, + pub materializations: Vec, + pub transmission_rules: Vec, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(rename_all = "snake_case")] +pub enum TransmissionMode { + Full, + Delta, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SequenceScope { + MaterializationSeriesProducerEpoch, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct FrameIdentityContract { + pub identity_version: u32, + pub sequence_scope: SequenceScope, + pub require_checkpoint_for_full: bool, + pub require_base_checkpoint_for_delta: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct TransmissionRule { + pub materialization: crate::sds::SummaryDefinitionId, + pub producer_id: String, + pub schema_id: String, + pub mode: TransmissionMode, + pub encoding: StateEncoding, + pub emit_every_ms: u64, + pub full_checkpoint_every_ms: Option, + pub destination_ref: String, + /// Plan-owned runtime knobs. These values are part of the immutable plan + /// generation; live feedback may only change them by publishing a + /// successor generation accepted by [`TransmissionPlan::authorize_successor`]. + #[serde(default)] + pub runtime_policy: RuntimeRulePolicy, +} + +/// How the collector admits updates before sketch maintenance. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "mode", rename_all = "snake_case", deny_unknown_fields)] +pub enum SamplingPolicy { + Disabled, + Fixed { + /// Probability in `(0, 1]`; `1` is valid but should normally be + /// represented by `Disabled`. + probability: f64, + estimator: SamplingEstimator, + }, +} + +impl Default for SamplingPolicy { + fn default() -> Self { + Self::Disabled + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SamplingEstimator { + /// Hash-threshold element sampling, used by cardinality summaries. + HashThreshold, + /// Geometric admission/Nitro-style update sampling, used by frequency + /// summaries. The sketch readout carries the corresponding correction. + GeometricAdmission, +} + +/// Norm-adaptive Group-of-Sketches delta gating. GOS is meaningful only for +/// CountSketch families and only when the transmission rule is in delta mode. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct GosPolicy { + pub epsilon_staleness: f64, + pub sites: u32, + pub threshold_mode: GosThresholdMode, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum GosThresholdMode { + Isotropic, + Anisotropic, +} + +/// Sparse-delta semantics within a delta transmission rule. An absolute +/// threshold of zero sends every changed cell. When `gos` is present it +/// replaces the fixed threshold with the GOS norm-adaptive threshold. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct DeltaPolicy { + pub absolute_threshold: f64, + pub gos: Option, +} + +/// Inclusive bounds for one floating-point adaptation knob. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct AdaptiveF64Bounds { + pub min: f64, + pub max: f64, + pub max_step: f64, +} + +/// Inclusive bounds for one integer adaptation knob. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct AdaptiveU64Bounds { + pub min: u64, + pub max: u64, + pub max_step: u64, +} + +/// Guardrails for telemetry-driven runtime adaptation. This is an +/// authorization contract, not an instruction to mutate the active plan. +/// Every accepted change becomes a staged successor PhysicalPlan. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct RuntimeAdaptationPolicy { + pub enabled: bool, + pub not_before_unix_ms: u64, + pub max_evidence_age_ms: u64, + pub min_evidence_samples: u64, + pub sample_probability: Option, + pub emit_every_ms: Option, + pub delta_threshold: Option, + pub gos_epsilon_staleness: Option, +} + +impl Default for RuntimeAdaptationPolicy { + fn default() -> Self { + Self { + enabled: false, + not_before_unix_ms: 0, + max_evidence_age_ms: 0, + min_evidence_samples: 0, + sample_probability: None, + emit_every_ms: None, + delta_threshold: None, + gos_epsilon_staleness: None, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +#[serde(deny_unknown_fields)] +pub struct RuntimeRulePolicy { + #[serde(default)] + pub sampling: SamplingPolicy, + pub delta: Option, + #[serde(default)] + pub adaptation: RuntimeAdaptationPolicy, +} + +/// Identity and sufficiency information for evidence authorizing one rule's +/// successor knobs. Raw measurements remain in the runtime-samples store; the +/// authorization boundary needs only their exact provenance and sample count. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct RuntimeAdaptationEvidence { + pub plan_id: u64, + pub plan_version: u64, + pub materialization: crate::sds::SummaryDefinitionId, + pub producer_id: String, + pub schema_id: String, + pub producer_version: String, + pub observed_at_unix_ms: u64, + pub sample_count: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct TransmissionPlan { + /// Absent only in legacy artifacts; catalog-aware validation requires it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub summary_catalog: Option, + pub envelope: PlanEnvelope, + pub frame_identity: FrameIdentityContract, + pub rules: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum SummaryFrameKind { + Full, + Delta, +} + +/// Identity attached to every summary record. Window bounds come from the +/// data point; the remaining fields are carried as reserved `asap.frame.*` +/// attributes until the modified-OTLP schema gains a dedicated message. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct SummaryFrameIdentity { + pub identity_version: u32, + pub plan_id: u64, + pub plan_version: u64, + pub backend_compat: String, + pub materialization: crate::sds::SummaryDefinitionId, + /// Canonical producer-side identity for one concrete retained-label group. + pub series_identity: String, + pub schema_id: String, + pub producer_id: String, + pub producer_epoch: String, + pub window_start_unix_nano: u64, + pub window_end_unix_nano: u64, + pub sequence: u64, + pub kind: SummaryFrameKind, + pub encoding: StateEncoding, + pub checkpoint_id: Option, + pub base_checkpoint_id: Option, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum TransmissionPlanError { + #[error("summary catalog mismatch: {0}")] + Catalog(String), + #[error("TransmissionPlan envelope differs from PrecomputePlan")] + EnvelopeMismatch, + #[error("transmission rules do not exactly match precompute producer bindings")] + ProducerSetMismatch, + #[error("invalid transmission rule for producer {0}")] + InvalidRule(String), + #[error("frame identity is invalid: {0}")] + InvalidFrame(String), + #[error("frame has no matching transmission rule")] + UnknownFrame, + #[error("runtime policy for producer {producer_id} is invalid: {reason}")] + InvalidRuntimePolicy { producer_id: String, reason: String }, + #[error("runtime adaptation successor is invalid: {0}")] + InvalidSuccessor(String), + #[error("runtime adaptation evidence for producer {0} is missing or invalid")] + InvalidAdaptationEvidence(String), + #[error("runtime adaptation for producer {producer_id} exceeds guardrails: {knob}")] + AdaptationOutOfBounds { + producer_id: String, + knob: &'static str, + }, +} + +fn validate_catalog_projection( + reference: Option<&crate::sds::CatalogGeneration>, + envelope: &PlanEnvelope, + materializations: impl IntoIterator, + catalog: &crate::summary_catalog::SummaryCatalog, +) -> Result<(), TransmissionPlanError> { + let expected = catalog + .reference() + .map_err(|error| TransmissionPlanError::Catalog(error.to_string()))?; + if reference != Some(&expected) + || envelope.plan_id != catalog.plan_id + || envelope.plan_version != catalog.plan_version + { + return Err(TransmissionPlanError::Catalog( + "missing or different snapshot reference".into(), + )); + } + for id in materializations { + if !catalog.materializations.contains_key(&id) { + return Err(TransmissionPlanError::Catalog(format!( + "unknown materialization {}", + id.as_u64() + ))); + } + } + Ok(()) +} + +impl CollectorPlan { + pub fn validate_against_catalog( + &self, + catalog: &crate::summary_catalog::SummaryCatalog, + ) -> Result<(), TransmissionPlanError> { + validate_catalog_projection( + self.summary_catalog.as_ref(), + &self.envelope, + self.materializations + .iter() + .map(|m| m.materialization) + .chain(self.transmission_rules.iter().map(|r| r.materialization)), + catalog, + ) + } +} + +impl TransmissionPlan { + pub fn validate_against_catalog( + &self, + catalog: &crate::summary_catalog::SummaryCatalog, + ) -> Result<(), TransmissionPlanError> { + validate_catalog_projection( + self.summary_catalog.as_ref(), + &self.envelope, + self.rules.iter().map(|r| r.materialization), + catalog, + ) + } + + pub fn validate(&self, precompute: &PrecomputePlan) -> Result<(), TransmissionPlanError> { + if self.summary_catalog != precompute.summary_catalog { + return Err(TransmissionPlanError::Catalog( + "transmission and precompute plans reference different catalog snapshots".into(), + )); + } + if self.envelope != precompute.envelope { + return Err(TransmissionPlanError::EnvelopeMismatch); + } + let expected: BTreeSet<_> = precompute + .producers + .iter() + .map(|producer| { + ( + producer.materialization, + producer.producer_id.as_str(), + producer.schema_id.as_str(), + ) + }) + .collect(); + let actual: BTreeSet<_> = self + .rules + .iter() + .map(|rule| { + ( + rule.materialization, + rule.producer_id.as_str(), + rule.schema_id.as_str(), + ) + }) + .collect(); + if expected != actual || actual.len() != self.rules.len() { + return Err(TransmissionPlanError::ProducerSetMismatch); + } + for rule in &self.rules { + let valid_checkpoint_cadence = match (rule.mode, rule.full_checkpoint_every_ms) { + (TransmissionMode::Full, None) => true, + (TransmissionMode::Delta, Some(full_every)) => { + rule.emit_every_ms > 0 + && full_every >= rule.emit_every_ms + && full_every % rule.emit_every_ms == 0 + } + _ => false, + }; + if rule.emit_every_ms == 0 + || rule.destination_ref.is_empty() + || !valid_checkpoint_cadence + { + return Err(TransmissionPlanError::InvalidRule(rule.producer_id.clone())); + } + let schema = precompute + .schemas + .iter() + .find(|schema| schema.materialization == rule.materialization) + .expect("producer set validation guarantees a matching schema"); + if !schema.encodings.contains(&rule.encoding) { + return Err(TransmissionPlanError::InvalidRule(rule.producer_id.clone())); + } + validate_runtime_rule_policy(rule, &schema.family)?; + } + Ok(()) + } + + /// Authorize a telemetry-driven successor without mutating this active + /// plan. Semantic identity, codecs, destination, transmission mode and + /// checkpoint cadence remain fixed. Only explicitly bounded runtime knobs + /// may move, and each changed rule needs fresh evidence attributed to the + /// exact active generation. + pub fn authorize_successor( + &self, + successor: &TransmissionPlan, + evidence: &[RuntimeAdaptationEvidence], + now_unix_ms: u64, + ) -> Result<(), TransmissionPlanError> { + if successor.envelope.plan_id != self.envelope.plan_id + || successor.envelope.plan_version != self.envelope.plan_version.saturating_add(1) + || successor.envelope.backend_compat != self.envelope.backend_compat + || successor.envelope.planner_revision != self.envelope.planner_revision + || successor.envelope.capability_snapshot_id != self.envelope.capability_snapshot_id + || successor.envelope.generated_at_unix_ms < self.envelope.generated_at_unix_ms + || successor.envelope.activation_unix_ms < successor.envelope.generated_at_unix_ms + || successor.rules.len() != self.rules.len() + || successor.frame_identity != self.frame_identity + { + return Err(TransmissionPlanError::InvalidSuccessor( + "successor must be the next version of the same semantic/capability generation" + .into(), + )); + } + + for current in &self.rules { + let Some(next) = successor.rules.iter().find(|candidate| { + candidate.materialization == current.materialization + && candidate.producer_id == current.producer_id + && candidate.schema_id == current.schema_id + }) else { + return Err(TransmissionPlanError::InvalidSuccessor(format!( + "missing rule for producer {}", + current.producer_id + ))); + }; + if current.mode != next.mode + || current.encoding != next.encoding + || current.full_checkpoint_every_ms != next.full_checkpoint_every_ms + || current.destination_ref != next.destination_ref + || current.runtime_policy.adaptation != next.runtime_policy.adaptation + || sampling_estimator(¤t.runtime_policy.sampling) + != sampling_estimator(&next.runtime_policy.sampling) + || delta_shape(¤t.runtime_policy.delta) + != delta_shape(&next.runtime_policy.delta) + { + return Err(TransmissionPlanError::InvalidSuccessor(format!( + "rule identity/codec/mode/guardrails drifted for producer {}", + current.producer_id + ))); + } + if current.emit_every_ms == next.emit_every_ms + && current.runtime_policy.sampling == next.runtime_policy.sampling + && current.runtime_policy.delta == next.runtime_policy.delta + { + continue; + } + let policy = ¤t.runtime_policy.adaptation; + if !policy.enabled || now_unix_ms < policy.not_before_unix_ms { + return Err(TransmissionPlanError::AdaptationOutOfBounds { + producer_id: current.producer_id.clone(), + knob: "adaptation_disabled_or_in_cooldown", + }); + } + let has_evidence = evidence.iter().any(|item| { + item.plan_id == self.envelope.plan_id + && item.plan_version == self.envelope.plan_version + && item.materialization == current.materialization + && item.producer_id == current.producer_id + && item.schema_id == current.schema_id + && !item.producer_version.trim().is_empty() + && item.sample_count >= policy.min_evidence_samples + && item.observed_at_unix_ms <= now_unix_ms + && now_unix_ms.saturating_sub(item.observed_at_unix_ms) + <= policy.max_evidence_age_ms + }); + if !has_evidence { + return Err(TransmissionPlanError::InvalidAdaptationEvidence( + current.producer_id.clone(), + )); + } + authorize_f64_change( + sampling_probability(¤t.runtime_policy.sampling), + sampling_probability(&next.runtime_policy.sampling), + policy.sample_probability.as_ref(), + ¤t.producer_id, + "sample_probability", + )?; + authorize_u64_change( + current.emit_every_ms, + next.emit_every_ms, + policy.emit_every_ms.as_ref(), + ¤t.producer_id, + "emit_every_ms", + )?; + authorize_f64_change( + delta_threshold(¤t.runtime_policy.delta), + delta_threshold(&next.runtime_policy.delta), + policy.delta_threshold.as_ref(), + ¤t.producer_id, + "delta_threshold", + )?; + authorize_f64_change( + gos_epsilon(¤t.runtime_policy.delta), + gos_epsilon(&next.runtime_policy.delta), + policy.gos_epsilon_staleness.as_ref(), + ¤t.producer_id, + "gos_epsilon_staleness", + )?; + } + Ok(()) + } + + pub fn validate_frame( + &self, + frame: &SummaryFrameIdentity, + ) -> Result<(), TransmissionPlanError> { + if frame.identity_version != self.frame_identity.identity_version + || frame.plan_id != self.envelope.plan_id + || frame.plan_version != self.envelope.plan_version + || frame.backend_compat != self.envelope.backend_compat + || frame.series_identity.is_empty() + || frame.producer_epoch.is_empty() + || frame.sequence == 0 + || frame.window_start_unix_nano >= frame.window_end_unix_nano + || (frame.kind == SummaryFrameKind::Full + && self.frame_identity.require_checkpoint_for_full + && frame.checkpoint_id.is_none()) + || (frame.kind == SummaryFrameKind::Delta + && self.frame_identity.require_base_checkpoint_for_delta + && frame.base_checkpoint_id.is_none()) + { + return Err(TransmissionPlanError::InvalidFrame( + "identity/lifecycle/window/checkpoint fields do not satisfy the active contract" + .into(), + )); + } + if self.rules.iter().any(|rule| { + rule.materialization == frame.materialization + && rule.producer_id == frame.producer_id + && rule.schema_id == frame.schema_id + // A delta rule necessarily emits periodic full checkpoints; + // a full-only rule must never emit deltas. + && (frame.kind == SummaryFrameKind::Full + || rule.mode == TransmissionMode::Delta) + && rule.encoding == frame.encoding + }) { + Ok(()) + } else { + Err(TransmissionPlanError::UnknownFrame) + } + } +} + +fn validate_runtime_rule_policy( + rule: &TransmissionRule, + family: &StateFamilyContract, +) -> Result<(), TransmissionPlanError> { + let invalid = |reason: &str| TransmissionPlanError::InvalidRuntimePolicy { + producer_id: rule.producer_id.clone(), + reason: reason.into(), + }; + if let SamplingPolicy::Fixed { + probability, + estimator, + } = &rule.runtime_policy.sampling + { + if !probability.is_finite() || !(0.0..=1.0).contains(probability) || *probability == 0.0 { + return Err(invalid("sample probability must be finite and in (0, 1]")); + } + let supported = matches!( + (family, *estimator), + ( + StateFamilyContract::Sketch { + algorithm: SketchAlgorithm::Hll, + .. + }, + SamplingEstimator::HashThreshold, + ) | ( + StateFamilyContract::Sketch { + algorithm: SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap, + .. + }, + SamplingEstimator::GeometricAdmission, + ) + ); + if !supported { + return Err(invalid( + "sampling estimator is not implemented for the materialization family", + )); + } + } + + if (rule.mode == TransmissionMode::Delta) != rule.runtime_policy.delta.is_some() { + return Err(invalid( + "delta policy must be present exactly when transmission mode is delta", + )); + } + if let Some(delta) = &rule.runtime_policy.delta { + if !delta.absolute_threshold.is_finite() || delta.absolute_threshold < 0.0 { + return Err(invalid("delta threshold must be finite and non-negative")); + } + if !matches!( + family, + StateFamilyContract::Sketch { + algorithm: SketchAlgorithm::DDSketch + | SketchAlgorithm::Hll + | SketchAlgorithm::Cms + | SketchAlgorithm::CmsWithHeap + | SketchAlgorithm::CountSketch + | SketchAlgorithm::CountSketchWithHeap, + .. + } + ) { + return Err(invalid( + "delta transmission is not implemented for the materialization family", + )); + } + if let Some(gos) = &delta.gos { + if !matches!( + family, + StateFamilyContract::Sketch { + algorithm: SketchAlgorithm::CountSketch | SketchAlgorithm::CountSketchWithHeap, + .. + } + ) || !gos.epsilon_staleness.is_finite() + || !(0.0..=1.0).contains(&gos.epsilon_staleness) + || gos.epsilon_staleness == 0.0 + || gos.sites == 0 + { + return Err(invalid( + "GOS requires a CountSketch family, epsilon in (0, 1], and at least one site", + )); + } + } + } + + let adaptation = &rule.runtime_policy.adaptation; + if adaptation.enabled + && (adaptation.max_evidence_age_ms == 0 || adaptation.min_evidence_samples == 0) + { + return Err(invalid( + "enabled adaptation requires non-zero evidence age and sample-count requirements", + )); + } + validate_f64_bounds(adaptation.sample_probability.as_ref(), 0.0, 1.0) + .map_err(|reason| invalid(reason))?; + validate_f64_bounds(adaptation.delta_threshold.as_ref(), 0.0, f64::MAX) + .map_err(|reason| invalid(reason))?; + validate_f64_bounds(adaptation.gos_epsilon_staleness.as_ref(), 0.0, 1.0) + .map_err(|reason| invalid(reason))?; + if let Some(bounds) = &adaptation.emit_every_ms { + if bounds.min == 0 + || bounds.min > bounds.max + || bounds.max_step == 0 + || !(bounds.min..=bounds.max).contains(&rule.emit_every_ms) + { + return Err(invalid("emit interval guardrails are invalid")); + } + } + if let Some(bounds) = &adaptation.sample_probability { + let current = sampling_probability(&rule.runtime_policy.sampling); + if current < bounds.min || current > bounds.max { + return Err(invalid( + "current sampling probability is outside guardrails", + )); + } + } + if let Some(bounds) = &adaptation.delta_threshold { + let Some(current) = rule + .runtime_policy + .delta + .as_ref() + .map(|policy| policy.absolute_threshold) + else { + return Err(invalid("delta guardrails require an active delta policy")); + }; + if current < bounds.min || current > bounds.max { + return Err(invalid("current delta threshold is outside guardrails")); + } + } + if let Some(bounds) = &adaptation.gos_epsilon_staleness { + let Some(current) = rule + .runtime_policy + .delta + .as_ref() + .and_then(|policy| policy.gos.as_ref()) + .map(|gos| gos.epsilon_staleness) + else { + return Err(invalid("GOS guardrails require an active GOS policy")); + }; + if current < bounds.min || current > bounds.max { + return Err(invalid("current GOS epsilon is outside guardrails")); + } + } + Ok(()) +} + +fn validate_f64_bounds( + bounds: Option<&AdaptiveF64Bounds>, + domain_min: f64, + domain_max: f64, +) -> Result<(), &'static str> { + let Some(bounds) = bounds else { + return Ok(()); + }; + if !bounds.min.is_finite() + || !bounds.max.is_finite() + || !bounds.max_step.is_finite() + || bounds.min < domain_min + || bounds.max > domain_max + || bounds.min > bounds.max + || bounds.max_step <= 0.0 + { + Err("floating-point adaptation guardrails are invalid") + } else { + Ok(()) + } +} + +fn sampling_probability(policy: &SamplingPolicy) -> f64 { + match policy { + SamplingPolicy::Disabled => 1.0, + SamplingPolicy::Fixed { probability, .. } => *probability, + } +} + +fn sampling_estimator(policy: &SamplingPolicy) -> Option { + match policy { + SamplingPolicy::Disabled => None, + SamplingPolicy::Fixed { estimator, .. } => Some(*estimator), + } +} + +fn delta_threshold(policy: &Option) -> f64 { + policy + .as_ref() + .map(|policy| policy.absolute_threshold) + .unwrap_or(0.0) +} + +fn gos_epsilon(policy: &Option) -> f64 { + policy + .as_ref() + .and_then(|policy| policy.gos.as_ref()) + .map(|gos| gos.epsilon_staleness) + .unwrap_or(0.0) +} + +fn delta_shape(policy: &Option) -> Option<(Option<(u32, GosThresholdMode)>,)> { + policy.as_ref().map(|policy| { + (policy + .gos + .as_ref() + .map(|gos| (gos.sites, gos.threshold_mode)),) + }) +} + +fn authorize_f64_change( + current: f64, + next: f64, + bounds: Option<&AdaptiveF64Bounds>, + producer_id: &str, + knob: &'static str, +) -> Result<(), TransmissionPlanError> { + if current == next { + return Ok(()); + } + let allowed = bounds.is_some_and(|bounds| { + next.is_finite() + && (bounds.min..=bounds.max).contains(&next) + && (next - current).abs() <= bounds.max_step + }); + if allowed { + Ok(()) + } else { + Err(TransmissionPlanError::AdaptationOutOfBounds { + producer_id: producer_id.into(), + knob, + }) + } +} + +fn authorize_u64_change( + current: u64, + next: u64, + bounds: Option<&AdaptiveU64Bounds>, + producer_id: &str, + knob: &'static str, +) -> Result<(), TransmissionPlanError> { + if current == next { + return Ok(()); + } + let allowed = bounds.is_some_and(|bounds| { + (bounds.min..=bounds.max).contains(&next) && current.abs_diff(next) <= bounds.max_step + }); + if allowed { + Ok(()) + } else { + Err(TransmissionPlanError::AdaptationOutOfBounds { + producer_id: producer_id.into(), + knob, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use planner_types::post_asap::SketchParams; + + #[test] + fn runtime_policy_is_family_and_mode_checked() { + let mut rule = TransmissionRule { + materialization: crate::PolicyFingerprint(1).into(), + producer_id: "test".into(), + schema_id: "test".into(), + mode: TransmissionMode::Full, + encoding: StateEncoding::SketchlibProtobufV1, + emit_every_ms: 60_000, + full_checkpoint_every_ms: None, + destination_ref: "backend".into(), + runtime_policy: Default::default(), + }; + rule.runtime_policy.sampling = SamplingPolicy::Fixed { + probability: 0.5, + estimator: SamplingEstimator::HashThreshold, + }; + let hll = StateFamilyContract::Sketch { + algorithm: SketchAlgorithm::Hll, + parameters: SketchParams::Hll { precision: 14 }, + }; + let count_sketch = StateFamilyContract::Sketch { + algorithm: SketchAlgorithm::CountSketch, + parameters: SketchParams::CountSketch { + width: 128, + depth: 4, + }, + }; + validate_runtime_rule_policy(&rule, &hll).expect("HLL supports hash-threshold sampling"); + assert!(matches!( + validate_runtime_rule_policy(&rule, &count_sketch), + Err(TransmissionPlanError::InvalidRuntimePolicy { .. }) + )); + + rule.runtime_policy.sampling = SamplingPolicy::Disabled; + rule.mode = TransmissionMode::Delta; + rule.full_checkpoint_every_ms = Some(300_000); + rule.runtime_policy.delta = Some(DeltaPolicy { + absolute_threshold: 0.0, + gos: Some(GosPolicy { + epsilon_staleness: 0.02, + sites: 2, + threshold_mode: GosThresholdMode::Isotropic, + }), + }); + validate_runtime_rule_policy(&rule, &count_sketch).expect("CountSketch supports delta GOS"); + assert!(matches!( + validate_runtime_rule_policy(&rule, &hll), + Err(TransmissionPlanError::InvalidRuntimePolicy { .. }) + )); + } +} diff --git a/data_plane/examples/audit_clickhouse_fallback.rs b/data_plane/examples/audit_clickhouse_fallback.rs index 2ada8ebaf..a5bdd1ac0 100644 --- a/data_plane/examples/audit_clickhouse_fallback.rs +++ b/data_plane/examples/audit_clickhouse_fallback.rs @@ -3,9 +3,7 @@ use axum::{ http::{HeaderMap, Method}, }; use control_plane::{ - physical::compiler::{ - PlanEnvelope, PrecomputePlan, TransmissionPlan, BACKEND_COMPAT, PLANNER_REVISION, - }, + physical::compiler::{PlanEnvelope, PrecomputePlan, BACKEND_COMPAT, PLANNER_REVISION}, query_plan::{ClickHousePlanningContext, QueryPlan}, }; use data_plane::{ @@ -76,8 +74,12 @@ async fn main() { let mut precompute_plan = PrecomputePlan::build_backend_local(envelope.clone(), vec![]).unwrap(); precompute_plan.summary_catalog = Some(reference.clone()); - let mut transmission_plan = - TransmissionPlan::build(envelope, &precompute_plan, &BTreeMap::new()).unwrap(); + let mut transmission_plan = control_plane::physical::compiler::compile_transmission_plan( + envelope, + &precompute_plan, + &BTreeMap::new(), + ) + .unwrap(); transmission_plan.summary_catalog = Some(reference); let query_plan = QueryPlan { plan_id: 27, diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 7e65edadc..2961bb7aa 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -1714,7 +1714,7 @@ async fn route_modified_otlp_sketches_to_precompute( crate::precompute_engine::frame_lineage::FrameLineageDecision::Apply, ) => { if frame.kind - == control_plane::physical::compiler::SummaryFrameKind::Full + == asap_types::producer_plan::SummaryFrameKind::Full { ingest_state .sketch_index @@ -2253,7 +2253,7 @@ fn preflight_summary_frames( mut dp: ModifiedOtlpSketchDp, ingest_state: &IngestState, active: &crate::storage_engines::types::ActivePhysicalPlan, - ) -> Result { + ) -> Result { let canonical_name = canonical_sketch_metric_name(metric_name, dp.algorithm.clone()); let frame = take_summary_frame_identity(&mut dp.attrs, dp.start_time_unix_nano, dp.time_unix_nano)?; @@ -2296,7 +2296,7 @@ fn preflight_summary_frames( // A malformed full snapshot must not be discovered after an earlier // frame in the request has already reached SketchStore. - if frame.kind == control_plane::physical::compiler::SummaryFrameKind::Full { + if frame.kind == asap_types::producer_plan::SummaryFrameKind::Full { decode_modified_otlp_sketch_bytes(dp.algorithm.clone(), dp.encoding, &dp.sketch) .map_err(|error| format!("invalid full frame for {metric_name}: {error}"))?; } else { @@ -2454,10 +2454,9 @@ fn take_summary_frame_identity( attrs: &mut HashMap, window_start_unix_nano: u64, window_end_unix_nano: u64, -) -> Result { - use control_plane::physical::compiler::{ - StateEncoding, SummaryFrameIdentity, SummaryFrameKind, - }; +) -> Result { + use asap_types::producer_plan::{SummaryFrameIdentity, SummaryFrameKind}; + use control_plane::physical::compiler::StateEncoding; fn required(attrs: &mut HashMap, key: &str) -> Result { attrs diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index be9905bb5..bbe1fbcad 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -895,9 +895,10 @@ mod tests { } fn physical_config(streaming: StreamingConfig) -> HotReloadStreamingConfig { + use asap_types::producer_plan::{FrameIdentityContract, SequenceScope, TransmissionPlan}; use control_plane::physical::compiler::{ - FrameIdentityContract, IngestContract, IngestProtocol, PlanEnvelope, PrecomputePlan, - SequenceScope, TimestampUnit, TransmissionPlan, PLANNER_REVISION, + IngestContract, IngestProtocol, PlanEnvelope, PrecomputePlan, TimestampUnit, + PLANNER_REVISION, }; let envelope = PlanEnvelope { plan_id: 7, diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index e219aca43..998a96ed3 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -2679,9 +2679,10 @@ mod tests { } async fn setup_remote_write_test_server() -> (u16, PrometheusRemoteWriteReceiver) { + use asap_types::producer_plan::{FrameIdentityContract, SequenceScope, TransmissionPlan}; use control_plane::physical::compiler::{ - FrameIdentityContract, IngestContract, IngestProtocol, PlanEnvelope, PrecomputePlan, - SequenceScope, TimestampUnit, TransmissionPlan, PLANNER_REVISION, + IngestContract, IngestProtocol, PlanEnvelope, PrecomputePlan, TimestampUnit, + PLANNER_REVISION, }; let streaming_config = Arc::new(StreamingConfig::default()); let envelope = PlanEnvelope { @@ -6045,7 +6046,7 @@ async fn handle_post_streaming_config( (StatusCode::OK, axum::Json(body)).into_response() } -pub use control_plane::physical::publication::PhysicalPlanInstallRequest; +pub use asap_types::plan_publication::PhysicalPlanInstallRequest; /// Decode and cross-validate every backend view before it can become visible. /// Used by both startup artifact loading and the staged HTTP install path. diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 23409c3b1..97f6386d8 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -768,13 +768,13 @@ async fn main() -> Result<()> { .collect(), executable_dags: Default::default(), }; - let initial_transmission_plan = control_plane::physical::compiler::TransmissionPlan { + let initial_transmission_plan = asap_types::producer_plan::TransmissionPlan { summary_catalog: None, envelope: initial_precompute_plan.envelope.clone(), - frame_identity: control_plane::physical::compiler::FrameIdentityContract { + frame_identity: asap_types::producer_plan::FrameIdentityContract { identity_version: 1, sequence_scope: - control_plane::physical::compiler::SequenceScope::MaterializationSeriesProducerEpoch, + asap_types::producer_plan::SequenceScope::MaterializationSeriesProducerEpoch, require_checkpoint_for_full: true, require_base_checkpoint_for_delta: true, }, diff --git a/data_plane/src/precompute_engine/frame_lineage.rs b/data_plane/src/precompute_engine/frame_lineage.rs index 6598a1384..7f5e0a847 100644 --- a/data_plane/src/precompute_engine/frame_lineage.rs +++ b/data_plane/src/precompute_engine/frame_lineage.rs @@ -5,7 +5,7 @@ //! the wire contract: deltas are only safe to apply after an observed full //! checkpoint, in sequence, within the exact producer/window lineage. -use control_plane::physical::compiler::{SummaryFrameIdentity, SummaryFrameKind}; +use asap_types::producer_plan::{SummaryFrameIdentity, SummaryFrameKind}; use dashmap::mapref::entry::Entry; use thiserror::Error; diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index 2d8b4a86d..907b7fea5 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs @@ -703,7 +703,7 @@ mod tests { ) .unwrap(); precompute.summary_catalog = Some(sds.reference().unwrap()); - let mut transmission = control_plane::physical::compiler::TransmissionPlan::build( + let mut transmission = control_plane::physical::compiler::compile_transmission_plan( envelope.clone(), &precompute, &BTreeMap::new(), 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 e6f355a4b..1066e9f70 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -537,8 +537,8 @@ struct IncompleteSummaryLineage { window_end_unix_ms: u64, } -impl From<&control_plane::physical::compiler::SummaryFrameIdentity> for IncompleteSummaryLineage { - fn from(frame: &control_plane::physical::compiler::SummaryFrameIdentity) -> Self { +impl From<&asap_types::producer_plan::SummaryFrameIdentity> for IncompleteSummaryLineage { + fn from(frame: &asap_types::producer_plan::SummaryFrameIdentity) -> Self { Self { plan_id: frame.plan_id, plan_version: frame.plan_version, @@ -2679,7 +2679,7 @@ impl SketchStore { pub fn mark_summary_lineage_incomplete( &self, sid: u64, - frame: &control_plane::physical::compiler::SummaryFrameIdentity, + frame: &asap_types::producer_plan::SummaryFrameIdentity, ) { self.incomplete_summary_lineages .entry(sid) @@ -2691,7 +2691,7 @@ impl SketchStore { pub fn clear_summary_lineage_incomplete( &self, sid: u64, - frame: &control_plane::physical::compiler::SummaryFrameIdentity, + frame: &asap_types::producer_plan::SummaryFrameIdentity, ) { let key = IncompleteSummaryLineage::from(frame); if let Some(mut lineages) = self.incomplete_summary_lineages.get_mut(&sid) { @@ -3613,9 +3613,8 @@ mod tests { #[test] fn incomplete_delta_window_fails_closed_until_matching_full_checkpoint() { - use control_plane::physical::compiler::{ - StateEncoding, SummaryFrameIdentity, SummaryFrameKind, - }; + use asap_types::producer_plan::{SummaryFrameIdentity, SummaryFrameKind}; + use control_plane::physical::compiler::StateEncoding; let idx = SketchStore::new(); idx.register(meta(12)); 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 9cba552f9..88786343c 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -92,7 +92,7 @@ pub struct ActivePhysicalPlan { /// Present for authoritative installations; legacy bootstrap has no catalog. pub summary_catalog: Option>, pub precompute_plan: asap_types::precompute_plan::PrecomputePlan, - pub transmission_plan: control_plane::physical::compiler::TransmissionPlan, + pub transmission_plan: asap_types::producer_plan::TransmissionPlan, pub runtime_config: Arc, pub query_plan: Arc, pub storage_routing: Arc, @@ -672,11 +672,9 @@ mod tests { summary_catalog: None, envelope: envelope.clone(), ingest: asap_types::precompute_plan::IngestContract { - protocol: - asap_types::precompute_plan::IngestProtocol::ModifiedOtlpMetricsV1, + protocol: asap_types::precompute_plan::IngestProtocol::ModifiedOtlpMetricsV1, endpoint_path: "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/v1/metrics".into(), - timestamp_unit: - asap_types::precompute_plan::TimestampUnit::UnixNanoseconds, + timestamp_unit: asap_types::precompute_plan::TimestampUnit::UnixNanoseconds, require_plan_identity: true, require_summary_definition_identity: true, require_registered_producer: true, @@ -684,9 +682,9 @@ mod tests { schemas: Vec::new(), producers: Vec::new(), executable_dags: Default::default(), - materializations: Vec::new(), + materializations: Vec::new(), }, - transmission_plan: control_plane::physical::compiler::TransmissionPlan { + transmission_plan: asap_types::producer_plan::TransmissionPlan { summary_catalog: None, envelope: asap_types::precompute_plan::PlanEnvelope { plan_id, @@ -698,9 +696,10 @@ mod tests { planner_revision: control_plane::physical::compiler::PLANNER_REVISION.into(), capability_snapshot_id: "test".into(), }, - frame_identity: control_plane::physical::compiler::FrameIdentityContract { + frame_identity: asap_types::producer_plan::FrameIdentityContract { identity_version: 1, - sequence_scope: control_plane::physical::compiler::SequenceScope::MaterializationSeriesProducerEpoch, + sequence_scope: + asap_types::producer_plan::SequenceScope::MaterializationSeriesProducerEpoch, require_checkpoint_for_full: true, require_base_checkpoint_for_delta: true, }, diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index 0bf5ff300..cc1b733fa 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -74,9 +74,9 @@ async fn post_full_config(client: &reqwest::Client, stack: &FullStack, json: &Js .any(|c| c.metric == "http_requests_total_latency_ms") { for rule in &mut artifact.transmission_plan.rules { - rule.mode = control_plane::physical::compiler::TransmissionMode::Delta; + rule.mode = asap_types::producer_plan::TransmissionMode::Delta; rule.full_checkpoint_every_ms = Some(rule.emit_every_ms); - rule.runtime_policy.delta = Some(control_plane::physical::compiler::DeltaPolicy { + rule.runtime_policy.delta = Some(asap_types::producer_plan::DeltaPolicy { absolute_threshold: 0.0, gos: None, }); diff --git a/data_plane/tests/support/physical_fixture.rs b/data_plane/tests/support/physical_fixture.rs index 00ae0e862..984fc0d16 100644 --- a/data_plane/tests/support/physical_fixture.rs +++ b/data_plane/tests/support/physical_fixture.rs @@ -37,8 +37,12 @@ pub fn artifact(config: &StreamingConfig) -> PhysicalPlanInstallRequest { let mut precompute = PrecomputePlan::build(envelope.clone(), configs, &["fixture".into()]).unwrap(); precompute.summary_catalog = Some(catalog.reference().unwrap()); - let mut transmission = - TransmissionPlan::build(envelope, &precompute, &BTreeMap::new()).unwrap(); + let mut transmission = control_plane::physical::compiler::compile_transmission_plan( + envelope, + &precompute, + &BTreeMap::new(), + ) + .unwrap(); transmission.summary_catalog = Some(catalog.reference().unwrap()); let mut query_plan = QueryPlan { plan_id: 1, diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index b712c0d94..09f2fab20 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -169,6 +169,13 @@ those types for existing callers and owns the `compile_bound*` and move into the shared contract. Data-plane engines import the shared types directly. No wrapper plan or second wire definition is introduced. +`asap_types::producer_plan` owns the installed collector and transmission +contracts, frame identities, runtime policy bounds and their validation. The +control plane allocates sampling/GOS budgets and constructs transmission rules +through `sampling_policy_from_accuracy_budget`, `gos_policy_from_accuracy_budget` +and `compile_transmission_plan`. Producers and the data plane import the shared +contracts directly; compilation is not a runtime dependency of those contracts. + The implemented ownership split is: 1. Move the SDS catalog contract into `asap_types`.