refactor(http): drop Arc<dyn Store> from HttpServer + runtime-info adapter

data_plane/src/drivers/query/adapters/prometheus_http.rs

@@ -360,18 +360,14 @@ impl HttpProtocolAdapter for PrometheusHttpAdapter {
 
     async fn handle_runtime_info(
         &self,
-        store: Arc<dyn crate::stores::Store>,
+        sketch_index: Arc<crate::stores::sketch_db::index::SketchIndex>,
     ) -> Result<Json<Value>, StatusCode> {
         debug!("Handling runtime info request in Prometheus adapter");
 
-        // Get earliest timestamp per aggregation ID from store
-        let earliest_timestamps = match store.get_earliest_timestamp_per_aggregation_id() {
-            Ok(timestamps) => timestamps,
-            Err(e) => {
-                error!("Error getting earliest timestamps: {}", e);
-                HashMap::new()
-            }
-        };
+        // M2.3.6g — earliest timestamps now come from SketchIndex's
+        // per-sid `first_seen_unix_ms` metadata. Wire field renamed
+        // accordingly below.
+        let earliest_timestamps = sketch_index.earliest_timestamps_per_sid();
 
         // Get runtime info from fallback if available
         let mut runtime_data = if let Some(fallback) = &self.config.fallback {
@@ -390,13 +386,12 @@ impl HttpProtocolAdapter for PrometheusHttpAdapter {
         // Merge local data with fallback data
         if let Some(data_obj) = runtime_data.as_object_mut() {
             data_obj.insert(
-                "earliest_timestamp_per_aggregation_id".to_string(),
+                "earliest_timestamp_per_sid".to_string(),
                 serde_json::to_value(earliest_timestamps).unwrap_or(json!({})),
             );
         } else {
-            // If runtime_data is not an object, just create a new one with local data
             runtime_data = json!({
-                "earliest_timestamp_per_aggregation_id": earliest_timestamps
+                "earliest_timestamp_per_sid": earliest_timestamps
             });
         }
 

data_plane/src/drivers/query/adapters/traits.rs

@@ -147,20 +147,23 @@ pub trait HttpProtocolAdapter: QueryRequestAdapter + QueryResponseAdapter + Send
 
     /// Handle runtime info request
     ///
-    /// The adapter can query the store for internal metrics and
+    /// The adapter can query the SketchIndex for internal metrics and
     /// optionally forward to fallback backend for additional info.
+    /// M2.3.6g — switched from `Arc<dyn Store>` to
+    /// `Arc<SketchIndex>` now that SketchIndex is the only data
+    /// backend.
     async fn handle_runtime_info(
         &self,
-        store: std::sync::Arc<dyn crate::stores::Store>,
+        sketch_index: std::sync::Arc<crate::stores::sketch_db::index::SketchIndex>,
     ) -> Result<Json<Value>, StatusCode>;
 
     async fn handle_runtime_info_with_headers(
         &self,
-        store: std::sync::Arc<dyn crate::stores::Store>,
+        sketch_index: std::sync::Arc<crate::stores::sketch_db::index::SketchIndex>,
         headers: HashMap<String, String>,
     ) -> Result<Json<Value>, StatusCode> {
         // Default implementation ignores headers and calls the old method
         let _ = headers;
-        self.handle_runtime_info(store).await
+        self.handle_runtime_info(sketch_index).await
     }
 }

data_plane/src/drivers/query/servers/http.rs

@@ -156,7 +156,8 @@ pub struct HttpServer {
     /// `KeyByLabelNames` Prometheus needs to populate the `metric`
     /// map. See `docs/design-gorilla-s3-cold-engine.md` §8.
     query_router: Arc<EngineRouter>,
-    store: Arc<dyn Store>,
+    /// M2.3.6g — SketchIndex replaces `Arc<dyn Store>`.
+    sketch_index: Arc<crate::stores::sketch_db::index::SketchIndex>,
     /// Hot-reloadable `StreamingConfig` source. `None` when hot-reload
     /// is not wired up by the caller (unit tests, legacy binaries).
     hot_reload_config: Option<crate::stores::types::HotReloadStreamingConfig>,
@@ -220,7 +221,10 @@ struct AppState {
     query_engine: Arc<ASAPQueryEngine>,
     /// See [`HttpServer::query_router`].
     query_router: Arc<EngineRouter>,
-    store: Arc<dyn Store>,
+    /// Phase 5 M2.3.6g — SketchIndex replaces `Arc<dyn Store>` as the
+    /// only data backend HTTP-side endpoints consult. Today the only
+    /// consumer is the runtime-info handler.
+    sketch_index: Arc<crate::stores::sketch_db::index::SketchIndex>,
     adapter: Arc<dyn HttpProtocolAdapter>,
     fallback: Option<Arc<dyn crate::drivers::query::fallback::FallbackClient>>,
     hot_reload_config: Option<crate::stores::types::HotReloadStreamingConfig>,
@@ -246,7 +250,7 @@ impl HttpServer {
     pub fn new(
         config: HttpServerConfig,
         query_engine: Arc<ASAPQueryEngine>,
-        store: Arc<dyn Store>,
+        sketch_index: Arc<crate::stores::sketch_db::index::SketchIndex>,
     ) -> Self {
         // Bootstrap the capability router with `ASAPQueryEngine`
         // registered under its canonical query-engine id.
@@ -257,7 +261,7 @@ impl HttpServer {
             config,
             query_engine,
             query_router,
-            store,
+            sketch_index,
             hot_reload_config: None,
             backend_storage_routing: None,
             schemas: None,
@@ -427,7 +431,7 @@ impl HttpServer {
             config: self.config.clone(),
             query_engine: self.query_engine,
             query_router: self.query_router,
-            store: self.store,
+            sketch_index: self.sketch_index,
             adapter: adapter.clone(),
             fallback: self.config.adapter_config.fallback.clone(),
             hot_reload_config: self.hot_reload_config.clone(),
@@ -518,7 +522,7 @@ impl HttpServer {
             config: self.config.clone(),
             query_engine: self.query_engine.clone(),
             query_router: self.query_router.clone(),
-            store: self.store.clone(),
+            sketch_index: self.sketch_index.clone(),
             adapter: adapter.clone(),
             fallback: self.config.adapter_config.fallback.clone(),
             hot_reload_config: self.hot_reload_config.clone(),
@@ -1541,7 +1545,7 @@ async fn handle_runtime_info(
     // Delegate to adapter for protocol-specific handling
     state
         .adapter
-        .handle_runtime_info_with_headers(state.store.clone(), forwarding_headers)
+        .handle_runtime_info_with_headers(state.sketch_index.clone(), forwarding_headers)
         .await
 }
 
@@ -1798,7 +1802,7 @@ mod tests {
             15000,
         ));
 
-        let mut server = HttpServer::new(config, query_engine, store);
+        let mut server = { let _store = store; let idx = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); HttpServer::new(config, query_engine, idx) };
         if let Some(handle) = hot_reload {
             server = server.with_hot_reload_config(handle);
         }
@@ -2045,7 +2049,7 @@ aggregations:
             streaming_config.clone(),
             15000,
         ));
-        let server = HttpServer::new(config, query_engine, store)
+        let server = { let _store = store; let idx = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); HttpServer::new(config, query_engine, idx) }
             .with_hot_reload_config(hot_reload)
             .with_schemas(schemas);
         server
@@ -2550,7 +2554,7 @@ aggregations:
             let sc = StreamingConfig::new(map);
             Arc::new(crate::stores::sketch_db::SchemaRegistry::from_streaming_config(&sc))
         };
-        let server = HttpServer::new(config, query_engine, store)
+        let server = { let _store = store; let idx = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); HttpServer::new(config, query_engine, idx) }
             .with_backfill_registry(registry)
             .with_schemas(schemas);
         server
@@ -2901,7 +2905,7 @@ aggregations:
             15000,
         ));
         let mut server =
-            HttpServer::new(config, query_engine, store).with_hot_reload_config(hot_reload);
+            { let _store = store; let idx = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); HttpServer::new(config, query_engine, idx) }.with_hot_reload_config(hot_reload);
         for engine in extra_engines {
             server = server.with_query_engine(engine);
         }
