feat(metrics): expose §6.3 barrier drops as Prometheus counter

asap-query-engine/src/precompute_engine/ingest_handler.rs

@@ -182,6 +182,14 @@ pub(crate) async fn route_decoded_samples(
         state
             .samples_blocked_by_schema_barrier
             .fetch_add(total_dropped, std::sync::atomic::Ordering::Relaxed);
+        // Also bump the Prometheus counter (per-agg label) so the
+        // drop rate is scrapable from /metrics without enabling debug
+        // logs in production.
+        for (agg_id, count) in &dropped_by_barrier {
+            crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER
+                .with_label_values(&[&agg_id.to_string()])
+                .inc_by(*count as f64);
+        }
         debug!(
             total_dropped,
             by_agg_id = ?dropped_by_barrier,
@@ -409,4 +417,40 @@ mod tests {
         drop(state);
         let _ = drain.await;
     }
+
+    /// Prometheus CounterVec is process-global via `lazy_static`, so
+    /// we key on a unique agg_id (9001) to get a fresh baseline that
+    /// no other test in the process has touched.
+    #[tokio::test]
+    async fn barrier_prom_counter_increments_per_agg_label() {
+        let (state, drain) = setup_state(9001, "metric_prom_test").await;
+        let label = "9001";
+        let baseline = crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER
+            .with_label_values(&[label])
+            .get();
+
+        assert!(state.schemas.force_expire(9001).is_some());
+
+        let _ = route_decoded_samples(
+            &state,
+            vec![
+                sample("metric_prom_test", 100, 1.0),
+                sample("metric_prom_test", 200, 2.0),
+                sample("metric_prom_test", 300, 3.0),
+                sample("metric_prom_test", 400, 4.0),
+            ],
+            std::time::Instant::now(),
+        )
+        .await;
+
+        let after = crate::stores::sketch_db::metrics::SAMPLES_BLOCKED_BY_SCHEMA_BARRIER
+            .with_label_values(&[label])
+            .get();
+        assert!(
+            (after - baseline - 4.0).abs() < f64::EPSILON,
+            "prom counter for agg_id={label} should have advanced by 4; baseline={baseline}, after={after}"
+        );
+        drop(state);
+        let _ = drain.await;
+    }
 }

asap-query-engine/src/stores/sketch_db/metrics.rs

@@ -0,0 +1,26 @@
+//! Prometheus metrics exposed on `/metrics` for the sketch DB.
+//!
+//! Scope: counters / gauges that turn otherwise-silent ingest or
+//! eviction behaviour into graphable signals. Registered via
+//! `lazy_static` + `prometheus::register_counter_vec!`, so a single
+//! `prometheus::gather()` at the HTTP handler picks them up
+//! automatically.
+
+use lazy_static::lazy_static;
+use prometheus::{register_counter_vec, CounterVec};
+
+lazy_static! {
+    /// §6.3 write-side barrier drops, per `agg_id`. Incremented when
+    /// the ingest path matches a sample's metric to an agg config
+    /// whose `AggSchema` is Retired or Expired, so the sample is
+    /// dropped instead of being routed to a worker. A non-zero rate
+    /// with no Retired/Expired schemas in the registry means the
+    /// reconcile / hot-reload path is lagging — silent drop becomes
+    /// a graphable alert condition.
+    pub static ref SAMPLES_BLOCKED_BY_SCHEMA_BARRIER: CounterVec = register_counter_vec!(
+        "queryengine_ingest_samples_blocked_by_schema_barrier_total",
+        "Ingest samples dropped by the §6.3 write-side schema barrier, keyed by agg_id",
+        &["agg_id"]
+    )
+    .unwrap();
+}

asap-query-engine/src/stores/sketch_db/mod.rs

@@ -34,6 +34,7 @@ pub mod backfill_processor;
 pub mod backfill_service;
 pub mod backfill_window_builder;
 pub mod backfill_worker;
+pub mod metrics;
 pub mod prometheus_reader;
 pub mod raw_sample_reader;
 pub mod schema;