fix(data_plane): stop signature reconcile from retiring sketch sids (warm quantile/cardinality miss)

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<u64> = 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]

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<String> = 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();