feat(sketch-db): manual schema retire/expire HTTP endpoints

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<serde_json::Value> = 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<AppState>,
+    axum::extract::Path(agg_id): axum::extract::Path<u64>,
+) -> 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<AppState>,
+    axum::extract::Path(agg_id): axum::extract::Path<u64>,
+) -> 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 {

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<AggSchema> {
+        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<AggSchema> {
+        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());
+    }
 }