diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 8d5a6c61..b0553685 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -1256,24 +1256,27 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu let required: crate::storage_engines::sketch_db::index::Capability = candidate.required_capability.clone(); let mut hit_sids: Vec = Vec::with_capacity(sids.len()); + // Track whether the candidate's sid set contained ONLY + // Ghost/Unknown sids with no usable Hit. A single + // metric/group-by selector legitimately resolves to a MIX + // of sids: freshly-minted Active sids carrying live sketch + // state (`Hit`) alongside retired-then-evicted or + // merged-away identities that no longer hold data + // (`Ghost`), plus stale sender-cache sids (`Unknown`). The + // earlier behavior aborted the whole query to CapabilityMiss + // on the FIRST Ghost/Unknown encountered — which, with the + // sid set iterated in ascending-u64 order, meant an older + // dataless sid masked the newer Active sketch sids that + // could answer. Skip non-Hit sids instead; only fail over + // to the archive when no Hit sid satisfies the capability + // (handled by the `hit_sids.is_empty()` check below, which + // preserves the all-ghost → CapabilityMiss contract). for sid in &sids { match idx.classify(*sid) { crate::storage_engines::sketch_db::index::SidLookup::Hit => {} crate::storage_engines::sketch_db::index::SidLookup::Ghost | crate::storage_engines::sketch_db::index::SidLookup::Unknown => { - let req = Self::requirements_from_candidate(candidate); - crate::drivers::control_plane_client::spawn_capability_miss_notify( - &self.control_plane_client, - &req, - ); - return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), - format!( - "SketchStore ghost/unknown sid {sid} for metric \ - `{}` — failing over to archive", - candidate.metric_name - ), - )); + continue; } } let meta = match idx.instance(*sid) { @@ -3385,6 +3388,66 @@ mod outer_agg_integration_tests { assert!(z1 > z0, "z1 ({z1}) > z0 ({z0}) — per-zone identity preserved"); } + /// Regression: a candidate's sid set legitimately contains a MIX of + /// `Ghost` (retired-then-evicted or merged-away, no data) and `Hit` + /// (Active, carrying live sketch state) sids under the same metric. + /// This is the exact production shape behind the warm-quantile miss: + /// the metric's `instances_matching` walk returns the older retired + /// sketch sids (now dataless ⇒ Ghost) alongside the freshly-minted + /// Active sketch sids. Iterating ascending-u64, the older Ghost sid + /// was hit first and aborted the WHOLE query to CapabilityMiss before + /// the Active sid could answer. After the fix, non-Hit sids are + /// skipped and the query resolves against the Active sid. + #[tokio::test] + async fn ghost_sid_does_not_mask_active_hit_sid_for_quantile() { + let idx = Arc::new(SketchStore::new()); + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let w_start = now_ms.saturating_sub(60_000); + let w_end = now_ms.saturating_sub(30_000); + + // Ghost sid (lower number ⇒ iterated first): registered metadata, + // never appended any sample state. `classify` → Ghost. + idx.register(dd_meta_for(1, "http_latency_ms", &["zone"])); + + // Active Hit sid (higher number): carries a real DDSketch window. + idx.register(dd_meta_for(2, "http_latency_ms", &["zone"])); + idx.append_sample( + 2, + BTreeMap::from([("zone".to_string(), "z0".to_string())]), + (w_start, w_end), + SketchSampleState { + bytes: dd_sketch_with_values(&[1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]), + encoding: SketchEncoding::MsgpackFull, + }, + ); + + let engine = build_engine_with_index(idx); + let result = engine + .execute("quantile_over_time(0.99, http_latency_ms[5m])") + .await + .expect( + "a dataless Ghost sid must not abort the query when an \ + Active Hit sid under the same metric can answer it", + ); + let vector = match result { + QueryResult::Vector(v) => v, + other => panic!("expected Vector, got {other:?}"), + }; + assert_eq!( + vector.values.len(), + 1, + "the single Active sid answers; the Ghost is skipped" + ); + assert!( + vector.values[0].value > 0.0, + "p99 of [1..=10] is a positive quantile, got {}", + vector.values[0].value + ); + } + /// `avg by (zone) (quantile_over_time(0.99, m[5m]))` — same shape, /// avg fold instead of max. Identity case ⇒ same per-zone values. #[tokio::test] diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs index 2e363fb2..4a41834d 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs @@ -110,6 +110,30 @@ pub fn reconcile_from_streaming_config( if !matches!(meta.status(), AggStatus::Active) { return; } + // `live_signatures` is built exclusively from + // `signature_from_agg_config`, which canonicalizes every + // streaming-config `AggregationConfig` to an `AggKind::ExactAgg` + // signature (`P`-prefixed). An `AggKind::Sketch` sid (OTLP + // modified-sketch ingest path: KLL / HLL / DDSketch / CMS / + // CountSketch) always produces an `S`-prefixed signature, so it + // can NEVER be present in `live_signatures` — meaning this scan + // would unconditionally orphan and retire EVERY sketch-backed sid + // on the first config reconcile. That is exactly the opposite of + // the documented intent (see this module's header: "sketch sids + // never compare equal so they're never retired by this path"). + // + // Retiring a sketch sid bars further ingest (the §6.3 write + // barrier rejects writes to Retired/Expired sids), then eviction + // drops its per-window state; the query path then classifies it + // as `Ghost`, and the analyzer-driven dispatch short-circuits the + // whole quantile / cardinality query to CapabilityMiss before it + // can reach a freshly-minted Active sketch sid carrying live data. + // Sketch-sid lifecycle is driven by the control plane's eviction + // RPC, NOT by this precompute-shaped signature reconcile, so skip + // them here. + if matches!(meta.agg_kind, AggKind::Sketch { .. }) { + return; + } scratch.clear(); signature_into( &meta.metric_name, @@ -387,6 +411,90 @@ mod tests { assert_eq!(summary.retired, vec![1]); } + fn meta_sketch( + sid: u64, + metric: &str, + kind: crate::storage_engines::sketch_db::data::SketchKindHandle, + config: crate::storage_engines::sketch_db::data::SketchConfig, + group_by: Vec<&str>, + ) -> SketchInstanceMetadata { + let group_by_keys: BTreeSet = group_by.into_iter().map(|s| s.to_string()).collect(); + SketchInstanceMetadata { + sid, + metric_name: metric.to_string(), + group_by_keys, + capability: None, + agg_kind: AggKind::Sketch { + kind, + config, + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + } + } + + // Regression: the precompute-shaped signature reconcile must NOT + // retire `AggKind::Sketch` sids. `build_live_signature_set` only ever + // emits `ExactAgg`-shaped (`P`-prefixed) signatures from the streaming + // config, so a sketch sid's `S`-prefixed signature can never be in the + // live set. Before the fix this scan unconditionally orphaned every + // OTLP-ingested KLL / HLL / DDSketch sid on the first reconcile, which + // barred ingest, let eviction drop their state, and turned them into + // query-time `Ghost`s that short-circuited warm quantile / cardinality + // queries to CapabilityMiss. Sketch-sid lifecycle is the control + // plane's eviction RPC's job, not this path's. + #[test] + fn sketch_sids_are_never_retired_by_signature_reconcile() { + use crate::storage_engines::sketch_db::data::{SketchConfig, SketchKindHandle}; + let store = SketchStore::new(); + store.register(meta_sketch( + 1, + "http_requests_total_latency_ms", + SketchKindHandle::Kll, + SketchConfig::Kll { k: 200 }, + vec!["node", "pod", "zone"], + )); + store.register(meta_sketch( + 2, + "unique_users_per_min", + SketchKindHandle::Hll, + SketchConfig::Hll { precision: 12 }, + vec!["zone"], + )); + // A precompute (ExactAgg) sid with no live config entry SHOULD + // still retire — the fix is scoped to sketch sids only. + store.register(meta(3, "cpu", AggregationType::Sum, vec!["host"])); + + // Live config carries an unrelated ExactAgg policy; none of the + // sketch sids' signatures can match it. + let cfg = streaming(vec![agg_config("mem", AggregationType::Sum, vec!["host"])]); + let summary = reconcile_from_streaming_config(&store, &cfg, Duration::from_secs(60)); + + // Only the orphaned precompute sid retires; both sketch sids stay + // Active so the warm query path can keep resolving them. + assert_eq!(summary.retired, vec![3], "only the ExactAgg orphan retires"); + assert!( + matches!(store.classify(1), crate::storage_engines::sketch_db::index::SidLookup::Ghost), + "sketch sid 1 stays registered + Active (Ghost only because no data appended in-test)" + ); + let inst1 = store.instance(1).expect("sketch sid 1 still registered"); + assert_eq!( + inst1.status(), + AggStatus::Active, + "KLL sketch sid must remain Active after reconcile" + ); + let inst2 = store.instance(2).expect("sketch sid 2 still registered"); + assert_eq!( + inst2.status(), + AggStatus::Active, + "HLL sketch sid must remain Active after reconcile" + ); + } + #[test] fn gate_skips_reconcile_on_unchanged_config_arc() { let store = SketchStore::new();