From c4d5f059d0c699e13b710690f55d41fcd519801a Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 12 May 2026 21:37:31 -0600 Subject: [PATCH] feat(data_plane): wire persistence flusher against SketchIndex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 5 M2.3.6c — adds `SketchIndex::start_persistence` and `SketchIndexPersistence` (the live persistence harness — manifest + part cache + flusher handle), and constructs one in `main.rs` when `--persistence-enabled` is passed. The new persistence layer writes to `/sketch_index/`, a subdir distinct from the legacy `SketchStore` layout — keeps the two coexisting until the legacy can be deleted in subsequent sub-PRs. The flusher loop is unchanged — it's generic over `EpochSource` (PR #162's `SketchIndex` impl drives the new path; the legacy `SketchStorePerKey` impl drives the deprecated one which now sees no writes after M2.3.6a / PR #160). Read-back from disk on query is NOT in scope here — the in-memory `query_range` / `query_precomputes_by_agg` paths don't yet consult `PartCache`. That's a follow-up: subsequent M2.3.6 sub-PRs add the disk-read glue (the legacy `SketchStorePerKey` similarly stubbed this; see `source.rs:35`'s "Ok(None) until then" comment). Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/main.rs | 59 +++++++++++++++++ data_plane/src/stores/sketch_db/index/mod.rs | 66 ++++++++++++++++++++ 2 files changed, 125 insertions(+) diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index b173eaa4..eb1e05f2 100644 --- a/data_plane/src/main.rs +++ b/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 diff --git a/data_plane/src/stores/sketch_db/index/mod.rs b/data_plane/src/stores/sketch_db/index/mod.rs index acf0fb63..efb1fb04 100644 --- a/data_plane/src/stores/sketch_db/index/mod.rs +++ b/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, + 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, + cfg: crate::stores::sketch_db::store::persistence::SketchStorePersistenceConfig, + ) -> crate::stores::sketch_db::store::persistence::PersistResult + { + 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`)