feat(data_plane): wire persistence flusher against SketchIndex

data_plane/src/main.rs

@@ -371,6 +371,65 @@ async fn main() -> Result<()> {
     let sketch_index =
         Arc::new(data_plane::stores::sketch_db::index::SketchIndex::new());
 
+    // M2.3.6c — also start a persistence layer behind the SketchIndex
+    // when --persistence-enabled. SketchIndex is now where all
+    // precompute + sketch writes land (M2.3.6a), so flushing it to
+    // disk is what makes Phase 5 warm-tier state survive restarts.
+    // The legacy `SketchStore::with_persistence_per_key` flusher
+    // constructed above is now a no-op (its source has no writes) —
+    // it stays in place until subsequent M2.3.6 sub-PRs delete the
+    // legacy SketchStore wholesale.
+    let _sketch_index_persistence = if args.persistence_enabled {
+        use data_plane::stores::sketch_db::store::persistence::SketchStorePersistenceConfig;
+        let disk_path = args
+            .persistence_dir
+            .clone()
+            .expect("--persistence-enabled requires --persistence-dir");
+        let memory_limit_bytes = args.persistence_memory_limit_mb * 1024 * 1024;
+        let hot_window_ms = if args.persistence_hot_window_secs == 0 {
+            None
+        } else {
+            Some(args.persistence_hot_window_secs * 1000)
+        };
+        let delete_older_than_ms = if args.persistence_delete_older_than_secs == 0 {
+            None
+        } else {
+            Some(args.persistence_delete_older_than_secs * 1000)
+        };
+        let part_cache_bytes = args
+            .persistence_part_cache_mb
+            .map(|mb| mb * 1024 * 1024)
+            .unwrap_or_else(|| {
+                let ten_pct = (memory_limit_bytes / 10) as u64;
+                ten_pct.min(512 * 1024 * 1024)
+            });
+        let index_persistence_dir =
+            std::path::PathBuf::from(&disk_path).join("sketch_index");
+        let cfg = SketchStorePersistenceConfig {
+            memory_limit_bytes,
+            memory_low_watermark_bytes: memory_limit_bytes * 8 / 10,
+            hard_cap_bytes: memory_limit_bytes * 125 / 100,
+            hot_window_ms,
+            delete_older_than_ms,
+            flush_interval: std::time::Duration::from_millis(
+                args.persistence_flush_interval_ms,
+            ),
+            disk_path: index_persistence_dir.clone(),
+            part_cache_bytes,
+        };
+        info!(
+            "SketchIndex persistence enabled: disk_path={:?}",
+            index_persistence_dir
+        );
+        Some(
+            sketch_index
+                .start_persistence(cfg)
+                .expect("SketchIndex::start_persistence failed"),
+        )
+    } else {
+        None
+    };
+
     // Setup query engine. ASAPQueryEngine shares the same
     // HotReloadStreamingConfig handle as the HTTP server, so a POST
     // to /api/v1/streaming-config is observable by the next query

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

@@ -902,6 +902,72 @@ impl SketchIndex {
     }
 }
 
+/// Persistence harness for `SketchIndex` — Phase 5 M2.3.6c.
+///
+/// Owns the manifest + flusher thread + part cache that back the
+/// sid-keyed warm tier. Constructed via [`SketchIndex::start_persistence`];
+/// the flusher reads sealed epochs through the
+/// [`EpochSource`](crate::stores::sketch_db::store::persistence::EpochSource)
+/// impl on `SketchIndex` and writes parts under `disk_path/parts/`.
+///
+/// Drop or call [`Self::shutdown`] to stop the flusher cleanly. The
+/// `part_cache` field is exposed so the query path can be wired up to
+/// read-back from disk in a subsequent sub-PR; today it sits idle
+/// because the in-memory `query_range` doesn't yet consult it.
+pub struct SketchIndexPersistence {
+    pub manifest: Arc<crate::stores::sketch_db::store::persistence::Manifest>,
+    pub part_cache: crate::stores::sketch_db::store::persistence::cache::PartCache,
+    pub flusher: crate::stores::sketch_db::store::persistence::flusher::FlusherHandle,
+    pub parts_root: std::path::PathBuf,
+}
+
+impl SketchIndexPersistence {
+    pub fn shutdown(&mut self) {
+        self.flusher.shutdown();
+    }
+}
+
+impl SketchIndex {
+    /// Spin up the persistence layer behind this `SketchIndex`. Runs
+    /// startup recovery (sweeps corrupt + orphan parts), opens the
+    /// manifest, and starts the background flusher thread with
+    /// `Arc::clone(self)` as its `EpochSource`. The returned
+    /// `SketchIndexPersistence` MUST stay alive for the lifetime of
+    /// the index — dropping it shuts the flusher down and stops
+    /// flushing to disk.
+    pub fn start_persistence(
+        self: &Arc<Self>,
+        cfg: crate::stores::sketch_db::store::persistence::SketchStorePersistenceConfig,
+    ) -> crate::stores::sketch_db::store::persistence::PersistResult<SketchIndexPersistence>
+    {
+        use crate::stores::sketch_db::store::persistence::{
+            cache::PartCache, flusher::FlusherHandle, recovery, Manifest,
+        };
+
+        let (_loaded_manifest, report) = recovery::recover(&cfg.disk_path)?;
+        tracing::info!(
+            live = report.live_parts,
+            corrupt_removed = report.corrupt_parts_removed,
+            orphans_removed = report.orphan_parts_removed,
+            "SketchIndex persistence recovery complete"
+        );
+
+        let manifest = Arc::new(Manifest::open_or_init(&cfg.disk_path)?);
+        let parts_root =
+            crate::stores::sketch_db::store::persistence::flusher::parts_root(&cfg.disk_path);
+        let part_cache = PartCache::new(parts_root.clone(), cfg.part_cache_bytes);
+
+        let flusher = FlusherHandle::start(cfg, Arc::clone(&manifest), Arc::clone(self))?;
+
+        Ok(SketchIndexPersistence {
+            manifest,
+            part_cache,
+            flusher,
+            parts_root,
+        })
+    }
+}
+
 // ── Phase 5 M2.3.6b — EpochSource impl ──────────────────────────────────────
 //
 // Lets the existing persistence flusher (`store/persistence/flusher.rs`)