From daaa42a73ff29d0fe89390a62c0b64a590897d1e Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Mon, 20 Apr 2026 09:37:35 -0400 Subject: [PATCH] =?UTF-8?q?fix(sketch-db):=20POST=20/api/v1/db/backfill=20?= =?UTF-8?q?uses=20create=5Fchecked=20(=C2=A710.5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The HTTP handler was calling `BackfillRegistry::create()`, which skips every invariant. PR #40 added `create_checked()` — the same entry point the `BackfillService` uses — but the HTTP surface kept the unchecked path, so controllers could file jobs against unknown agg_ids or time ranges that overlap live ingest / fall outside retention, and the worker would only discover the mistake later (or silently waste I/O on soon-to-be-evicted windows). Changes: - `handle_post_backfill_job` now requires both the backfill registry AND the schema registry (503 if either is absent), calls `create_checked(&schemas, ..., data_retention_ms)`, and maps `CreateError` to distinct HTTP statuses: 404 for `UnknownAgg`, 409 for `Overlap` / `OutOfRetention`, 201 on success. 400 on malformed body / inverted range is preserved. - `HttpServer::with_data_retention_ms(u64)` threads the SimpleMapStore data-retention horizon into AppState so Method B rejects stale ranges at create time. `main.rs` wires it from `--persistence-delete-older-than-secs` when > 0. - Test helper renamed `setup_test_server_with_backfill_and_schemas` and now takes an `&[u64]` of agg_ids to pre-register as Active. All four existing backfill tests updated. - Two new tests exercise the new error paths: `test_backfill_post_unknown_agg_returns_404` and `test_backfill_post_overlap_with_live_ingest_returns_409`. 725 lib tests pass (was 723), clippy clean, fmt clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/drivers/query/servers/http.rs | 190 ++++++++++++++++-- asap-query-engine/src/main.rs | 3 + 2 files changed, 176 insertions(+), 17 deletions(-) diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 0c56ca28..76f631ce 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -46,6 +46,13 @@ pub struct HttpServer { /// HTTP endpoints stay `Queued` and are visible via the list /// endpoint — useful shadow-mode testing before workers exist. backfill: Option>, + /// SimpleMapStore data-retention horizon in millis, mirroring + /// `--persistence-delete-older-than-secs` at the CLI. Used by the + /// `POST /api/v1/db/backfill` handler to gate job creation via + /// `BackfillRegistry::create_checked` (§10.5 Method B). `None` + /// disables the retention precheck — the handler still enforces + /// the §10.5 time-disjoint invariant. + data_retention_ms: Option, } #[derive(Clone)] @@ -66,6 +73,8 @@ struct AppState { schemas: Option>, /// Backfill registry (sketch DB §10). See `HttpServer::backfill`. backfill: Option>, + /// See `HttpServer::data_retention_ms`. + data_retention_ms: Option, } impl HttpServer { @@ -83,6 +92,7 @@ impl HttpServer { hot_reload_config: None, schemas: None, backfill: None, + data_retention_ms: None, } } @@ -124,6 +134,17 @@ impl HttpServer { self } + /// Declare the SimpleMapStore data-retention horizon (the value of + /// `--persistence-delete-older-than-secs` * 1000). When set, the + /// `POST /api/v1/db/backfill` handler runs `create_checked` with + /// this bound, so jobs that would write windows older than the + /// retention horizon are rejected up-front instead of being + /// silently evicted right after write (§10.5 Method B). + pub fn with_data_retention_ms(mut self, data_retention_ms: u64) -> Self { + self.data_retention_ms = Some(data_retention_ms); + self + } + pub async fn run(self) -> Result<(), Box> { // Create adapter using factory let adapter = create_http_adapter(self.config.adapter_config.clone()); @@ -147,6 +168,7 @@ impl HttpServer { hot_reload_config: self.hot_reload_config.clone(), schemas: self.schemas.clone(), backfill: self.backfill.clone(), + data_retention_ms: self.data_retention_ms, }; let range_query_endpoint = adapter.get_range_query_endpoint(); @@ -212,6 +234,7 @@ impl HttpServer { hot_reload_config: self.hot_reload_config.clone(), schemas: self.schemas.clone(), backfill: self.backfill.clone(), + data_retention_ms: self.data_retention_ms, }; let range_query_endpoint = adapter.get_range_query_endpoint(); @@ -1461,8 +1484,14 @@ aggregations: // ─── Phase 5d: backfill HTTP endpoint tests ───────────────────────────── - async fn setup_test_server_with_backfill( + /// Build a test server wired with a backfill registry and a + /// `SchemaRegistry` that pre-registers the listed `agg_ids` as + /// Active. `POST /api/v1/db/backfill` runs `create_checked`, which + /// requires both registries — tests that hit that endpoint must + /// populate the schema side here. + async fn setup_test_server_with_backfill_and_schemas( registry: Arc, + active_agg_ids: &[u64], ) -> u16 { let adapter_config = AdapterConfig::prometheus_promql("http://127.0.0.1:9999".to_string(), false); @@ -1487,8 +1516,40 @@ aggregations: 15000, crate::data_model::QueryLanguage::promql, )); - let server = - HttpServer::new(config, query_engine, store, None).with_backfill_registry(registry); + let schemas = { + use asap_types::aggregation_config::AggregationConfig; + use asap_types::enums::{AggregationType, WindowType}; + use promql_utilities::data_model::key_by_label_names::KeyByLabelNames; + let mut map: std::collections::HashMap = + std::collections::HashMap::new(); + for agg_id in active_agg_ids { + let cfg = AggregationConfig::new( + *agg_id, + AggregationType::CountMinSketch, + String::new(), + std::collections::HashMap::new(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 60, + 60, + WindowType::Tumbling, + String::new(), + format!("metric_{agg_id}"), + None, + None, + None, + None, + ); + map.insert(*agg_id, cfg); + } + let sc = StreamingConfig::new(map); + Arc::new(crate::stores::sketch_db::SchemaRegistry::from_streaming_config(&sc)) + }; + let server = HttpServer::new(config, query_engine, store, None) + .with_backfill_registry(registry) + .with_schemas(schemas); server .start_test_server() .await @@ -1498,7 +1559,8 @@ aggregations: #[tokio::test] async fn test_backfill_full_lifecycle_through_http() { let registry = Arc::new(crate::stores::sketch_db::BackfillRegistry::new()); - let server_port = setup_test_server_with_backfill(registry.clone()).await; + let server_port = + setup_test_server_with_backfill_and_schemas(registry.clone(), &[42]).await; let client = Client::new(); // POST creates a Queued job. @@ -1597,7 +1659,7 @@ aggregations: #[tokio::test] async fn test_backfill_post_rejects_inverted_range() { let registry = Arc::new(crate::stores::sketch_db::BackfillRegistry::new()); - let server_port = setup_test_server_with_backfill(registry).await; + let server_port = setup_test_server_with_backfill_and_schemas(registry, &[1]).await; let client = Client::new(); let req = serde_json::json!({ @@ -1619,7 +1681,7 @@ aggregations: #[tokio::test] async fn test_backfill_get_unknown_job_returns_404() { let registry = Arc::new(crate::stores::sketch_db::BackfillRegistry::new()); - let server_port = setup_test_server_with_backfill(registry).await; + let server_port = setup_test_server_with_backfill_and_schemas(registry, &[]).await; let client = Client::new(); let resp = client .get(format!( @@ -1678,7 +1740,7 @@ aggregations: #[tokio::test] async fn test_backfill_list_bogus_status_returns_400() { let registry = Arc::new(crate::stores::sketch_db::BackfillRegistry::new()); - let server_port = setup_test_server_with_backfill(registry).await; + let server_port = setup_test_server_with_backfill_and_schemas(registry, &[]).await; let client = Client::new(); let resp = client .get(format!( @@ -1689,6 +1751,64 @@ aggregations: .unwrap(); assert_eq!(resp.status(), reqwest::StatusCode::BAD_REQUEST); } + + // §10.5: create_checked invariants surfacing through HTTP. + + #[tokio::test] + async fn test_backfill_post_unknown_agg_returns_404() { + let registry = Arc::new(crate::stores::sketch_db::BackfillRegistry::new()); + // Empty schema registry — agg_id 42 is unknown. + let server_port = setup_test_server_with_backfill_and_schemas(registry, &[]).await; + let client = Client::new(); + let req = serde_json::json!({ + "agg_id": 42, + "start_ms": 100, + "end_ms": 500, + "source": { "Prometheus": { "url": "http://prom.local" } }, + "windows_total": 1, + }); + let resp = client + .post(format!("http://127.0.0.1:{server_port}/api/v1/db/backfill")) + .json(&req) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["status"], "error"); + assert!(body["error"] + .as_str() + .unwrap() + .contains("unknown agg_id 42")); + } + + #[tokio::test] + async fn test_backfill_post_overlap_with_live_ingest_returns_409() { + let registry = Arc::new(crate::stores::sketch_db::BackfillRegistry::new()); + // Schema registered at `now` — any `end_ms` > created_at_ms + // overlaps live ingest. + let server_port = setup_test_server_with_backfill_and_schemas(registry, &[7]).await; + let client = Client::new(); + let future_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64 + + 3_600_000; + let req = serde_json::json!({ + "agg_id": 7, + "start_ms": 0, + "end_ms": future_ms, + "source": { "Prometheus": { "url": "http://prom.local" } }, + "windows_total": 1, + }); + let resp = client + .post(format!("http://127.0.0.1:{server_port}/api/v1/db/backfill")) + .json(&req) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), reqwest::StatusCode::CONFLICT); + } } // ── Controller integration: PrecomputeJob execution ────────────────────────── @@ -2212,18 +2332,35 @@ fn service_unavailable_no_backfill() -> axum::response::Response { (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response() } -/// `POST /api/v1/db/backfill` — create a queued backfill job. Returns -/// 201 with `{job_id}` on success; 400 on bad body (malformed JSON -/// or inverted range); 503 when no registry is attached. +/// `POST /api/v1/db/backfill` — create a queued backfill job. +/// +/// Validates the request against the §10.5 invariants via +/// `BackfillRegistry::create_checked`: +/// * `agg_id` must be known to the schema registry → 404 on miss. +/// * `end_ms` must not extend past the agg's `created_at_ms` (no +/// race against live ingest) → 409 on overlap. +/// * `start_ms` must be within the SimpleMapStore data-retention +/// window when one is configured (Method B) → 409 on stale range. +/// +/// 400 on malformed body / inverted range; 503 when no registry or +/// schema registry is attached. async fn handle_post_backfill_job( State(state): State, body: axum::body::Bytes, ) -> axum::response::Response { + use crate::stores::sketch_db::CreateError; use axum::response::IntoResponse; let Some(registry) = state.backfill else { return service_unavailable_no_backfill(); }; + let Some(schemas) = state.schemas else { + let body = serde_json::json!({ + "status": "error", + "error": "schema registry not attached; backfill retention check requires HttpServer::with_schemas", + }); + return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response(); + }; let req: CreateBackfillJobRequest = match serde_json::from_slice(&body) { Ok(v) => v, @@ -2246,17 +2383,36 @@ async fn handle_post_backfill_job( return (StatusCode::BAD_REQUEST, axum::Json(body)).into_response(); } - let job_id = registry.create( + match registry.create_checked( + &schemas, req.agg_id, (req.start_ms, req.end_ms), req.source, req.windows_total, - ); - let body = serde_json::json!({ - "status": "success", - "job_id": job_id, - }); - (StatusCode::CREATED, axum::Json(body)).into_response() + state.data_retention_ms, + ) { + Ok(job_id) => { + let body = serde_json::json!({ + "status": "success", + "job_id": job_id, + }); + (StatusCode::CREATED, axum::Json(body)).into_response() + } + Err(e @ CreateError::UnknownAgg { .. }) => { + let body = serde_json::json!({ + "status": "error", + "error": e.to_string(), + }); + (StatusCode::NOT_FOUND, axum::Json(body)).into_response() + } + Err(e @ (CreateError::Overlap { .. } | CreateError::OutOfRetention { .. })) => { + let body = serde_json::json!({ + "status": "error", + "error": e.to_string(), + }); + (StatusCode::CONFLICT, axum::Json(body)).into_response() + } + } } /// `GET /api/v1/db/backfill/jobs` — list all jobs with optional diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index 48a6dddf..81d37a3a 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -670,6 +670,9 @@ async fn main() -> Result<()> { if let Some(ingest_state) = precompute_ingest_state.as_ref() { server = server.with_schemas(ingest_state.schemas.clone()); } + if args.persistence_delete_older_than_secs > 0 { + server = server.with_data_retention_ms(args.persistence_delete_older_than_secs * 1000); + } // Backfill registry (sketch DB §10). A single `Arc` lives in // `main` so the HTTP endpoints (Phase 5d) can inspect / cancel // jobs and the upcoming worker pool (Phase 5e) can drain them.