feat(sketch-db): Phase 5 eviction — SchemaEvictionService background task

asap-query-engine/src/main.rs

@@ -207,6 +207,26 @@ struct Args {
     #[arg(long)]
     enable_backfill_worker: bool,
 
+    /// Spawn the Phase 5 schema eviction service. Periodically
+    /// scans the schema registry for `Expired` schemas, cancels
+    /// any in-flight backfill jobs targeting them, and drops the
+    /// agg_id's data from the store. Requires the schema registry
+    /// (same precompute-engine dependency as --enable-backfill-worker).
+    #[arg(long)]
+    enable_schema_eviction: bool,
+
+    /// Poll interval for the schema eviction service, in seconds.
+    /// Default 300s (5 minutes) — eviction is not latency-sensitive.
+    #[arg(long, default_value = "300")]
+    schema_eviction_poll_secs: u64,
+
+    /// When true, the eviction service logs what it would drop but
+    /// doesn't actually call `drop_agg_id` / `remove_schema`. Use
+    /// to validate a new retention value before letting it delete
+    /// anything.
+    #[arg(long)]
+    schema_eviction_dry_run: bool,
+
     /// Enable automatic query tracking and planning
     #[arg(long)]
     enable_query_tracker: bool,
@@ -703,6 +723,51 @@ async fn main() -> Result<()> {
         None
     };
 
+    // Phase 5: schema eviction service. On every poll interval,
+    // scans the schema registry for `Expired` schemas, cancels any
+    // in-flight backfills targeting them, and drops the agg_id's
+    // data from the store. Complements the age-based data retention
+    // in SimpleMapStore — see `SchemaEvictionService` module doc for
+    // the ordering rationale.
+    let schema_eviction_handle = if let (true, Some(ingest_state)) = (
+        args.enable_schema_eviction,
+        precompute_ingest_state.as_ref(),
+    ) {
+        let data_retention_opt = if args.persistence_delete_older_than_secs == 0 {
+            None
+        } else {
+            Some(std::time::Duration::from_secs(
+                args.persistence_delete_older_than_secs,
+            ))
+        };
+        query_engine_rust::stores::sketch_db::warn_if_retention_inverted(
+            data_retention_opt,
+            ingest_state.schemas.retirement_retention(),
+        );
+        let svc = query_engine_rust::stores::sketch_db::SchemaEvictionService::new(
+            ingest_state.schemas.clone(),
+            backfill_registry.clone(),
+            store.clone(),
+            query_engine_rust::stores::sketch_db::SchemaEvictionConfig {
+                poll_interval: std::time::Duration::from_secs(args.schema_eviction_poll_secs),
+                dry_run: args.schema_eviction_dry_run,
+            },
+        );
+        info!(
+            poll_secs = args.schema_eviction_poll_secs,
+            dry_run = args.schema_eviction_dry_run,
+            "Spawning SchemaEvictionService"
+        );
+        Some(svc.spawn())
+    } else {
+        if args.enable_schema_eviction {
+            warn!(
+                "--enable-schema-eviction set but precompute engine isn't enabled; eviction service NOT spawned"
+            );
+        }
+        None
+    };
+
     info!("Starting HTTP server on port {}", args.http_port);
 
     // Wait for shutdown signal
@@ -723,6 +788,11 @@ async fn main() -> Result<()> {
         handle.shutdown().await;
     }
 
+    if let Some(handle) = schema_eviction_handle {
+        info!("Shutting down schema eviction service...");
+        handle.shutdown().await;
+    }
+
     if let Some(handle) = kafka_handle {
         info!("Shutting down Kafka consumer...");
         handle.abort();

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

@@ -37,6 +37,7 @@ pub mod backfill_worker;
 pub mod prometheus_reader;
 pub mod raw_sample_reader;
 pub mod schema;
+pub mod schema_eviction;
 pub mod simple_map_store;
 
 pub use accuracy::{AccuracyKind, AccuracyProfile};
@@ -55,4 +56,7 @@ pub use raw_sample_reader::{
     LabelFilter, MockRawSampleReader, RawSample, RawSampleReader, RawSampleReaderError,
 };
 pub use schema::{AggSchema, AggStatus, SchemaRegistry, TimelineCoverage, TimelineSegment};
+pub use schema_eviction::{
+    warn_if_retention_inverted, SchemaEvictionConfig, SchemaEvictionHandle, SchemaEvictionService,
+};
 pub use simple_map_store::SimpleMapStore;

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

@@ -179,10 +179,14 @@ impl AggSchema {
     }
 }
 
-/// Default retention for retired schemas — one hour. A future
-/// `AggregationConfig` field could let the controller override this
-/// per-agg; for Phase 2a a single global default is enough.
-pub const DEFAULT_RETIREMENT_RETENTION: Duration = Duration::from_secs(3600);
+/// Default retention for a retired schema before eviction.
+/// 24 hours — covers dashboards / ad-hoc queries that may still
+/// reference the old agg_id mid-reconfigure. Shorter than the
+/// typical SimpleMapStore data retention (7d+) so schema eviction
+/// runs first, freeing space cleanly without fighting per-record
+/// retention. Override via `SchemaRegistry::set_retention_for_testing`
+/// or the CLI flag plumbed through `SchemaEvictionService`.
+pub const DEFAULT_RETIREMENT_RETENTION: Duration = Duration::from_secs(24 * 3600);
 
 /// In-memory registry of `AggSchema` keyed by `aggregation_id`.
 ///
@@ -404,6 +408,26 @@ impl SchemaRegistry {
         self.schemas.read().ok()?.get(&agg_id).cloned()
     }
 
+    /// Remove a schema record from the registry. Used by
+    /// `SchemaEvictionService` after it's dropped the agg's data
+    /// from the store. Returns the removed schema if it existed,
+    /// `None` if the agg_id was already absent (idempotent).
+    ///
+    /// Triggers a `save_to_disk_if_persistent` so restart behaviour
+    /// stays consistent (an evicted agg won't reappear on restart
+    /// from a stale persisted snapshot).
+    pub fn remove_schema(&self, agg_id: u64) -> Option<AggSchema> {
+        let removed = if let Ok(mut map) = self.schemas.write() {
+            map.remove(&agg_id)
+        } else {
+            None
+        };
+        if removed.is_some() {
+            self.save_to_disk_if_persistent();
+        }
+        removed
+    }
+
     /// Iterate (clones) all schemas matching a status filter. Used by
     /// the controller-facing `/api/v1/db/schemas?status=…` endpoint
     /// (§15.2 of the design).
@@ -585,9 +609,28 @@ impl SchemaRegistry {
         segments
     }
 
-    /// Override the default retirement retention. Test-only for now;
-    /// production tunable will land in Phase 2b's controller-facing
-    /// API.
+    /// Override the retirement retention. Plumbed through from
+    /// `SchemaEvictionService` at startup so deployments can pick
+    /// a retention that's ≤ their SimpleMapStore
+    /// `persistence_delete_older_than` (see module-level doc on
+    /// retention ordering).
+    ///
+    /// Takes `&mut self` because registry construction patterns
+    /// already produce a mutable local before wrapping in `Arc`.
+    /// Once wrapped, retention is immutable.
+    pub fn set_retention(&mut self, retention: Duration) {
+        self.retirement_retention = retention;
+    }
+
+    /// Expose the current retirement retention so the eviction
+    /// service can log it + diff it against the data-retention
+    /// config at startup.
+    pub fn retirement_retention(&self) -> Duration {
+        self.retirement_retention
+    }
+
+    /// Test-only alias for `set_retention`; kept as a separate
+    /// name so pre-Phase-5h test call sites don't need renaming.
     #[cfg(test)]
     pub fn set_retention_for_testing(&mut self, retention: Duration) {
         self.retirement_retention = retention;