@@ -2946,7 +2950,7 @@ aggregations:
             streaming_arc,
             15000,
         ));
-        let mut server = HttpServer::new(config, query_engine, store)
+        let mut server = { let _store = store; let idx = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); HttpServer::new(config, query_engine, idx) }
             .with_hot_reload_config(hot_reload)
             .with_backend_storage_routing(Arc::new(routing));
         for engine in extra_engines {
@@ -3694,7 +3698,7 @@ aggregations:
             15000,
         ));
         let routing_handle = HotReloadBackendStorageRouting::empty();
-        let server = HttpServer::new(config, query_engine, store)
+        let server = { let _store = store; let idx = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); HttpServer::new(config, query_engine, idx) }
             .with_hot_reload_config(hot_reload)
             .with_hot_reload_backend_storage_routing(routing_handle.clone());
         let port = server.start_test_server().await.expect("start ok");
@@ -4152,7 +4156,7 @@ aggregations:
             15000,
         ));
         let mut server =
-            HttpServer::new(config, query_engine, store).with_hot_reload_config(hot_reload);
+            { let _store = store; let idx = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); HttpServer::new(config, query_engine, idx) }.with_hot_reload_config(hot_reload);
         for engine in engines {
             server = server.with_query_engine(engine);
         }
@@ -4200,7 +4204,7 @@ aggregations:
             15000,
         ));
         let cache = Arc::new(crate::query_engines::routing::FreshnessProbeCache::new());
-        let server = HttpServer::new(config, query_engine, store).with_probe_cache(cache.clone());
+        let server = { let _store = store; let idx = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); HttpServer::new(config, query_engine, idx) }.with_probe_cache(cache.clone());
         let port = server
             .start_test_server()
             .await
