diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 977eff55..0c56ca28 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -168,6 +168,14 @@ impl HttpServer { get(handle_get_streaming_config).post(handle_post_streaming_config), ) .route("/api/v1/db/schemas", get(handle_get_schemas)) + .route( + "/api/v1/db/schemas/:agg_id/retire", + post(handle_post_schema_retire), + ) + .route( + "/api/v1/db/schemas/:agg_id/expire", + post(handle_post_schema_expire), + ) .route("/api/v1/db/timeline", get(handle_get_timeline)) .route("/api/v1/db/backfill", post(handle_post_backfill_job)) .route("/api/v1/db/backfill/jobs", get(handle_get_backfill_jobs)) @@ -219,6 +227,14 @@ impl HttpServer { get(handle_get_streaming_config).post(handle_post_streaming_config), ) .route("/api/v1/db/schemas", get(handle_get_schemas)) + .route( + "/api/v1/db/schemas/:agg_id/retire", + post(handle_post_schema_retire), + ) + .route( + "/api/v1/db/schemas/:agg_id/expire", + post(handle_post_schema_expire), + ) .route("/api/v1/db/timeline", get(handle_get_timeline)) .route("/api/v1/db/backfill", post(handle_post_backfill_job)) .route("/api/v1/db/backfill/jobs", get(handle_get_backfill_jobs)) @@ -1947,17 +1963,7 @@ async fn handle_get_schemas( let mut entries: Vec = Vec::new(); for status in statuses { for s in schemas.list_by_status(*status) { - let accuracy = s.accuracy_profile(); - entries.push(serde_json::json!({ - "agg_id": s.agg_id, - "metric_name": s.metric_name, - "status": status_str(*status), - "created_at_ms": s.created_at_ms, - "retired_at_ms": s.retired_at_ms, - "expires_at_ms": s.expires_at_ms, - "aggregation_type": format!("{:?}", s.config.aggregation_type), - "accuracy_profile": accuracy, - })); + entries.push(schema_to_json(&s)); } } entries.sort_by_key(|v| v.get("agg_id").and_then(|x| x.as_u64()).unwrap_or(0)); @@ -1979,6 +1985,88 @@ fn status_str(s: crate::stores::sketch_db::AggStatus) -> &'static str { } } +fn schema_to_json(s: &crate::stores::sketch_db::AggSchema) -> serde_json::Value { + serde_json::json!({ + "agg_id": s.agg_id, + "metric_name": s.metric_name, + "status": status_str(s.status()), + "created_at_ms": s.created_at_ms, + "retired_at_ms": s.retired_at_ms, + "expires_at_ms": s.expires_at_ms, + "aggregation_type": format!("{:?}", s.config.aggregation_type), + "accuracy_profile": s.accuracy_profile(), + }) +} + +/// `POST /api/v1/db/schemas/:agg_id/retire` — manually transition an +/// Active schema to Retired (kicking off the retirement retention +/// clock). Idempotent: already-Retired or Expired schemas return 200 +/// with their current state unchanged. Returns 404 if the agg_id is +/// unknown, 503 if no registry is attached. +async fn handle_post_schema_retire( + State(state): State, + axum::extract::Path(agg_id): axum::extract::Path, +) -> axum::response::Response { + use axum::response::IntoResponse; + let Some(schemas) = state.schemas else { + let body = serde_json::json!({ + "status": "error", + "error": "schema registry not attached", + }); + return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response(); + }; + match schemas.force_retire(agg_id) { + Some(schema) => { + let body = serde_json::json!({ + "status": "success", + "schema": schema_to_json(&schema), + }); + (StatusCode::OK, axum::Json(body)).into_response() + } + None => { + let body = serde_json::json!({ + "status": "error", + "error": format!("agg_id {agg_id} not found"), + }); + (StatusCode::NOT_FOUND, axum::Json(body)).into_response() + } + } +} + +/// `POST /api/v1/db/schemas/:agg_id/expire` — manually transition a +/// schema to Expired immediately. The next `SchemaEvictionService` +/// tick drops the agg's data + removes the schema. Idempotent; +/// 404 if the agg_id is unknown, 503 if no registry is attached. +async fn handle_post_schema_expire( + State(state): State, + axum::extract::Path(agg_id): axum::extract::Path, +) -> axum::response::Response { + use axum::response::IntoResponse; + let Some(schemas) = state.schemas else { + let body = serde_json::json!({ + "status": "error", + "error": "schema registry not attached", + }); + return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response(); + }; + match schemas.force_expire(agg_id) { + Some(schema) => { + let body = serde_json::json!({ + "status": "success", + "schema": schema_to_json(&schema), + }); + (StatusCode::OK, axum::Json(body)).into_response() + } + None => { + let body = serde_json::json!({ + "status": "error", + "error": format!("agg_id {agg_id} not found"), + }); + (StatusCode::NOT_FOUND, axum::Json(body)).into_response() + } + } +} + fn coverage_str(c: crate::stores::sketch_db::TimelineCoverage) -> &'static str { use crate::stores::sketch_db::TimelineCoverage; match c { diff --git a/asap-query-engine/src/stores/sketch_db/schema.rs b/asap-query-engine/src/stores/sketch_db/schema.rs index 4984bfcf..da9627ad 100644 --- a/asap-query-engine/src/stores/sketch_db/schema.rs +++ b/asap-query-engine/src/stores/sketch_db/schema.rs @@ -428,6 +428,50 @@ impl SchemaRegistry { removed } + /// Force a specific `agg_id` into `Retired` status, starting the + /// configured retirement retention clock. Idempotent — re-retiring + /// a Retired or Expired schema is a no-op and returns `Some(schema)` + /// reflecting the current (unchanged) state. Returns `None` if + /// the `agg_id` is unknown. + /// + /// Intended for operator / debug-endpoint use so the eviction path + /// can be driven without waiting for a `StreamingConfig` swap to + /// drop the agg. + pub fn force_retire(&self, agg_id: u64) -> Option { + let retention = self.retirement_retention; + let updated = { + let mut map = self.schemas.write().ok()?; + let schema = map.get_mut(&agg_id)?; + if matches!(schema.status(), AggStatus::Active) { + schema.retire(retention); + } + schema.clone() + }; + self.save_to_disk_if_persistent(); + Some(updated) + } + + /// Force a specific `agg_id` into `Expired` status immediately by + /// setting both `retired_at_ms` and `expires_at_ms` to now. The + /// next `SchemaEvictionService` tick will drop its data and + /// remove the schema. Returns the new state, or `None` if the + /// `agg_id` is unknown. + /// + /// Intended for operator / debug-endpoint use so eviction can be + /// observed in e2e tests without waiting out retirement retention. + pub fn force_expire(&self, agg_id: u64) -> Option { + let updated = { + let mut map = self.schemas.write().ok()?; + let schema = map.get_mut(&agg_id)?; + let now = now_ms(); + schema.retired_at_ms = Some(now); + schema.expires_at_ms = Some(now); + schema.clone() + }; + self.save_to_disk_if_persistent(); + Some(updated) + } + /// Iterate (clones) all schemas matching a status filter. Used by /// the controller-facing `/api/v1/db/schemas?status=…` endpoint /// (§15.2 of the design). @@ -1187,4 +1231,48 @@ mod tests { let _ = registry.reconcile(&make_streaming_config(&[2])); assert!(!path.exists()); } + + // --- manual retire / expire endpoints (debug/operator surface) --- + + #[test] + fn force_retire_active_transitions_to_retired() { + let r = SchemaRegistry::from_streaming_config(&make_streaming_config(&[1])); + assert_eq!(r.get(1).unwrap().status(), AggStatus::Active); + let out = r.force_retire(1).expect("should return new state"); + assert_eq!(out.status(), AggStatus::Retired); + assert!(out.retired_at_ms.is_some()); + assert!(out.expires_at_ms.is_some()); + assert_eq!(r.get(1).unwrap().status(), AggStatus::Retired); + } + + #[test] + fn force_retire_is_idempotent_on_retired() { + let r = SchemaRegistry::from_streaming_config(&make_streaming_config(&[1])); + let first = r.force_retire(1).unwrap(); + let first_exp = first.expires_at_ms; + std::thread::sleep(std::time::Duration::from_millis(5)); + let second = r.force_retire(1).unwrap(); + assert_eq!(second.expires_at_ms, first_exp); + } + + #[test] + fn force_retire_unknown_returns_none() { + let r = SchemaRegistry::from_streaming_config(&make_streaming_config(&[1])); + assert!(r.force_retire(999).is_none()); + } + + #[test] + fn force_expire_active_transitions_to_expired() { + let r = SchemaRegistry::from_streaming_config(&make_streaming_config(&[1])); + assert_eq!(r.get(1).unwrap().status(), AggStatus::Active); + let out = r.force_expire(1).expect("should return new state"); + assert_eq!(out.status(), AggStatus::Expired); + assert_eq!(r.get(1).unwrap().status(), AggStatus::Expired); + } + + #[test] + fn force_expire_unknown_returns_none() { + let r = SchemaRegistry::from_streaming_config(&make_streaming_config(&[1])); + assert!(r.force_expire(999).is_none()); + } }