diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index 7fd82482e..230f70953 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -370,6 +370,7 @@ impl BackendClient { backend_plan: Vec, query_plan: &crate::query_plan::QueryPlan, storage_routing: Option, + adaptation_evidence: &[crate::physical::compiler::RuntimeAdaptationEvidence], ) -> std::result::Result<(), BackendPostError> { let url = derive_physical_plan_url(&self.endpoint); let response = self @@ -381,6 +382,7 @@ impl BackendClient { "backend_plan": backend_plan, "query_plan": query_plan, "storage_routing": storage_routing, + "adaptation_evidence": adaptation_evidence, })) .send() .await diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index 0d8a08ce9..a748377df 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -255,6 +255,7 @@ async fn push_documents_coupled( let transmission_plan = match crate::physical::compiler::TransmissionPlan::build( precompute_plan.envelope.clone(), precompute_plan, + &Default::default(), ) { Ok(plan) => plan, Err(error) => { @@ -271,6 +272,7 @@ async fn push_documents_coupled( plan_bytes.clone(), &query_plan, Some(routing.clone()), + &[], ) .await { diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index f3ba480c5..f0f5d804c 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -583,6 +583,8 @@ struct PhysicalPlanQueryRequest { accuracy: types_v2::AccuracyTarget, lifecycle: physical::compiler::LifecyclePlanningInput, window_implementations: Vec, + #[serde(default)] + runtime_policy: physical::compiler::RuntimeRulePolicy, } #[derive(Debug, Deserialize)] @@ -593,6 +595,8 @@ struct CompileAndPublishPhysicalPlanRequest { capability_snapshot_id: String, #[serde(default)] evidence: HashMap, + #[serde(default)] + runtime_adaptation_evidence: Vec, planner_revision: String, max_evidence_age_ms: u64, plan_version: u64, @@ -624,10 +628,11 @@ async fn handle_compile_and_publish_physical_plan( State(st): State, Json(request): Json, ) -> impl IntoResponse { - let (bundle, collector_ids, apply_timeout) = match compile_physical_plan_request(request) { - Ok(compiled) => compiled, - Err(response) => return response.into_response(), - }; + let (bundle, collector_ids, apply_timeout, adaptation_evidence) = + match compile_physical_plan_request(request) { + Ok(compiled) => compiled, + Err(response) => return response.into_response(), + }; let Some(backend) = st.backend_client.as_ref() else { return ( @@ -654,6 +659,7 @@ async fn handle_compile_and_publish_physical_plan( bundle.backend_plan.encode_to_vec(), &bundle.query_plan, None, + &adaptation_evidence, ) .await { @@ -714,7 +720,15 @@ async fn handle_compile_and_publish_physical_plan( // Only the Send-safe compiled bundle crosses an await point. fn compile_physical_plan_request( request: CompileAndPublishPhysicalPlanRequest, -) -> Result<(physical::compiler::PhysicalPlan, Vec, Duration), (StatusCode, String)> { +) -> Result< + ( + physical::compiler::PhysicalPlan, + Vec, + Duration, + Vec, + ), + (StatusCode, String), +> { if request.queries.is_empty() || request.collector_ids.is_empty() { return Err(( StatusCode::UNPROCESSABLE_ENTITY, @@ -780,6 +794,7 @@ fn compile_physical_plan_request( accuracy: query.accuracy, lifecycle: query.lifecycle, window_implementations: query.window_implementations, + runtime_policy: query.runtime_policy, }); } @@ -804,7 +819,12 @@ fn compile_physical_plan_request( Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; let apply_timeout = Duration::from_millis(request.apply_timeout_ms); - Ok((bundle, request.collector_ids, apply_timeout)) + Ok(( + bundle, + request.collector_ids, + apply_timeout, + request.runtime_adaptation_evidence, + )) } // ── Handlers ────────────────────────────────────────────────────────────────── diff --git a/control_plane/src/opamp/mod.rs b/control_plane/src/opamp/mod.rs index 80603a285..74e73550b 100644 --- a/control_plane/src/opamp/mod.rs +++ b/control_plane/src/opamp/mod.rs @@ -54,6 +54,7 @@ pub const PLAN_STATUS_MESSAGE: &str = "plan_status"; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub enum CollectorPlanStatusKind { + Staged, Applied, Failed, } @@ -283,8 +284,8 @@ impl OpampServer { } /// Publish all per-target physical plans and require an exact semantic - /// APPLIED report from every Collector. Config hashes are deliberately not - /// accepted as plan activation evidence. + /// STAGED report from every Collector. Activation is synchronized later at + /// the envelope timestamp; config hashes are not accepted as evidence. pub async fn publish_collector_plans( &self, plans: &[crate::physical::compiler::CollectorPlan], @@ -327,7 +328,7 @@ impl OpampServer { plan_id: plan.envelope.plan_id, plan_version: plan.envelope.plan_version, })?; - if report.status != CollectorPlanStatusKind::Applied { + if report.status != CollectorPlanStatusKind::Staged { return Err(CollectorPlanPublishError::Rejected { collector_id: plan.collector_id.clone(), plan_id: report.plan_id, @@ -1044,6 +1045,7 @@ mod tests { emit_every_ms: 60_000, full_checkpoint_every_ms: None, destination_ref: "asapquery-backend".into(), + runtime_policy: crate::physical::compiler::RuntimeRulePolicy::default(), }], } } @@ -1065,7 +1067,7 @@ mod tests { } #[tokio::test] - async fn typed_plan_publication_waits_for_capability_and_exact_applied_status() { + async fn typed_plan_publication_waits_for_capability_and_exact_staged_status() { let (srv, addr) = start_server().await; let mut agent_ws = connect_ws_client(addr, "edge-a", "agent").await; @@ -1105,7 +1107,7 @@ mod tests { let status = serde_json::to_vec(&CollectorPlanStatus { plan_id: 42, plan_version: 1, - status: CollectorPlanStatusKind::Applied, + status: CollectorPlanStatusKind::Staged, error: None, }) .unwrap(); @@ -1126,7 +1128,7 @@ mod tests { assert_eq!(reports.len(), 1); assert_eq!(reports[0].plan_id, 42); assert_eq!(reports[0].plan_version, 1); - assert_eq!(reports[0].status, CollectorPlanStatusKind::Applied); + assert_eq!(reports[0].status, CollectorPlanStatusKind::Staged); } #[tokio::test] diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index a6489165a..ea9f7050b 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -65,6 +65,10 @@ pub struct PlanningQuery { /// identities and exposes only framework + complete weighted cost to /// Planner. An empty or stale set fails closed. pub window_implementations: Vec, + /// Physical runtime policy selected for this Planner materialization. + /// It is validated against the selected summary family during physical + /// compilation and becomes part of the immutable plan generation. + pub runtime_policy: RuntimeRulePolicy, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] @@ -438,6 +442,18 @@ impl PrecomputePlan { )); } } + let mut schema_ids = BTreeSet::new(); + for schema in &self.schemas { + if schema.schema_id.trim().is_empty() + || !schema_ids.insert(schema.schema_id.as_str()) + || schema.schema_version == 0 + || schema.encodings.is_empty() + { + return Err(PrecomputePlanError::InvalidSchema { + schema_id: schema.schema_id.clone(), + }); + } + } let schemas: BTreeSet<_> = self .schemas .iter() @@ -446,13 +462,6 @@ impl PrecomputePlan { if schemas != materializations || schemas.len() != self.schemas.len() { return Err(PrecomputePlanError::SchemaSetMismatch); } - for schema in &self.schemas { - if schema.schema_version == 0 || schema.encodings.is_empty() { - return Err(PrecomputePlanError::InvalidSchema { - schema_id: schema.schema_id.clone(), - }); - } - } let schema_by_materialization: BTreeMap<_, _> = self .schemas .iter() @@ -541,7 +550,7 @@ pub enum TransmissionMode { #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum SequenceScope { - MaterializationWindowProducerEpoch, + MaterializationSeriesProducerEpoch, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -553,7 +562,7 @@ pub struct FrameIdentityContract { pub require_base_checkpoint_for_delta: bool, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] pub struct TransmissionRule { pub materialization: asap_types::PolicyFingerprint, @@ -564,10 +573,211 @@ pub struct TransmissionRule { 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::PolicyFingerprint, + 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 { pub envelope: PlanEnvelope, pub frame_identity: FrameIdentityContract, @@ -592,13 +802,11 @@ pub struct SummaryFrameIdentity { pub plan_version: u64, pub backend_compat: String, pub materialization: asap_types::PolicyFingerprint, + /// 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, - /// Canonical identity of the output series inside the materialization. - /// This remains stable when the sender switches between attribute-bearing - /// and SID-only frames, and prevents two series from sharing a receipt. - pub series_fingerprint: String, pub window_start_unix_nano: u64, pub window_end_unix_nano: u64, pub sequence: u64, @@ -620,12 +828,24 @@ pub enum TransmissionPlanError { 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, + }, } impl TransmissionPlan { pub fn build( envelope: PlanEnvelope, precompute: &PrecomputePlan, + runtime_policies: &BTreeMap, ) -> Result { if envelope != precompute.envelope { return Err(TransmissionPlanError::EnvelopeMismatch); @@ -647,15 +867,27 @@ impl TransmissionPlan { .iter() .find(|m| m.policy_fingerprint() == producer.materialization) .expect("validated PrecomputePlan materialization binding"); + let runtime_policy = runtime_policies + .get(&producer.materialization) + .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: TransmissionMode::Full, + mode, encoding: schema.encodings[0].clone(), - emit_every_ms: materialization.window_size.saturating_mul(1_000), - full_checkpoint_every_ms: None, + 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(); @@ -663,7 +895,7 @@ impl TransmissionPlan { envelope, frame_identity: FrameIdentityContract { identity_version: 1, - sequence_scope: SequenceScope::MaterializationWindowProducerEpoch, + sequence_scope: SequenceScope::MaterializationSeriesProducerEpoch, require_checkpoint_for_full: true, require_base_checkpoint_for_delta: true, }, @@ -703,15 +935,145 @@ impl TransmissionPlan { 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() - || (rule.mode == TransmissionMode::Delta - && rule - .full_checkpoint_every_ms - .map_or(true, |value| value == 0)) + || !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(()) } @@ -724,8 +1086,8 @@ impl TransmissionPlan { || 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.series_fingerprint.is_empty() || frame.sequence == 0 || frame.window_start_unix_nano >= frame.window_end_unix_nano || (frame.kind == SummaryFrameKind::Full @@ -740,15 +1102,14 @@ impl TransmissionPlan { .into(), )); } - let mode = match frame.kind { - SummaryFrameKind::Full => TransmissionMode::Full, - SummaryFrameKind::Delta => TransmissionMode::Delta, - }; if self.rules.iter().any(|rule| { rule.materialization == frame.materialization && rule.producer_id == frame.producer_id && rule.schema_id == frame.schema_id - && rule.mode == mode + // 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(()) @@ -758,6 +1119,258 @@ impl TransmissionPlan { } } +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("planner revision mismatch: request={request}, compiler={compiler}")] @@ -817,6 +1430,7 @@ impl PhysicalCompiler { let mut aggregations = Vec::with_capacity(request.queries.len()); let mut readouts = Vec::with_capacity(request.queries.len()); let mut collector_materializations = Vec::with_capacity(request.queries.len()); + let mut runtime_policies = BTreeMap::new(); for query in &request.queries { let evidence = request.evidence.get(&query.query_id); @@ -885,6 +1499,18 @@ impl PhysicalCompiler { let precompute_materialization = backend_plan::aggregation_config_for_materialization(&aggregation)?; let materialization = precompute_materialization.policy_fingerprint(); + if let Some(existing) = + runtime_policies.insert(materialization, query.runtime_policy.clone()) + { + if existing != query.runtime_policy { + return Err(CompileError::Query { + query_id: query.query_id.clone(), + reason: + "queries sharing one materialization specify different runtime policies" + .into(), + }); + } + } aggregations.push(aggregation); readouts.push(BackendReadout { aggregation_id, @@ -968,11 +1594,12 @@ impl PhysicalCompiler { query_id: "precompute-plan".into(), reason: error.to_string(), })?; - let transmission_plan = TransmissionPlan::build(envelope.clone(), &precompute_plan) - .map_err(|error| CompileError::Query { - query_id: "transmission-plan".into(), - reason: error.to_string(), - })?; + let 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 collector_plans = producer_ids .into_iter() .map(|collector_id| CollectorPlan { @@ -1195,7 +1822,8 @@ fn validate_evidence( let age = env .observed_at_unix_ms .saturating_sub(evidence.observed_at_unix_ms); - let valid = evidence.selected_lower_bound.is_finite() + let valid = evidence.observed_at_unix_ms <= env.observed_at_unix_ms + && evidence.selected_lower_bound.is_finite() && evidence.excluded_upper_bound.is_finite() && evidence.selected_lower_bound > evidence.excluded_upper_bound && (0.0..=1.0).contains(&evidence.interval_failure_probability) @@ -1249,7 +1877,8 @@ fn validate_window_implementations( let age = environment .observed_at_unix_ms .saturating_sub(evidence.observed_at_unix_ms); - let valid = !candidate.implementation_id.trim().is_empty() + let valid = evidence.observed_at_unix_ms <= environment.observed_at_unix_ms + && !candidate.implementation_id.trim().is_empty() && ids.insert(candidate.implementation_id.clone()) && !candidate.state_layout.trim().is_empty() && !evidence.model_version.trim().is_empty() @@ -1539,6 +2168,7 @@ mod tests { weighted_cost: 1.0, }, }], + runtime_policy: RuntimeRulePolicy::default(), }], evidence: evidence_by_query, planner_revision: PLANNER_REVISION.into(), @@ -1576,10 +2206,10 @@ mod tests { plan_version: bundle.envelope.plan_version, backend_compat: bundle.envelope.backend_compat.clone(), materialization: rule.materialization, + series_identity: "service=checkout,zone=a".into(), schema_id: rule.schema_id.clone(), producer_id: rule.producer_id.clone(), producer_epoch: "boot-1".into(), - series_fingerprint: "service=api".into(), window_start_unix_nano: 1, window_end_unix_nano: 2, sequence: 1, @@ -1743,6 +2373,30 @@ mod tests { )); } + #[test] + fn precompute_plan_rejects_empty_or_duplicate_schema_ids() { + let bundle = PhysicalCompiler + .compile( + request("q-quantile", "quantile_over_time(0.99, m[1m])"), + environment(10_000), + ) + .expect("compile schema"); + + let mut empty = bundle.precompute_plan.clone(); + empty.schemas[0].schema_id.clear(); + assert!(matches!( + empty.validate(), + Err(PrecomputePlanError::InvalidSchema { .. }) + )); + + let mut duplicate = bundle.precompute_plan; + duplicate.schemas.push(duplicate.schemas[0].clone()); + assert!(matches!( + duplicate.validate(), + Err(PrecomputePlanError::InvalidSchema { .. }) + )); + } + #[test] fn topk_fails_closed_without_membership_evidence() { assert!(request_with_evidence("q-topk", "topk(5, m)", None).is_err()); @@ -1768,6 +2422,35 @@ mod tests { assert!(matches!(error, CompileError::InvalidEvidence { .. })); } + #[test] + fn future_topk_and_window_evidence_are_rejected() { + let topk = request_with_evidence( + "q-topk", + "topk(5, count_over_time(m[1m]))", + Some(TopKMembershipEvidence { + selected_lower_bound: 101.0, + excluded_upper_bound: 100.0, + interval_failure_probability: 0.005, + observed_at_unix_ms: 10_001, + source: "runtime-margin-monitor".into(), + }), + ) + .expect("selection occurs before deployment-time freshness validation"); + assert!(matches!( + PhysicalCompiler.compile(topk, environment(10_000)), + Err(CompileError::InvalidEvidence { .. }) + )); + + let mut window = request("q-window", "quantile_over_time(0.99, m[1m])"); + window.queries[0].window_implementations[0] + .cost + .observed_at_unix_ms = 10_001; + assert!(matches!( + PhysicalCompiler.compile(window, environment(10_000)), + Err(CompileError::Lifecycle { .. }) + )); + } + #[test] fn fresh_topk_evidence_enables_physical_compilation() { let request = request_with_evidence( @@ -1814,4 +2497,207 @@ mod tests { Err(CompileError::Lifecycle { .. }) )); } + + #[test] + fn runtime_policy_uses_canonical_sampling_and_gos_allocators() { + let sampling = SamplingPolicy::from_accuracy_budget( + 0.05, + 1_000.0, + SamplingEstimator::GeometricAdmission, + ); + let SamplingPolicy::Fixed { probability, .. } = sampling else { + panic!("positive budget and rate must enable sampling"); + }; + 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) + .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() { + 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] + fn compiler_emits_delta_rule_and_accepts_periodic_full_checkpoint() { + let mut request = request_with_evidence( + "q-topk", + "topk(5, count_over_time(m[1m]))", + Some(TopKMembershipEvidence { + selected_lower_bound: 101.0, + excluded_upper_bound: 100.0, + interval_failure_probability: 0.005, + observed_at_unix_ms: 9_500, + source: "runtime-margin-monitor".into(), + }), + ) + .expect("frequency selection"); + request.queries[0].runtime_policy.delta = Some(DeltaPolicy { + absolute_threshold: 0.0, + gos: None, + }); + let bundle = PhysicalCompiler + .compile(request, environment(10_000)) + .expect("delta-capable physical plan"); + let rule = &bundle.transmission_plan.rules[0]; + assert_eq!(rule.mode, TransmissionMode::Delta); + assert!(rule.full_checkpoint_every_ms.is_some()); + + for (kind, sequence) in [(SummaryFrameKind::Full, 1), (SummaryFrameKind::Delta, 2)] { + let frame = SummaryFrameIdentity { + identity_version: 1, + plan_id: bundle.envelope.plan_id, + plan_version: bundle.envelope.plan_version, + backend_compat: bundle.envelope.backend_compat.clone(), + materialization: rule.materialization, + series_identity: "service=checkout,zone=a".into(), + schema_id: rule.schema_id.clone(), + producer_id: rule.producer_id.clone(), + producer_epoch: "boot-1".into(), + window_start_unix_nano: 1, + window_end_unix_nano: 2, + sequence, + encoding: rule.encoding.clone(), + checkpoint_id: (kind == SummaryFrameKind::Full).then(|| "cp-1".into()), + base_checkpoint_id: (kind == SummaryFrameKind::Delta).then(|| "cp-1".into()), + kind, + }; + bundle + .transmission_plan + .validate_frame(&frame) + .expect("delta rule accepts its deltas and recovery full frames"); + } + } + + #[test] + fn runtime_adaptation_requires_fresh_exact_evidence_and_successor_version() { + let bundle = PhysicalCompiler + .compile( + request("q", "quantile_over_time(0.99, m[1m])"), + environment(10_000), + ) + .expect("compile"); + let mut current = bundle.transmission_plan; + let rule = &mut current.rules[0]; + rule.runtime_policy.adaptation = RuntimeAdaptationPolicy { + enabled: true, + not_before_unix_ms: 10_500, + max_evidence_age_ms: 5_000, + min_evidence_samples: 100, + sample_probability: None, + emit_every_ms: Some(AdaptiveU64Bounds { + min: 30_000, + max: 120_000, + max_step: 10_000, + }), + delta_threshold: None, + gos_epsilon_staleness: None, + }; + let mut successor = current.clone(); + successor.envelope.plan_version += 1; + successor.envelope.generated_at_unix_ms = 11_000; + successor.envelope.activation_unix_ms = 11_000; + successor.rules[0].emit_every_ms += 5_000; + let evidence = RuntimeAdaptationEvidence { + plan_id: current.envelope.plan_id, + plan_version: current.envelope.plan_version, + materialization: current.rules[0].materialization, + producer_id: current.rules[0].producer_id.clone(), + schema_id: current.rules[0].schema_id.clone(), + producer_version: "collector.v1".into(), + observed_at_unix_ms: 10_900, + sample_count: 100, + }; + current + .authorize_successor(&successor, std::slice::from_ref(&evidence), 11_000) + .expect("bounded change with exact fresh evidence"); + + let mut oversized = successor.clone(); + oversized.rules[0].emit_every_ms += 20_000; + assert!(matches!( + current.authorize_successor(&oversized, std::slice::from_ref(&evidence), 11_000), + Err(TransmissionPlanError::AdaptationOutOfBounds { .. }) + )); + let stale = RuntimeAdaptationEvidence { + observed_at_unix_ms: 1, + ..evidence.clone() + }; + assert!(matches!( + current.authorize_successor(&successor, &[stale], 11_000), + Err(TransmissionPlanError::InvalidAdaptationEvidence(_)) + )); + let future = RuntimeAdaptationEvidence { + observed_at_unix_ms: 11_001, + ..evidence + }; + assert!(matches!( + current.authorize_successor(&successor, &[future], 11_000), + Err(TransmissionPlanError::InvalidAdaptationEvidence(_)) + )); + let mut different_plan_id = successor.clone(); + different_plan_id.envelope.plan_id += 1; + assert!(matches!( + current.authorize_successor(&different_plan_id, &[], 11_000), + Err(TransmissionPlanError::InvalidSuccessor(_)) + )); + let mut in_place = successor; + in_place.envelope.plan_version = current.envelope.plan_version; + assert!(matches!( + current.authorize_successor(&in_place, &[], 11_000), + Err(TransmissionPlanError::InvalidSuccessor(_)) + )); + } } diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index d28f2c345..1af9c00f0 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -873,10 +873,18 @@ async fn route_modified_otlp_sketches_to_precompute( use asap_otel_proto::tonic::metrics::v1::metric::Data; let ingest_received_at = Instant::now(); - let snap = ingest_state.config_snapshot(); - let active_physical_plan = ingest_state - .physical_plan_snapshot() - .filter(|plan| plan.backend_plan.plan_id != 0); + // Load the generation exactly once. Deriving both the runtime config and + // transmission contract from this Arc prevents an activation between two + // independent ArcSwap loads from producing a torn ingest view. + let physical_plan_snapshot = ingest_state.physical_plan_snapshot(); + let snap = physical_plan_snapshot + .as_ref() + .map(|plan| plan.runtime_config.clone()) + .unwrap_or_else(|| ingest_state.config_snapshot()); + let active_physical_plan = physical_plan_snapshot.filter(|plan| plan.backend_plan.plan_id != 0); + let lineage_batch_guard = active_physical_plan + .as_ref() + .map(|_| ingest_state.observability.frame_lineage.lock_batch()); if let Some(active) = active_physical_plan.as_ref() { // Validate the complete request before mutating the SID registry, // sketch store, snapshot cache, or worker queues. This makes the @@ -1284,10 +1292,9 @@ async fn route_modified_otlp_sketches_to_precompute( // and its key set IS the group-by KEY set. { use crate::storage_engines::sketch_db::index::{ - AccuracyBound, Capability, SketchAlgorithm, SketchEncoding, - SketchInstanceMetadata, SketchSampleState, + AccuracyBound, Capability, SketchAlgorithm, SketchInstanceMetadata, }; - use std::collections::{BTreeMap, BTreeSet}; + use std::collections::BTreeSet; if ingest_state.sketch_index.instance(sid).is_none() { let algorithm = sketch_algorithm_for(&dp); @@ -1348,7 +1355,6 @@ async fn route_modified_otlp_sketches_to_precompute( // below so the query engine can answer per-item estimate(key) // (the CMS/CountSketch FrequencyEstimate gate consults it). let item_label_for_sid: Option = { - let snap = ingest_state.config_snapshot(); snap.get_aggregation_config(policy_fp.as_u64()) .or_else(|| { snap.get_all_aggregation_configs() @@ -1432,27 +1438,6 @@ async fn route_modified_otlp_sketches_to_precompute( ); } } - - let label_values: BTreeMap = dp - .attrs - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect(); - let window: crate::storage_engines::sketch_db::index::epoch_columnar::TimestampRange = ( - dp.start_time_unix_nano / 1_000_000, - dp.time_unix_nano / 1_000_000, - ); - let encoding = - encoding_to_handle(dp.encoding).unwrap_or(SketchEncoding::ProtoFull); - ingest_state.sketch_index.append_sample( - sid, - label_values, - window, - SketchSampleState { - bytes: dp.sketch.clone(), - encoding, - }, - ); } // Encoding dispatch: full frames (PROTO / @@ -1588,17 +1573,6 @@ async fn route_modified_otlp_sketches_to_precompute( ); continue; } - ingest_state.sketch_snapshots.insert( - series_key.clone(), - crate::precompute_engine::ingest_handler::SnapshotCacheEntry { - core: merged.clone_boxed_core(), - window_start: dp.start_time_unix_nano, - }, - ); - // RES-1 — opportunistic eviction sweep keyed by the - // entry's window_start, bounding cache growth for - // churning high-cardinality series. - ingest_state.note_window_and_sweep(dp.start_time_unix_nano); merged } else { match decode_modified_otlp_sketch_bytes( @@ -1606,19 +1580,7 @@ async fn route_modified_otlp_sketches_to_precompute( dp.encoding, &dp.sketch, ) { - Ok(acc) => { - ingest_state.sketch_snapshots.insert( - series_key.clone(), - crate::precompute_engine::ingest_handler::SnapshotCacheEntry { - core: acc.clone_boxed_core(), - window_start: dp.start_time_unix_nano, - }, - ); - // RES-1 — sweep stale per-series bases on the - // full-frame insert too. - ingest_state.note_window_and_sweep(dp.start_time_unix_nano); - acc - } + Ok(acc) => acc, Err(e) => { ingest_state .observability @@ -1661,6 +1623,91 @@ async fn route_modified_otlp_sketches_to_precompute( } }; + // Stateful lineage is committed only after the payload + // has decoded successfully. Everything after this gate + // (snapshot replacement and SketchStore insertion) is + // synchronous/infallible, so an HTTP/gRPC success cannot + // acknowledge a sequence whose payload was never applied. + if let Some(frame) = frame_identity.as_ref() { + match ingest_state.observability.frame_lineage.observe(frame) { + Ok( + crate::precompute_engine::frame_lineage::FrameLineageDecision::Apply, + ) => { + if frame.kind + == control_plane::physical::compiler::SummaryFrameKind::Full + { + ingest_state + .sketch_index + .clear_summary_lineage_incomplete(sid, frame); + } + } + Ok( + crate::precompute_engine::frame_lineage::FrameLineageDecision::Duplicate, + ) => { + debug!( + plan_id = frame.plan_id, + plan_version = frame.plan_version, + materialization = frame.materialization.0, + producer = %frame.producer_id, + producer_epoch = %frame.producer_epoch, + sequence = frame.sequence, + "ignored duplicate summary frame" + ); + continue; + } + Err(error) => { + ingest_state + .sketch_index + .mark_summary_lineage_incomplete(sid, frame); + warn!( + plan_id = frame.plan_id, + plan_version = frame.plan_version, + materialization = frame.materialization.0, + producer = %frame.producer_id, + producer_epoch = %frame.producer_epoch, + sequence = frame.sequence, + %error, + "rejected summary frame lineage" + ); + return Err(error.to_string()); + } + } + } + + ingest_state.sketch_snapshots.insert( + series_key.clone(), + crate::precompute_engine::ingest_handler::SnapshotCacheEntry { + core: accumulator.clone_boxed_core(), + window_start: dp.start_time_unix_nano, + }, + ); + ingest_state.note_window_and_sweep(dp.start_time_unix_nano); + + use crate::storage_engines::sketch_db::index::{ + SketchEncoding, SketchSampleState, + }; + use std::collections::BTreeMap; + let label_values: BTreeMap = dp + .attrs + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect(); + let window: crate::storage_engines::sketch_db::index::epoch_columnar::TimestampRange = ( + dp.start_time_unix_nano / 1_000_000, + dp.time_unix_nano / 1_000_000, + ); + let encoding = + encoding_to_handle(dp.encoding).unwrap_or(SketchEncoding::ProtoFull); + ingest_state.sketch_index.append_sample( + sid, + label_values, + window, + SketchSampleState { + bytes: dp.sketch.clone(), + encoding, + }, + ); + // Collect the configs whose metric matches this DP. // Detection is independent of the legacy dual-write // (it only drives the routed/unconfigured accounting), @@ -1755,6 +1802,10 @@ async fn route_modified_otlp_sketches_to_precompute( flush_barrier_drops(ingest_state, &barrier_drops, "otlp-modified-proto"); + // No lineage-protected store mutation occurs after this point. Do not + // carry a synchronous mutex guard across the async worker-queue flush. + drop(lineage_batch_guard); + if !messages.is_empty() { if let Err(e) = ingest_state .router @@ -2098,7 +2149,7 @@ fn preflight_summary_frames( mut dp: ModifiedOtlpSketchDp, ingest_state: &IngestState, active: &crate::storage_engines::types::ActivePhysicalPlan, - ) -> Result<(), String> { + ) -> 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)?; @@ -2112,18 +2163,31 @@ fn preflight_summary_frames( .validate_frame(&frame) .map_err(|error| error.to_string())?; - if !dp.attrs.is_empty() { + let schema = active + .precompute_plan + .schemas + .iter() + .find(|schema| schema.materialization == frame.materialization) + .ok_or_else(|| format!("summary frame for {metric_name} has no active schema"))?; + if dp.attrs.is_empty() && !schema.group_by.is_empty() { + return Err(format!( + "summary frame for {metric_name} omits labels required by its grouped schema" + )); + } + let observed_series = if dp.attrs.is_empty() { + "".to_string() + } else { let pairs: Vec<(&str, &str)> = dp .attrs .iter() .map(|(key, value)| (key.as_str(), value.as_str())) .collect(); - let observed_series = crate::drivers::ingest::canonical_attrs_fingerprint(&pairs); - if observed_series != frame.series_fingerprint { - return Err(format!( - "summary frame for {metric_name} declares a series fingerprint different from its labels" - )); - } + crate::drivers::ingest::canonical_attrs_fingerprint(&pairs) + }; + if observed_series != frame.series_identity { + return Err(format!( + "summary frame for {metric_name} declares a series identity different from its labels" + )); } // A malformed full snapshot must not be discovered after an earlier @@ -2131,6 +2195,33 @@ fn preflight_summary_frames( if frame.kind == control_plane::physical::compiler::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 { + let series_key = format_series_key(canonical_name, &dp.attrs); + let (mut base, base_window_start) = ingest_state + .sketch_snapshots + .get(&series_key) + .map(|entry| (entry.core.clone_boxed_core(), entry.window_start)) + .or_else(|| { + empty_accumulator_for_delta_bootstrap( + dp.algorithm.clone(), + &dp.container_config, + dp.encoding, + ) + .map(|base| (base, dp.start_time_unix_nano)) + }) + .ok_or_else(|| { + format!("delta frame for {metric_name} has no reconstructable base") + })?; + if base_window_start != dp.start_time_unix_nano { + base.reset_to_empty(); + } + apply_modified_otlp_delta_bytes( + dp.algorithm.clone(), + dp.encoding, + &mut base, + &dp.sketch, + ) + .map_err(|error| format!("invalid delta frame for {metric_name}: {error}"))?; } // Attribute-elided retries can recover the policy from their known @@ -2162,9 +2253,10 @@ fn preflight_summary_frames( frame.materialization.0, observed.0 )); } - Ok(()) + Ok(frame) } + let mut frames = Vec::new(); for resource_metrics in &request.resource_metrics { let resource_attrs = resource_metrics .resource @@ -2187,7 +2279,7 @@ fn preflight_summary_frames( ($points:expr, $algorithm:expr, $config:expr) => {{ let config = $config; for point in &$points { - validate_one( + frames.push(validate_one( &metric.name, ModifiedOtlpSketchDp { algorithm: $algorithm, @@ -2201,7 +2293,7 @@ fn preflight_summary_frames( }, ingest_state, active, - )?; + )?); } }}; } @@ -2246,6 +2338,11 @@ fn preflight_summary_frames( } } } + ingest_state + .observability + .frame_lineage + .validate_batch(frames.iter()) + .map_err(|error| error.to_string())?; Ok(()) } @@ -2277,10 +2374,10 @@ fn take_summary_frame_identity( let backend_compat = required(attrs, "asap.frame.backend_compat")?; let materialization = asap_types::PolicyFingerprint(number(attrs, "asap.frame.materialization")?); + let series_identity = required(attrs, "asap.frame.series_identity")?; let schema_id = required(attrs, "asap.frame.schema_id")?; let producer_id = required(attrs, "asap.frame.producer_id")?; let producer_epoch = required(attrs, "asap.frame.producer_epoch")?; - let series_fingerprint = required(attrs, "asap.frame.series_fingerprint")?; let sequence = number(attrs, "asap.frame.sequence")?; let kind = match required(attrs, "asap.frame.kind")?.as_str() { "full" => SummaryFrameKind::Full, @@ -2306,10 +2403,10 @@ fn take_summary_frame_identity( plan_version, backend_compat, materialization, + series_identity, schema_id, producer_id, producer_epoch, - series_fingerprint, window_start_unix_nano, window_end_unix_nano, sequence, @@ -4599,10 +4696,13 @@ mod sid_bucketing_tests { "asap-query-backend.v1".into(), ), ("asap.frame.materialization".into(), "99".into()), + ( + "asap.frame.series_identity".into(), + "service=checkout,zone=a".into(), + ), ("asap.frame.schema_id".into(), "schema-99".into()), ("asap.frame.producer_id".into(), "edge-a".into()), ("asap.frame.producer_epoch".into(), "boot-7".into()), - ("asap.frame.series_fingerprint".into(), "service=api".into()), ("asap.frame.sequence".into(), "8".into()), ("asap.frame.kind".into(), "full".into()), ("asap.frame.encoding".into(), "sketchlib_protobuf_v1".into()), diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 25b0d5264..95f41a8a7 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -5301,6 +5301,8 @@ struct PhysicalPlanInstallRequest { backend_plan: Vec, query_plan: control_plane::query_plan::QueryPlan, storage_routing: Option, + #[serde(default)] + adaptation_evidence: Vec, } /// Validate and stage all backend views. Staging never changes query routing; @@ -5351,6 +5353,23 @@ async fn handle_post_physical_plan( ) .into_response(); } + let current = active_handle.snapshot(); + if current.transmission_plan.envelope.plan_id != 0 { + if let Err(error) = current.transmission_plan.authorize_successor( + &request.transmission_plan, + &request.adaptation_evidence, + unix_time_ms(), + ) { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({ + "status": "error", + "error": format!("runtime adaptation authorization error: {error}") + })), + ) + .into_response(); + } + } let new_config = crate::storage_engines::types::StreamingConfig::new(runtime_materializations); let new_plan = match control_plane::backend_plan::BackendPlan::decode(&request.backend_plan) { Ok(plan) => plan, diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 58e925ed9..cee6385c8 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -456,7 +456,7 @@ async fn main() -> Result<()> { frame_identity: control_plane::physical::compiler::FrameIdentityContract { identity_version: 1, sequence_scope: - control_plane::physical::compiler::SequenceScope::MaterializationWindowProducerEpoch, + control_plane::physical::compiler::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 new file mode 100644 index 000000000..a400dda30 --- /dev/null +++ b/data_plane/src/precompute_engine/frame_lineage.rs @@ -0,0 +1,366 @@ +//! Receiver-side ordering and checkpoint validation for summary frames. +//! +//! [`TransmissionPlan::validate_frame`] validates a frame's static shape +//! against the active physical plan. This module owns the stateful half of +//! 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 dashmap::mapref::entry::Entry; +use thiserror::Error; + +/// The result of accepting a frame into its lineage. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameLineageDecision { + /// This frame advances (or re-establishes) the lineage and may be applied. + Apply, + /// This exact frame was already accepted. Callers must acknowledge it but + /// must not apply or persist it a second time. + Duplicate, +} + +/// Stateful frame-contract failures. Every failure is recoverable by the +/// producer emitting a new full checkpoint for the same scoped lineage. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum FrameLineageError { + #[error("full frame is missing its checkpoint id")] + MissingCheckpoint, + #[error("delta frame is missing its base checkpoint id")] + MissingBaseCheckpoint, + #[error("delta has no established full checkpoint")] + MissingBase, + #[error("delta base checkpoint does not match the established checkpoint")] + BaseCheckpointMismatch, + #[error("frame sequence is stale or conflicts with an accepted frame")] + StaleOrConflictingSequence, + #[error("delta sequence gap: expected {expected}, received {received}")] + SequenceGap { expected: u64, received: u64 }, + #[error("lineage is incomplete and requires a new full checkpoint")] + Incomplete, +} + +/// Exact scope mandated by `SequenceScope::MaterializationSeriesProducerEpoch`. +/// `producer_id` is included because epochs are only unique within a producer. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct FrameLineageKey { + plan_id: u64, + plan_version: u64, + materialization: asap_types::PolicyFingerprint, + series_identity: String, + producer_id: String, + producer_epoch: String, +} + +impl From<&SummaryFrameIdentity> for FrameLineageKey { + fn from(frame: &SummaryFrameIdentity) -> Self { + Self { + plan_id: frame.plan_id, + plan_version: frame.plan_version, + materialization: frame.materialization, + series_identity: frame.series_identity.clone(), + producer_id: frame.producer_id.clone(), + producer_epoch: frame.producer_epoch.clone(), + } + } +} + +#[derive(Debug, Clone)] +struct FrameLineageState { + checkpoint_id: Option, + last_frame: SummaryFrameIdentity, + incomplete: bool, +} + +/// Concurrency-safe receiver state for frame checkpoint/sequence lineages. +/// +/// The occupied DashMap entry is held across each transition. Acceptance is +/// therefore linearizable for one lineage (two concurrent copies of sequence +/// N+1 cannot both be applied) without serializing unrelated producers. +#[derive(Debug, Default)] +pub struct FrameLineageTracker { + lineages: dashmap::DashMap, + batch_gate: std::sync::Mutex<()>, +} + +impl FrameLineageTracker { + /// Serialize validation and application of one OTLP frame batch. The + /// caller releases this guard after every frame has reached the store. + pub fn lock_batch(&self) -> std::sync::MutexGuard<'_, ()> { + self.batch_gate.lock().expect("frame lineage batch lock") + } + + /// Dry-run a complete batch against a snapshot of committed lineage. + /// This detects duplicates, gaps and checkpoint mismatches before the + /// caller applies the first payload in the request. + pub fn validate_batch<'a>( + &self, + frames: impl IntoIterator, + ) -> Result<(), FrameLineageError> { + let scratch = FrameLineageTracker::default(); + for entry in &self.lineages { + scratch + .lineages + .insert(entry.key().clone(), entry.value().clone()); + } + for frame in frames { + scratch.observe(frame)?; + } + Ok(()) + } + + /// Validate and atomically record one frame. + /// + /// Full frames establish (or replace) the checkpoint and clear an + /// incomplete lineage. Delta frames must reference that checkpoint and + /// advance the last accepted sequence by exactly one. An exact replay of + /// the last frame is idempotently classified as [`Duplicate`]. + pub fn observe( + &self, + frame: &SummaryFrameIdentity, + ) -> Result { + match frame.kind { + SummaryFrameKind::Full if frame.checkpoint_id.is_none() => { + return Err(FrameLineageError::MissingCheckpoint); + } + SummaryFrameKind::Delta if frame.base_checkpoint_id.is_none() => { + return Err(FrameLineageError::MissingBaseCheckpoint); + } + _ => {} + } + let key = FrameLineageKey::from(frame); + match self.lineages.entry(key) { + Entry::Vacant(entry) => match frame.kind { + SummaryFrameKind::Full => { + entry.insert(FrameLineageState { + checkpoint_id: frame.checkpoint_id.clone(), + last_frame: frame.clone(), + incomplete: false, + }); + Ok(FrameLineageDecision::Apply) + } + // A rejected frame never advances receiver state. In + // particular, a producer may recover this exact sequence by + // retransmitting it as a full checkpoint. + SummaryFrameKind::Delta => Err(FrameLineageError::MissingBase), + }, + Entry::Occupied(mut entry) => { + let state = entry.get_mut(); + if state.last_frame == *frame { + return Ok(FrameLineageDecision::Duplicate); + } + + match frame.kind { + // A full frame is the explicit recovery boundary. It may + // jump over missing delta sequences and resets the base. + SummaryFrameKind::Full => { + if frame.sequence <= state.last_frame.sequence { + return Err(FrameLineageError::StaleOrConflictingSequence); + } + state.checkpoint_id = frame.checkpoint_id.clone(); + state.last_frame = frame.clone(); + state.incomplete = false; + Ok(FrameLineageDecision::Apply) + } + SummaryFrameKind::Delta => { + if state.incomplete { + return Err(FrameLineageError::Incomplete); + } + if frame.base_checkpoint_id.as_ref() != state.checkpoint_id.as_ref() { + return Err(FrameLineageError::BaseCheckpointMismatch); + } + if frame.sequence <= state.last_frame.sequence { + return Err(FrameLineageError::StaleOrConflictingSequence); + } + let expected = state.last_frame.sequence.saturating_add(1); + if frame.sequence != expected { + state.incomplete = true; + return Err(FrameLineageError::SequenceGap { + expected, + received: frame.sequence, + }); + } + state.last_frame = frame.clone(); + Ok(FrameLineageDecision::Apply) + } + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use control_plane::physical::compiler::StateEncoding; + + fn frame(sequence: u64, kind: SummaryFrameKind) -> SummaryFrameIdentity { + SummaryFrameIdentity { + identity_version: 1, + plan_id: 7, + plan_version: 3, + backend_compat: "asap-query-backend.v1".into(), + materialization: asap_types::PolicyFingerprint(41), + series_identity: "service=checkout,zone=a".into(), + schema_id: "schema-41".into(), + producer_id: "edge-a".into(), + producer_epoch: "boot-9".into(), + window_start_unix_nano: 100, + window_end_unix_nano: 200, + sequence, + kind: kind.clone(), + encoding: StateEncoding::SketchlibProtobufV1, + checkpoint_id: (kind == SummaryFrameKind::Full).then(|| format!("cp-{sequence}")), + base_checkpoint_id: (kind == SummaryFrameKind::Delta).then(|| "cp-1".into()), + } + } + + #[test] + fn full_establishes_base_and_contiguous_delta_advances() { + let tracker = FrameLineageTracker::default(); + assert_eq!( + tracker.observe(&frame(1, SummaryFrameKind::Full)), + Ok(FrameLineageDecision::Apply) + ); + assert_eq!( + tracker.observe(&frame(2, SummaryFrameKind::Delta)), + Ok(FrameLineageDecision::Apply) + ); + } + + #[test] + fn exact_duplicate_is_idempotently_ignored() { + let tracker = FrameLineageTracker::default(); + let full = frame(1, SummaryFrameKind::Full); + let delta = frame(2, SummaryFrameKind::Delta); + tracker.observe(&full).unwrap(); + tracker.observe(&delta).unwrap(); + assert_eq!(tracker.observe(&delta), Ok(FrameLineageDecision::Duplicate)); + } + + #[test] + fn batch_validation_is_atomic_and_does_not_commit() { + let tracker = FrameLineageTracker::default(); + let full = frame(1, SummaryFrameKind::Full); + let delta = frame(2, SummaryFrameKind::Delta); + tracker + .validate_batch([&full, &delta]) + .expect("contiguous batch is valid"); + + assert_eq!( + tracker.observe(&delta), + Err(FrameLineageError::MissingBase), + "dry-run validation must not expose an un-applied checkpoint" + ); + + let gap = frame(3, SummaryFrameKind::Delta); + assert!(matches!( + tracker.validate_batch([&full, &gap]), + Err(FrameLineageError::SequenceGap { + expected: 2, + received: 3 + }) + )); + assert_eq!( + tracker.observe(&full), + Ok(FrameLineageDecision::Apply), + "a rejected batch must leave committed lineage unchanged" + ); + } + + #[test] + fn delta_requires_the_established_checkpoint() { + let tracker = FrameLineageTracker::default(); + assert_eq!( + tracker.observe(&frame(1, SummaryFrameKind::Delta)), + Err(FrameLineageError::MissingBase) + ); + assert_eq!( + tracker.observe(&frame(1, SummaryFrameKind::Full)), + Ok(FrameLineageDecision::Apply), + "a rejected leading delta must not consume its sequence" + ); + + let tracker = FrameLineageTracker::default(); + tracker.observe(&frame(1, SummaryFrameKind::Full)).unwrap(); + let mut wrong_base = frame(2, SummaryFrameKind::Delta); + wrong_base.base_checkpoint_id = Some("some-other-checkpoint".into()); + assert_eq!( + tracker.observe(&wrong_base), + Err(FrameLineageError::BaseCheckpointMismatch) + ); + } + + #[test] + fn checkpoint_identity_is_required_for_both_frame_kinds() { + let tracker = FrameLineageTracker::default(); + let mut full = frame(1, SummaryFrameKind::Full); + full.checkpoint_id = None; + assert_eq!( + tracker.observe(&full), + Err(FrameLineageError::MissingCheckpoint) + ); + + let mut delta = frame(2, SummaryFrameKind::Delta); + delta.base_checkpoint_id = None; + assert_eq!( + tracker.observe(&delta), + Err(FrameLineageError::MissingBaseCheckpoint) + ); + } + + #[test] + fn gap_marks_lineage_incomplete_until_next_full() { + let tracker = FrameLineageTracker::default(); + tracker.observe(&frame(1, SummaryFrameKind::Full)).unwrap(); + assert_eq!( + tracker.observe(&frame(3, SummaryFrameKind::Delta)), + Err(FrameLineageError::SequenceGap { + expected: 2, + received: 3 + }) + ); + assert_eq!( + tracker.observe(&frame(2, SummaryFrameKind::Delta)), + Err(FrameLineageError::Incomplete) + ); + + let recovery = frame(4, SummaryFrameKind::Full); + assert_eq!(tracker.observe(&recovery), Ok(FrameLineageDecision::Apply)); + let mut next = frame(5, SummaryFrameKind::Delta); + next.base_checkpoint_id = recovery.checkpoint_id.clone(); + assert_eq!(tracker.observe(&next), Ok(FrameLineageDecision::Apply)); + } + + #[test] + fn plan_series_and_epoch_are_independent_but_sequence_crosses_windows() { + let tracker = FrameLineageTracker::default(); + let first = frame(1, SummaryFrameKind::Full); + tracker.observe(&first).unwrap(); + + let mut next_plan = first.clone(); + next_plan.plan_version += 1; + assert_eq!(tracker.observe(&next_plan), Ok(FrameLineageDecision::Apply)); + + let mut next_series = first.clone(); + next_series.series_identity = "service=checkout,zone=b".into(); + assert_eq!( + tracker.observe(&next_series), + Ok(FrameLineageDecision::Apply) + ); + + let mut next_epoch = first.clone(); + next_epoch.producer_epoch = "boot-10".into(); + assert_eq!( + tracker.observe(&next_epoch), + Ok(FrameLineageDecision::Apply) + ); + + let mut next_window = frame(2, SummaryFrameKind::Delta); + next_window.window_start_unix_nano = 200; + next_window.window_end_unix_nano = 300; + assert_eq!( + tracker.observe(&next_window), + Ok(FrameLineageDecision::Apply) + ); + } +} diff --git a/data_plane/src/precompute_engine/ingest_handler.rs b/data_plane/src/precompute_engine/ingest_handler.rs index e5eacc1d0..b0fc300a5 100644 --- a/data_plane/src/precompute_engine/ingest_handler.rs +++ b/data_plane/src/precompute_engine/ingest_handler.rs @@ -45,6 +45,10 @@ pub struct IngestObservability { /// series, used as the eviction sweep's reference point. Advanced /// monotonically on each cached base insert. pub snapshot_newest_window_start: AtomicU64, + /// Stateful checkpoint and sequence validation for physical-plan summary + /// frames. Kept with the ingest-wide shared state so concurrent OTLP + /// requests observe one linearizable lineage per producer/window. + pub frame_lineage: super::frame_lineage::FrameLineageTracker, } impl IngestObservability { @@ -69,6 +73,7 @@ impl IngestObservability { dropped_policy_miss: AtomicU64::new(0), snapshot_max_window_lag_nanos: AtomicU64::new(lag), snapshot_newest_window_start: AtomicU64::new(0), + frame_lineage: super::frame_lineage::FrameLineageTracker::default(), } } } diff --git a/data_plane/src/precompute_engine/mod.rs b/data_plane/src/precompute_engine/mod.rs index ce74b89bb..afbabf030 100644 --- a/data_plane/src/precompute_engine/mod.rs +++ b/data_plane/src/precompute_engine/mod.rs @@ -1,6 +1,7 @@ pub mod accumulator_factory; pub mod config; mod engine; +pub mod frame_lineage; pub mod ingest_handler; mod metrics; pub mod operators; 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 e06ddeb33..533658730 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -19,7 +19,7 @@ //! See design doc §4.6 ("OTLP metadata model + backend store layout") at //! `docs/design_docs/series-identity.md`. -use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::sync::{Arc, RwLock}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; @@ -275,6 +275,29 @@ impl SketchInstanceMetadata { /// `SketchStore` can host both sketches and precomputes. type SidStore = Arc, AggPayload>>>; +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct IncompleteSummaryLineage { + plan_id: u64, + plan_version: u64, + producer_id: String, + producer_epoch: String, + window_start_unix_ms: u64, + window_end_unix_ms: u64, +} + +impl From<&control_plane::physical::compiler::SummaryFrameIdentity> for IncompleteSummaryLineage { + fn from(frame: &control_plane::physical::compiler::SummaryFrameIdentity) -> Self { + Self { + plan_id: frame.plan_id, + plan_version: frame.plan_version, + producer_id: frame.producer_id.clone(), + producer_epoch: frame.producer_epoch.clone(), + window_start_unix_ms: frame.window_start_unix_nano / 1_000_000, + window_end_unix_ms: frame.window_end_unix_nano / 1_000_000, + } + } +} + /// Two-level sketch index. Replaces the legacy `aggregation_id`-keyed /// SimpleStore lookup once Phase 5 wiring lands at the streaming engine /// ingest path and the query path. @@ -302,6 +325,10 @@ pub struct SketchStore { /// key) for ghost sids — query path detects this and falls through /// to Thanos archive. series: DashMap, + /// Receiver-observed delta gaps. Query reads overlapping an incomplete + /// lineage fail closed to the exact tier until a recovery full checkpoint + /// for that exact producer/window lineage is accepted. + incomplete_summary_lineages: DashMap>, /// Reverse index: `policy_fp → {sids}`. Lets the query path resolve /// "which sids belong to this policy?" in O(1) without walking /// `instances`. Maintained by [`Self::register`] / @@ -594,6 +621,18 @@ impl SketchStore { start_unix_ms: u64, end_unix_ms: u64, ) -> Vec { + if self + .incomplete_summary_lineages + .get(&sid) + .is_some_and(|lineages| { + lineages.iter().any(|lineage| { + lineage.window_start_unix_ms < end_unix_ms + && lineage.window_end_unix_ms > start_unix_ms + }) + }) + { + return Vec::new(); + } // Result is keyed by the resolved label MAP so the in-memory tier // (its own intern space) and the durable disk tier (independent // intern space) union by label identity, not `LabelValuesId`. @@ -1740,9 +1779,34 @@ impl SketchStore { }; if removed.is_some() { self.series.remove(&sid); + self.incomplete_summary_lineages.remove(&sid); } removed } + + /// Mark a producer/window delta lineage unsafe for warm reads. + pub fn mark_summary_lineage_incomplete( + &self, + sid: u64, + frame: &control_plane::physical::compiler::SummaryFrameIdentity, + ) { + self.incomplete_summary_lineages + .entry(sid) + .or_default() + .insert(frame.into()); + } + + /// A full checkpoint repairs only its exact producer/window lineage. + pub fn clear_summary_lineage_incomplete( + &self, + sid: u64, + frame: &control_plane::physical::compiler::SummaryFrameIdentity, + ) { + let key = IncompleteSummaryLineage::from(frame); + if let Some(mut lineages) = self.incomplete_summary_lineages.get_mut(&sid) { + lineages.remove(&key); + } + } } impl SketchStore { @@ -2388,6 +2452,49 @@ mod tests { assert_eq!(s_b.samples.len(), 2); } + #[test] + fn incomplete_delta_window_fails_closed_until_matching_full_checkpoint() { + use control_plane::physical::compiler::{ + StateEncoding, SummaryFrameIdentity, SummaryFrameKind, + }; + + let idx = SketchStore::new(); + idx.register(meta(12)); + idx.append_sample(12, BTreeMap::new(), (0, 10), sample(0)); + idx.append_sample(12, BTreeMap::new(), (1000, 1010), sample(1)); + let frame = SummaryFrameIdentity { + identity_version: 1, + plan_id: 7, + plan_version: 2, + backend_compat: "asap-query-backend.v1".into(), + materialization: PolicyFingerprint(41), + series_identity: "service=checkout,zone=a".into(), + schema_id: "schema-41".into(), + producer_id: "edge-a".into(), + producer_epoch: "boot-1".into(), + window_start_unix_nano: 1_000_000_000, + window_end_unix_nano: 1_010_000_000, + sequence: 3, + kind: SummaryFrameKind::Delta, + encoding: StateEncoding::SketchlibProtobufV1, + checkpoint_id: None, + base_checkpoint_id: Some("cp-1".into()), + }; + idx.mark_summary_lineage_incomplete(12, &frame); + assert!(idx.query_range(12, 1000, 1010).is_empty()); + assert_eq!(idx.query_range(12, 0, 999).len(), 1); + + let recovery = SummaryFrameIdentity { + kind: SummaryFrameKind::Full, + checkpoint_id: Some("cp-4".into()), + base_checkpoint_id: None, + sequence: 4, + ..frame + }; + idx.clear_summary_lineage_incomplete(12, &recovery); + assert_eq!(idx.query_range(12, 1000, 1010).len(), 1); + } + #[test] fn range_query_uses_overlap_not_containment() { // The sketch read path uses HALF-OPEN OVERLAP, not containment: 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 2b0151528..9ceab4638 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -650,7 +650,7 @@ mod tests { }, frame_identity: control_plane::physical::compiler::FrameIdentityContract { identity_version: 1, - sequence_scope: control_plane::physical::compiler::SequenceScope::MaterializationWindowProducerEpoch, + sequence_scope: control_plane::physical::compiler::SequenceScope::MaterializationSeriesProducerEpoch, require_checkpoint_for_full: true, require_base_checkpoint_for_delta: true, }, diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index 4f980c550..9e0182477 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -153,6 +153,22 @@ materializations are accepted; table sources and unexecutable selected families return `CompileError`. Versioned stage/activate/drain/retire lifecycle is part of publication; richer topology placement remains follow-up work. +`TransmissionRule.runtime_policy` is the single physical contract for sampling, +delta gating, and GOS. Sampling is typed by its implemented estimator; GOS is a +CountSketch delta policy, not a summary family. Unsupported family/policy +combinations fail compilation. Delta rules declare a periodic full-checkpoint +cadence and accept both deltas and their recovery full frames. + +Runtime feedback cannot mutate an active rule. `authorize_successor` accepts +only the next `plan_version`, requires fresh evidence for the exact +plan/materialization/schema/producer tuple, and bounds each permitted knob by +the active rule's guardrails. Family, parameters, grouping, window, schema, +encoding, destination, mode, and checkpoint semantics stay fixed; changing +them requires ordinary recompilation and staged activation. +The atomic physical-plan publication carries the evidence records alongside +the successor; the backend performs this authorization before staging it, so a +caller cannot bypass guardrails by posting a changed rule directly. + Output definitions: | Output | Definition | @@ -280,15 +296,23 @@ matching raw transmission and ingest/archive declarations. Until modified OTLP has dedicated identity fields, Collector attaches reserved data-point attributes under `asap.frame.*`: identity version, plan ID/version, backend compatibility, materialization and schema IDs, producer ID/epoch, -canonical series fingerprint, sequence, full/delta kind, encoding, and +canonical series identity, sequence, full/delta kind, encoding, and checkpoint/base IDs. Window start/end remain the typed data-point timestamps. The backend removes reserved attributes before building the series label key and rejects the complete request before -writing any frame when one identity, schema, encoding, materialization, or full +writing any frame when one identity, schema, encoding, materialization, or payload does not match the active TransmissionPlan. HTTP 2xx / gRPC OK is the delivery acknowledgement. Retrying the same full frame is idempotent because the identity selects the same SID, label set, and window replacement; no second -application-level ACK or transport WAL is part of this contract. +application-level ACK or transport WAL is part of this contract. Receiver +lineage is scoped by plan/version, materialization, concrete series, producer, +and producer epoch; sequence/checkpoint continuity crosses logical windows, +while every frame still identifies its own window. Grouped frames retain labels; +the singleton ungrouped series uses `` as its identity. Exact retries +are ignored idempotently; a missing base or sequence gap leaves the lineage +incomplete until a newer full checkpoint arrives. Lineage receipts are not a +second durable transport log: after backend restart the receiver safely rejects +deltas until the producer supplies a new full checkpoint. The compiler error must identify an unsupported capability, invalid placement, window incompatibility, identity conflict, or invalid selected guarantee. It diff --git a/docs/developer_docs/control-plane/plan-publication.md b/docs/developer_docs/control-plane/plan-publication.md index 036209930..a5cda3fef 100644 --- a/docs/developer_docs/control-plane/plan-publication.md +++ b/docs/developer_docs/control-plane/plan-publication.md @@ -10,7 +10,9 @@ typed `BackendPlan` plus one target-specific `CollectorPlan` per Collector. - `queries`: query ID, PromQL, metric, window seconds, grouping labels, a typed `AccuracyTarget`, and lifecycle evidence (evaluation interval, - ingestion rate/freshness, optimization horizon, and primitive state costs); + ingestion rate/freshness, optimization horizon, and primitive state costs), + plus executor-feasible `window_implementations` with versioned, + workload-scoped physical cost evidence; - `collector_ids`: the required OpAMP agent IDs; - `capability_snapshot_id` and the exact `planner_revision`; - `plan_version`, activation/optional expiry timestamps, and the exact backend @@ -24,11 +26,15 @@ successful only after every target has applied the exact generated `(plan_id, plan_version)` and the backend activates that generation. The physical compiler passes normalized recurrence, data-arrival evidence, -runtime capabilities, and costs to ASAPPlanner's summary-maintenance lifecycle -planner. The selected guarantee is copied into each Collector materialization. -For the current tumbling-window Collector runtime the executable commitment is +runtime capabilities, and complete implementation costs to ASAPPlanner's +summary-maintenance lifecycle and abstract-window selection. It retains the +concrete implementation identity corresponding to Planner's selected +framework and copies both decisions into each Collector materialization. +Missing, stale, or incomplete implementation evidence fails closed. For the +current tumbling-window Collector runtime the executable commitment is `continuously_maintained / incremental / per_update / summary_state`; missing -or stale lifecycle evidence and any unexecutable lifecycle fail closed. +or stale lifecycle evidence, a non-tumbling selected framework, mismatched pane +width, and any unexecutable lifecycle fail closed. ## Installation order and failure semantics