@@ -4557,21 +4561,14 @@ async fn handle_store_metrics(State(state): State<AppState>) -> axum::response::
     use axum::http::StatusCode;
     use axum::response::IntoResponse;
 
-    match state.store.get_earliest_timestamp_per_aggregation_id() {
-        Ok(timestamps) => {
-            let body = serde_json::json!({
-                "status": "success",
-                "aggregation_count": timestamps.len(),
-                "earliest_timestamps": timestamps});
-            (StatusCode::OK, axum::Json(body)).into_response()
-        }
-        Err(e) => {
-            let body = serde_json::json!({
-                "status": "error",
-                "error": format!("{}", e)});
-            (StatusCode::INTERNAL_SERVER_ERROR, axum::Json(body)).into_response()
-        }
-    }
+    // M2.3.6g — earliest timestamps come from SketchIndex's per-sid
+    // `first_seen_unix_ms` metadata. Always succeeds (no I/O).
+    let timestamps = state.sketch_index.earliest_timestamps_per_sid();
+    let body = serde_json::json!({
+        "status": "success",
+        "sid_count": timestamps.len(),
+        "earliest_timestamps_per_sid": timestamps});
+    (StatusCode::OK, axum::Json(body)).into_response()
 }
 
 // ─── StreamingConfig hot-reload (PR E) ───────────────────────────────────

data_plane/src/main.rs

@@ -610,7 +610,7 @@ async fn main() -> Result<()> {
     // design, §6). When precompute isn't enabled, the registry is
     // absent and the swap handler no-ops on schema reconciliation
     // (legacy per-batch reconcile in ingest still works).
-    let mut server = HttpServer::new(http_config, engine, store.clone())
+    let mut server = HttpServer::new(http_config, engine, sketch_index.clone())
         .with_hot_reload_config(hot_reload_config.clone())
         .with_probe_cache(probe_cache.clone());
 

data_plane/src/stores/sketch_db/index/mod.rs

@@ -903,6 +903,19 @@ impl SketchIndex {
 }
 
 impl SketchIndex {
+    /// Phase 5 M2.3.6g — runtime-info / diagnostic helper. Returns the
+    /// per-sid `first_seen_unix_ms` for every registered sid. The
+    /// legacy `Store::get_earliest_timestamp_per_aggregation_id` returned
+    /// an analogous `agg_id → ts` map; this is the SketchIndex
+    /// equivalent. HTTP server's `/api/v1/status/runtimeinfo` adapter
+    /// surfaces it under the JSON field `earliest_timestamp_per_sid`.
+    pub fn earliest_timestamps_per_sid(&self) -> std::collections::HashMap<u64, u64> {
+        let g = self.instances.read().unwrap();
+        g.iter()
+            .map(|(sid, m)| (*sid, m.first_seen_unix_ms.max(0) as u64))
+            .collect()
+    }
+
     /// Phase 5 M2.3.6e — write-side helper. Given an
     /// `AggregationConfig` and one `(PrecomputedOutput, AggregateCore)`
     /// pair (the shape both the live worker AND the backfill processor

data_plane/src/tests/capability_miss_http_e2e_tests.rs

@@ -175,7 +175,9 @@ async fn start_backend(controller_url: String, hot_reload: HotReloadStreamingCon
         port: 0,
         handle_http_requests: true,
         adapter_config};
-    let server = HttpServer::new(config, engine, store).with_hot_reload_config(hot_reload.clone());
+    let _store = store;
+    let idx = std::sync::Arc::new(crate::stores::sketch_db::index::SketchIndex::new());
+    let server = HttpServer::new(config, engine, idx).with_hot_reload_config(hot_reload.clone());
     server
         .start_test_server()
         .await

data_plane/src/tests/prometheus_forwarding_tests.rs

@@ -82,7 +82,9 @@ async fn setup_test_server(prometheus_port: u16) -> (HttpServer, u16) {
         15000, // 15s scrape interval
     ));
 
-    let server = HttpServer::new(config, query_engine, store);
+    let _store = store;
+    let idx = std::sync::Arc::new(crate::stores::sketch_db::index::SketchIndex::new());
+    let server = HttpServer::new(config, query_engine, idx);
     let actual_port = server
         .start_test_server()
         .await
@@ -170,7 +172,9 @@ async fn test_forwarding_disabled() {
         15000, // 15s scrape interval
     ));
 
-    let server = HttpServer::new(config, query_engine, store);
+    let _store = store;
+    let idx = std::sync::Arc::new(crate::stores::sketch_db::index::SketchIndex::new());
+    let server = HttpServer::new(config, query_engine, idx);
     let server_port = server
         .start_test_server()
         .await
@@ -220,7 +224,9 @@ async fn test_prometheus_server_unreachable() {
         15000, // 15s scrape interval
     ));
 
-    let server = HttpServer::new(config, query_engine, store);
+    let _store = store;
+    let idx = std::sync::Arc::new(crate::stores::sketch_db::index::SketchIndex::new());
+    let server = HttpServer::new(config, query_engine, idx);
     let server_port = server
         .start_test_server()
         .await