From c597838f290f4fd7de5571c59f9d148b2e894951 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 25 May 2026 08:22:08 -0600 Subject: [PATCH] feat(data_plane): durable disk-backed tier for warm SketchStore Compose hot (current_epoch) -> sealed (in-mem, pending flush) -> disk parts into a single tiered store so warm-sketch memory is bounded by flush-then-evict rather than #327's age-based drop. Reads union all three tiers across the requested range. - Sealing now fires under persistence: SidStoreData gains a seal_window_count cadence (default 20 windows ~= 10 min of 30s panes) so current_epoch rotates into sealed_epochs for the flusher to persist. max_epochs drop is disabled under persistence (the flusher owns sealed-epoch lifecycle). In-memory-only deploys keep #327. - query_range/union_disk_parts_into consult PartCache+Manifest for the evicted portion of the range, rebuilding the full label key->value map from the sid's group_by_keys and preserving the #323-#326 read contract (half-open overlap + delta-stitching carry-in) across the in-mem/on-disk boundary -- incl. a carry-in Full base that now lives on disk. Part format round-trips SketchEncoding via a repurposed v1 pad byte (legacy 0 decodes as Full). - enforce_retention is a no-op under persistence so retention never drops un-flushed sealed/current data; the disk-tier TTL bounds the durable copy. In-memory-only path is unchanged. - start_persistence installs a read handle + seal cadence and recovers the manifest+parts so a restart immediately serves recovered data. - New CLI flag --persistence-seal-window-count plumbs the cadence. Tests: seal-fires, flush+evict bounds memory, query-from-disk incl. disk carry-in base, restart recovery, and persistence-disabled non-regression. Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/main.rs | 9 + .../sketch_db/index/epoch_columnar.rs | 140 +++- .../storage_engines/sketch_db/index/mod.rs | 617 +++++++++++++++++- .../sketch_db/persistence/config.rs | 17 + .../sketch_db/persistence/flusher.rs | 2 + .../sketch_db/persistence/part.rs | 35 +- .../sketch_db/persistence/recovery.rs | 1 + .../sketch_db/persistence/source.rs | 8 + 8 files changed, 805 insertions(+), 24 deletions(-) diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 41f06418..95369582 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -238,6 +238,14 @@ struct Args { #[arg(long)] persistence_part_cache_mb: Option, + /// Seal cadence in DISTINCT WINDOWS. The per-sid hot epoch is sealed + /// into the (pending-flush) sealed ring once it accumulates this + /// many distinct windows, giving the flusher sealed epochs to make + /// durable. ~30s panes ⇒ 20 windows ≈ 10 min per part. 0 disables + /// cadence sealing (no durable tier even with --persistence-enabled). + #[arg(long, default_value = "20")] + persistence_seal_window_count: usize, + /// Path to the per-metric backend storage routing YAML /// (`{metric_name: storage_backend}` map). Loaded at startup and /// consulted by the HTTP query handler on every PromQL request to @@ -376,6 +384,7 @@ async fn main() -> Result<()> { ), disk_path: index_persistence_dir.clone(), part_cache_bytes, + seal_window_count: args.persistence_seal_window_count, }; info!( "SketchStore persistence enabled: disk_path={:?}", diff --git a/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs index e60f40a1..fe168088 100644 --- a/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs +++ b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs @@ -736,6 +736,29 @@ pub struct SidStoreData { /// carry-in reach, so recent range/instant queries never lose a /// window or its Full base. See `evict_window_ends_before`. pub retention_horizon_ms: Option, + /// Seal cadence (in DISTINCT WINDOWS) for the durable tier. When + /// `Some(n)`, `current_epoch` is sealed into `sealed_epochs` once it + /// holds `n` distinct windows, so the persistence flusher has sealed + /// epochs to flush to disk. `None` (the default) means "never seal + /// on cadence" — the in-memory-only deployment where #327 retention + /// bounds memory by dropping aged windows from `current_epoch`. + /// + /// Distinct from [`Self::epoch_capacity`], which is the test-only + /// rotation threshold; both feed [`Self::maybe_rotate_epoch`], and + /// the smaller of the two (when set) wins. The store sets THIS field + /// (not `epoch_capacity`) when persistence is enabled so the + /// production seal cadence is decoupled from the test knob. + pub seal_window_count: Option, + /// When `true`, this sid's memory is bounded by the persistence + /// flush-then-evict loop, NOT by [`Self::enforce_retention`]. The + /// flusher owns the lifecycle of sealed epochs (seal → flush to a + /// durable disk part → evict from memory), and the disk-tier TTL + /// (`delete_older_than_ms`) bounds the durable copy. Retention must + /// not drop a sealed epoch out from under a pending flush, nor evict + /// un-sealed `current_epoch` windows that were never made durable. + /// So when this is `true`, `enforce_retention` is a no-op. When + /// `false` (the default), #327 retention is the memory bound. + pub persistence_enabled: bool, } /// Default in-memory retention horizon (ms) for the WARM sketch store. @@ -776,6 +799,8 @@ impl SidStoreData { epoch_capacity: None, max_epochs: 4, retention_horizon_ms: default_retention_horizon_ms(), + seal_window_count: None, + persistence_enabled: false, } } @@ -803,6 +828,14 @@ impl SidStoreData { /// Full base both still resolve. Eviction keys on window-END so a /// straddling pane survives until fully behind the horizon. fn enforce_retention(&mut self) { + // Persistence-enabled sids are bounded by the flush-then-evict + // loop, not by dropping. Retention must NOT race the flusher by + // dropping a sealed epoch before it has been made durable, nor + // evict un-sealed `current_epoch` windows that were never + // flushed. The flusher's disk-tier TTL bounds the durable copy. + if self.persistence_enabled { + return; + } let Some(horizon) = self.retention_horizon_ms else { return; }; @@ -837,11 +870,17 @@ impl SidStoreData { } fn maybe_rotate_epoch(&mut self) { - let cap = match self.epoch_capacity { - Some(c) if c > 0 => c, - _ => return, + // The effective rotation threshold is the smaller of the + // test-only `epoch_capacity` and the durable-tier + // `seal_window_count` (whichever is set); when both are unset, + // we never seal on cadence. + let cap = match (self.epoch_capacity, self.seal_window_count) { + (Some(a), Some(b)) => a.min(b), + (Some(a), None) => a, + (None, Some(b)) => b, + (None, None) => return, }; - if self.current_epoch.distinct_windows() < cap { + if cap == 0 || self.current_epoch.distinct_windows() < cap { return; } // Seal the current epoch and rotate. @@ -853,13 +892,20 @@ impl SidStoreData { self.sealed_epochs.insert(self.current_epoch_id, sealed); self.current_epoch_id += 1; - // Drop oldest sealed if we exceed `max_epochs`. - while self.sealed_epochs.len() + 1 > self.max_epochs { - // BTreeMap::pop_first is stable in 1.66+ - if let Some((id, _)) = self.sealed_epochs.iter().next().map(|(k, _)| (*k, ())) { - self.sealed_epochs.remove(&id); - } else { - break; + // Drop oldest sealed if we exceed `max_epochs` — but ONLY when + // persistence is OFF. Under persistence the flusher owns sealed- + // epoch lifecycle (seal → durable part → evict); dropping a + // sealed epoch here would discard data that was never flushed, + // defeating the durable tier. So persistence-enabled sids keep + // every sealed epoch in memory until the flusher evicts it. + if !self.persistence_enabled { + while self.sealed_epochs.len() + 1 > self.max_epochs { + // BTreeMap::pop_first is stable in 1.66+ + if let Some((id, _)) = self.sealed_epochs.iter().next().map(|(k, _)| (*k, ())) { + self.sealed_epochs.remove(&id); + } else { + break; + } } } } @@ -1109,4 +1155,76 @@ mod tests { "30m range query within horizon returned no windows — read path regressed" ); } + + #[test] + fn seal_window_count_seals_current_epoch_on_cadence() { + // Persistence mode: seal every 3 distinct windows. After 7 + // windows we expect 2 sealed epochs (windows 0..3, 3..6) plus a + // partial current epoch (window 6). Nothing is dropped — the + // flusher owns sealed-epoch lifecycle, so max_epochs does NOT + // bite under persistence. + let mut s = SidStoreData::::new(); + s.seal_window_count = Some(3); + s.persistence_enabled = true; + s.max_epochs = 2; // would normally cap sealed at 1; ignored here + let window_ms = 30_000u64; + let mut start = 0u64; + for i in 0..7u32 { + s.insert((start, start + window_ms), "series".into(), i); + start += window_ms; + } + assert_eq!( + s.sealed_epochs.len(), + 2, + "expected 2 sealed epochs at cadence 3 over 7 windows; max_epochs must not drop under persistence" + ); + assert!(s.current_epoch.distinct_windows() >= 1); + } + + #[test] + fn persistence_enabled_disables_retention_drop() { + // With persistence on, enforce_retention must be a no-op even + // when a retention horizon is set — the flush-then-evict loop + // (not dropping) is the memory bound. A long stream keeps every + // window in memory until the flusher evicts the sealed epochs. + let mut s = SidStoreData::::new(); + s.persistence_enabled = true; + s.retention_horizon_ms = Some(60 * 60 * 1000); // 1h — would normally drop + // No seal cadence: everything stays in current_epoch. + let window_ms = 30_000u64; + let mut start = 0u64; + for i in 0..480u32 { + // 4h of ingest + s.insert((start, start + window_ms), "series".into(), i); + start += window_ms; + } + assert_eq!( + s.current_epoch.distinct_windows(), + 480, + "persistence mode must not retention-drop; the flusher bounds memory instead" + ); + } + + #[test] + fn sealed_epochs_survive_for_flush_under_persistence() { + // Sealed epochs accumulate (pending flush) and are NOT dropped by + // either max_epochs rotation or retention while persistence is on. + let mut s = SidStoreData::::new(); + s.seal_window_count = Some(2); + s.persistence_enabled = true; + s.retention_horizon_ms = Some(1); // aggressive; must be ignored + s.max_epochs = 2; + let window_ms = 30_000u64; + let mut start = 0u64; + for i in 0..10u32 { + s.insert((start, start + window_ms), "series".into(), i); + start += window_ms; + } + // 10 windows / cadence 2 = up to 5 sealed epochs; none dropped. + assert!( + s.sealed_epochs.len() >= 4, + "sealed epochs were dropped under persistence (got {})", + s.sealed_epochs.len() + ); + } } diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 61277db8..0f27b580 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -45,6 +45,34 @@ fn now_ms() -> u64 { .unwrap_or(0) } +/// Map a [`SketchEncoding`] to the on-disk encoding tag stored per part +/// entry, so the disk read-back path can reconstruct the Full-vs-Delta +/// distinction the delta-stitching carry-in relies on. +fn encoding_to_tag(enc: SketchEncoding) -> u8 { + use crate::storage_engines::sketch_db::persistence::part::encoding_tag as t; + match enc { + SketchEncoding::ProtoFull => t::PROTO_FULL, + SketchEncoding::ProtoDelta => t::PROTO_DELTA, + SketchEncoding::MsgpackFull => t::MSGPACK_FULL, + SketchEncoding::MsgpackDelta => t::MSGPACK_DELTA, + } +} + +/// Inverse of [`encoding_to_tag`]. The unknown tag (`0`, written by the +/// original v1 part writer) decodes to `ProtoFull` — the safe default +/// for a carry-in base, since a Full snapshot establishes its own +/// rolling state with no predecessor. +fn tag_to_encoding(tag: u8) -> SketchEncoding { + use crate::storage_engines::sketch_db::persistence::part::encoding_tag as t; + match tag { + t::PROTO_DELTA => SketchEncoding::ProtoDelta, + t::MSGPACK_FULL => SketchEncoding::MsgpackFull, + t::MSGPACK_DELTA => SketchEncoding::MsgpackDelta, + // t::PROTO_FULL and t::UNKNOWN (legacy) both → Full. + _ => SketchEncoding::ProtoFull, + } +} + /// Joint helper shared by [`SketchStore::ingest_precompute_for_agg_config`] /// and [`SketchStore::ingest_precompute_with_sid`] — folds the /// grouping-label values on `output` against the @@ -236,6 +264,33 @@ pub struct SketchStore { /// `0` (the `Default`) means "never reconciled" so the first batch /// always runs. A real `Arc` data pointer is never null. last_reconciled_config_ptr: std::sync::atomic::AtomicUsize, + /// Durable-tier read handle, installed by [`Self::start_persistence`] + /// when `--persistence-enabled`. `None` (the default) means the + /// in-memory-only deployment: `query_range` reads HOT + SEALED + /// in-memory state and #327 retention bounds memory. When `Some`, + /// `query_range` ALSO unions in flushed-then-evicted DISK parts for + /// the portion of the range that has left memory, and the per-sid + /// `SidStoreData` is configured to seal on a cadence (so the flusher + /// has sealed epochs to persist) with retention-drop disabled (the + /// flush-then-evict loop is the memory bound). + persistence_read: RwLock>>, + /// Seal cadence in distinct windows, applied to every per-sid + /// `SidStoreData` once persistence is enabled. `0` (the default) + /// disables cadence sealing. Set by [`Self::enable_persistence_mode`]. + seal_window_count: std::sync::atomic::AtomicUsize, +} + +/// Read-side handle to the durable tier — the manifest of live disk +/// parts plus the byte-bounded `PartCache` that mmaps them. Cloned (as +/// an `Arc`) into `SketchStore::persistence_read` so the query path can +/// consult disk parts without holding a reference to the flusher. +/// +/// Also recovered on restart: `start_persistence` installs a fresh +/// handle pointing at the recovered manifest, so a reopened store sees +/// every part that was durable before the crash. +pub struct PersistenceReadHandle { + pub manifest: Arc, + pub part_cache: crate::storage_engines::sketch_db::index::persistence::cache::PartCache, } /// Three possible outcomes of looking up a sid in the SketchStore. @@ -341,12 +396,28 @@ impl SketchStore { let store = self .series .entry(sid) - .or_insert_with(|| Arc::new(RwLock::new(SidStoreData::new()))) + .or_insert_with(|| Arc::new(RwLock::new(self.fresh_sid_store()))) .clone(); let mut guard = store.write().unwrap(); guard.insert(window, series_label_values, AggPayload::Sketch(sample)); } + /// Build a `SidStoreData` pre-configured for the store's current + /// persistence mode. When persistence is enabled it seals on the + /// configured window cadence and disables retention-drop (the + /// flush-then-evict loop bounds memory). When disabled it's the + /// plain in-memory store with #327 retention. + fn fresh_sid_store(&self) -> SidStoreData, AggPayload> { + use std::sync::atomic::Ordering; + let mut data = SidStoreData::new(); + let cadence = self.seal_window_count.load(Ordering::Relaxed); + if cadence > 0 { + data.seal_window_count = Some(cadence); + data.persistence_enabled = true; + } + data + } + /// Append a window's exact-aggregation (Sum/Count/Avg/Rate/MinMax) /// state under `sid`. Mirror of [`Self::append_sample`] for the /// exact-agg branch — Phase 5 M2.3.3. @@ -366,7 +437,7 @@ impl SketchStore { let store = self .series .entry(sid) - .or_insert_with(|| Arc::new(RwLock::new(SidStoreData::new()))) + .or_insert_with(|| Arc::new(RwLock::new(self.fresh_sid_store()))) .clone(); let mut guard = store.write().unwrap(); guard.insert(window, series_label_values, AggPayload::ExactAgg(payload)); @@ -383,10 +454,19 @@ impl SketchStore { start_unix_ms: u64, end_unix_ms: u64, ) -> Vec { - let store = match self.series.get(&sid) { - Some(s) => s.clone(), - None => return Vec::new(), - }; + // Result is keyed by the resolved label MAP so the in-memory tier + // (its own intern space) and the durable disk tier (independent + // intern space) union by label identity, not `LabelValuesId`. + let mut by_label_map: HashMap, BTreeMap> = + HashMap::new(); + + // ── In-memory tier ────────────────────────────────────────────── + // Absent series is NOT an early return: under persistence the + // sid's hot+sealed state may have been fully flushed-then-evicted + // (or recovered from disk after a restart with no fresh ingest + // yet), so the answer can live entirely on disk. We still run the + // disk union below. + if let Some(store) = self.series.get(&sid).map(|s| s.clone()) { let guard = store.write().unwrap(); // exact_query may build the lazy index let mut by_label_id: HashMap> = HashMap::new(); @@ -515,10 +595,31 @@ impl SketchStore { } } - by_label_id + // Materialize the in-memory result keyed by the resolved label + // MAP so the disk tier (which has its own intern space) can be + // unioned by label identity rather than `LabelValuesId`. + for (label_id, samples) in by_label_id { + let label_values = guard.intern.resolve(label_id).cloned().unwrap_or_default(); + by_label_map.entry(label_values).or_default().extend(samples); + } + // Release the per-sid lock before touching disk — disk reads can + // mmap/decode and must not hold the hot ingest lock. + drop(guard); + } // end in-memory tier + + // Union the DURABLE DISK TIER for the part of `[start, end)` that + // has been flushed-then-evicted from memory. Preserves the + // #323–#326 read contract across the in-mem/on-disk boundary: + // the same half-open overlap admits straddling panes, and a + // delta-stitching carry-in Full base is fetched from disk when + // it has aged out of memory. In-memory samples win on a + // window-end collision (disk is a strict older suffix in steady + // state; the guard is belt-and-suspenders). + self.union_disk_parts_into(sid, start_unix_ms, end_unix_ms, &mut by_label_map); + + by_label_map .into_iter() - .map(|(label_id, samples)| { - let label_values = guard.intern.resolve(label_id).cloned().unwrap_or_default(); + .map(|(label_values, samples)| { SketchTimeSeries { sid, series_label_values: label_values, @@ -528,6 +629,193 @@ impl SketchStore { .collect() } + /// Resolve the sorted group-by KEYS for a sid from its instance + /// metadata. Disk parts store only label VALUES (a `KeyByLabelValues` + /// vector); the per-sid intern table records `BTreeMap` + /// (key-sorted), so `values()` yields values in key-sorted order. + /// Zipping the sorted `group_by_keys` against a disk values vector + /// rebuilds the exact `BTreeMap` that the in-memory + /// path produced — no part-format change needed to round-trip keys. + fn sid_group_by_keys(&self, sid: u64) -> Option> { + self.instances + .read() + .ok()? + .get(&sid) + .map(|m| m.group_by_keys.iter().cloned().collect()) + } + + /// Reconstruct the full label MAP for one disk entry by zipping the + /// sid's sorted group-by keys against the stored values vector. + fn rebuild_label_map( + keys: &[String], + label: &Option, + ) -> BTreeMap { + let mut out = BTreeMap::new(); + if let Some(kv) = label { + for (k, v) in keys.iter().zip(kv.labels.iter()) { + out.insert(k.clone(), v.clone()); + } + } + out + } + + /// Union the durable disk tier into `by_label_map` for the requested + /// `[start, end)`. No-op when persistence is disabled. Mirrors the + /// in-memory read contract: half-open overlap admits straddling + /// panes, and the delta-stitching carry-in fetches a Full base from + /// disk when it has aged out of memory. In-memory samples already in + /// `by_label_map` win on a window-end collision. + fn union_disk_parts_into( + &self, + sid: u64, + start_unix_ms: u64, + end_unix_ms: u64, + by_label_map: &mut HashMap, BTreeMap>, + ) { + let handle = { + let g = self.persistence_read.read().unwrap(); + match g.as_ref() { + Some(h) => Arc::clone(h), + None => return, + } + }; + let Some(keys) = self.sid_group_by_keys(sid) else { + return; + }; + + // ---- Overlap scan over disk parts in [start, end) ---- + let parts = handle + .manifest + .live_parts_overlapping(start_unix_ms, end_unix_ms); + for pe in &parts { + let reader = match handle.part_cache.get_or_load(pe.part_id) { + Ok(r) => r, + Err(e) => { + tracing::warn!(part_id = pe.part_id, error = %e, "sketch disk read: open part failed"); + continue; + } + }; + for rec in reader.index_records() { + if rec.agg_id != sid { + continue; + } + // Half-open overlap, matching the in-memory scan: + // `end_ts > start && start_ts < end`. + if !(rec.end_ts > start_unix_ms && rec.start_ts < end_unix_ms) { + continue; + } + let Ok(entry) = reader.load_entry(&rec) else { + continue; + }; + let label_map = Self::rebuild_label_map(&keys, &entry.label); + let sample = SketchSampleState { + bytes: entry.sketch_bytes, + encoding: tag_to_encoding(entry.encoding_tag), + }; + by_label_map + .entry(label_map) + .or_default() + .entry(rec.end_ts as i64) + // In-memory wins — only fill window-ends disk uniquely + // owns. + .or_insert(sample); + } + } + + // ---- Delta-stitching carry-in from disk ---- + // For each label whose earliest in-window sample is a Delta and + // which lacks a Full base ending before `start`, fetch the + // most-recent Full snapshot ending at/before `start-1` from disk. + if start_unix_ms == 0 { + return; + } + let before = start_unix_ms.saturating_sub(1); + let need_base: std::collections::HashSet> = by_label_map + .iter() + .filter(|(_, samples)| { + // Earliest sample is a Delta and there is no Full base + // already present at/before `start`. + let earliest_is_delta = samples + .values() + .next() + .map(|s| { + matches!( + s.encoding, + SketchEncoding::ProtoDelta | SketchEncoding::MsgpackDelta + ) + }) + .unwrap_or(false); + let has_base_before = samples + .iter() + .any(|(w_end, s)| { + *w_end < start_unix_ms as i64 + && matches!( + s.encoding, + SketchEncoding::ProtoFull | SketchEncoding::MsgpackFull + ) + }); + earliest_is_delta && !has_base_before + }) + .map(|(label_map, _)| label_map.clone()) + .collect(); + if need_base.is_empty() { + return; + } + + let carry_parts = handle.manifest.live_parts_overlapping(0, before); + // latest Full per label-map (by window-end). + let mut latest_full: HashMap, (i64, SketchSampleState)> = + HashMap::new(); + for pe in &carry_parts { + let reader = match handle.part_cache.get_or_load(pe.part_id) { + Ok(r) => r, + Err(_) => continue, + }; + for rec in reader.index_records() { + if rec.agg_id != sid || rec.end_ts > before { + continue; + } + let Ok(entry) = reader.load_entry(&rec) else { + continue; + }; + let encoding = tag_to_encoding(entry.encoding_tag); + if !matches!( + encoding, + SketchEncoding::ProtoFull | SketchEncoding::MsgpackFull + ) { + continue; + } + let label_map = Self::rebuild_label_map(&keys, &entry.label); + if !need_base.contains(&label_map) { + continue; + } + let w_end = rec.end_ts as i64; + match latest_full.get(&label_map) { + Some((prev_end, _)) if *prev_end >= w_end => {} + _ => { + latest_full.insert( + label_map, + ( + w_end, + SketchSampleState { + bytes: entry.sketch_bytes, + encoding, + }, + ), + ); + } + } + } + } + for (label_map, (w_end, state)) in latest_full { + by_label_map + .entry(label_map) + .or_default() + .entry(w_end) + .or_insert(state); + } + } + /// Range-query the ExactAgg state for ONE sid. Sister of /// [`Self::query_range`] for the exact-aggregation branch — same /// `[start, end]` semantics, but yields `Box` @@ -829,6 +1117,20 @@ impl SketchStore { self.series.len() } + /// Total count of in-memory SEALED epochs across all sids — i.e. + /// epochs sealed (pending flush) but not yet evicted to disk. `0` + /// once the flusher has drained everything. Used by tests and + /// `/runtime` diagnostics to observe the flush-then-evict loop. + pub fn list_sealed_epochs_len(&self) -> usize { + let mut n = 0usize; + for entry in self.series.iter() { + if let Ok(data) = entry.value().read() { + n += data.sealed_epochs.len(); + } + } + n + } + /// Number of registered instances (includes ghosts). pub fn instance_count(&self) -> usize { self.instances.read().unwrap().len() @@ -1213,6 +1515,19 @@ impl SketchStore { ); let part_cache = PartCache::new(parts_root.clone(), cfg.part_cache_bytes); + // Install the durable-tier read handle + seal cadence so the + // query path unions disk parts and the per-sid stores seal on + // cadence with retention-drop disabled. Done BEFORE the flusher + // starts so any series created between here and the first flush + // tick are already in persistence mode. + self.enable_persistence_mode( + cfg.seal_window_count, + Arc::new(PersistenceReadHandle { + manifest: Arc::clone(&manifest), + part_cache: part_cache.clone(), + }), + ); + let flusher = FlusherHandle::start(cfg, Arc::clone(&manifest), Arc::clone(self))?; Ok(SketchIndexPersistence { @@ -1222,6 +1537,32 @@ impl SketchStore { parts_root, }) } + + /// Switch the store into durable-tier mode: install the read handle + /// the query path uses to consult disk parts, set the per-sid seal + /// cadence, and retro-fit any already-created `SidStoreData` so they + /// seal on cadence and stop dropping aged windows (the flush-then- + /// evict loop becomes the memory bound). Idempotent. + pub fn enable_persistence_mode( + &self, + seal_window_count: usize, + read_handle: Arc, + ) { + use std::sync::atomic::Ordering; + *self.persistence_read.write().unwrap() = Some(read_handle); + self.seal_window_count + .store(seal_window_count, Ordering::Relaxed); + if seal_window_count > 0 { + // Retro-fit existing per-sid stores (e.g. series that + // ingested before persistence finished starting). + for entry in self.series.iter() { + if let Ok(mut data) = entry.value().write() { + data.seal_window_count = Some(seal_window_count); + data.persistence_enabled = true; + } + } + } + } } // ── Phase 5 M2.3.6b — EpochSource impl ────────────────────────────────────── @@ -1311,14 +1652,15 @@ impl crate::storage_engines::sketch_db::index::persistence::EpochSource for Sket }) } }); - let (type_name, bytes) = match payload { + let (type_name, encoding_tag, bytes) = match payload { AggPayload::Sketch(s) => ( sketch_kind_label .clone() .unwrap_or_else(|| "UnknownSketch".to_string()), + encoding_to_tag(s.encoding), s.bytes.clone(), ), - AggPayload::ExactAgg(p) => (p.type_name().to_string(), { + AggPayload::ExactAgg(p) => (p.type_name().to_string(), 0u8, { use asap_types::traits::SerializableToSink; p.serialize_to_bytes() }), @@ -1329,6 +1671,7 @@ impl crate::storage_engines::sketch_db::index::persistence::EpochSource for Sket end_ts: window.1, label: label_kv, sketch_type_name: type_name, + encoding_tag, sketch_bytes: bytes, }); } @@ -2106,6 +2449,258 @@ mod tests { // Empty entry collapses — policy_count drops to 0. assert_eq!(idx.policy_count(), 0); } + + // ── Durable disk-backed tier (feat/sketch-durable-tier) ───────────── + + use crate::storage_engines::sketch_db::index::persistence::EpochSource; + use crate::storage_engines::sketch_db::index::persistence::SketchStorePersistenceConfig; + + /// Metadata with a single group-by key `host`, so the disk read-back + /// path can rebuild the `{host: }` label map from the stored + /// values vector. + fn meta_with_host_key(sid: u64) -> SketchInstanceMetadata { + let mut m = meta(sid); + m.group_by_keys = ["host".to_string()].into_iter().collect(); + m + } + + fn lv_host(v: &str) -> BTreeMap { + let mut m = BTreeMap::new(); + m.insert("host".to_string(), v.to_string()); + m + } + + /// Aggressive persistence config: seal every window, force-flush + /// everything (hot_window=0), tiny flush interval, no disk TTL, small + /// part cache. Memory limit high so the seal/flush is driven by the + /// hot-window watermark, not memory pressure — keeps the test + /// deterministic. + fn durable_cfg(disk_path: std::path::PathBuf) -> SketchStorePersistenceConfig { + SketchStorePersistenceConfig { + memory_limit_bytes: 1 << 30, + memory_low_watermark_bytes: 1 << 29, + hard_cap_bytes: 1 << 31, + hot_window_ms: Some(0), // every sealed epoch is "old" → flush now + delete_older_than_ms: None, + flush_interval: std::time::Duration::from_millis(5), + disk_path, + part_cache_bytes: 1 << 20, + seal_window_count: 1, // seal on every distinct window + } + } + + fn wait_until bool>(f: F, timeout: std::time::Duration) -> bool { + let deadline = std::time::Instant::now() + timeout; + while std::time::Instant::now() < deadline { + if f() { + return true; + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + f() + } + + #[test] + fn sealing_fires_under_persistence() { + let tmp = tempfile::TempDir::new().unwrap(); + let idx = Arc::new(SketchStore::new()); + idx.register(meta_with_host_key(101)); + let _p = idx.start_persistence(durable_cfg(tmp.path().to_path_buf())).unwrap(); + + // seal_window_count = 1: each *new distinct window* seals the + // prior one. Append several distinct windows for one series. + for i in 0..5u64 { + let s = i * 30_000; + idx.append_sample(101, lv_host("a"), (s, s + 30_000), sample(i as u8)); + } + // At least some sealed epochs must exist (each new window seals + // the prior current_epoch). The flusher may evict some before we + // look, so we assert that sealing happened OR a part landed. + let sealed_now = !idx.list_sealed_epochs().is_empty(); + let flushed = wait_until(|| !_p.manifest.live_parts().is_empty(), std::time::Duration::from_secs(3)); + assert!( + sealed_now || flushed, + "no epochs sealed and nothing flushed — sealing did not fire under persistence" + ); + } + + #[test] + fn sealed_epochs_flush_to_disk_and_memory_drops() { + let tmp = tempfile::TempDir::new().unwrap(); + let idx = Arc::new(SketchStore::new()); + idx.register(meta_with_host_key(202)); + let p = idx.start_persistence(durable_cfg(tmp.path().to_path_buf())).unwrap(); + + for i in 0..10u64 { + let s = i * 30_000; + idx.append_sample(202, lv_host("a"), (s, s + 30_000), sample(i as u8)); + } + + // The flusher should drain sealed epochs to disk; memory + // (sealed-epoch bytes) drops to ~0 and parts appear. + let drained = wait_until( + || idx.approx_memory_bytes() == 0 && !p.manifest.live_parts().is_empty(), + std::time::Duration::from_secs(5), + ); + assert!( + drained, + "flush+evict did not bound memory: sealed_bytes={}, parts={}", + idx.approx_memory_bytes(), + p.manifest.live_parts().len() + ); + } + + #[test] + fn query_resolves_from_disk_after_flush_evict() { + let tmp = tempfile::TempDir::new().unwrap(); + let idx = Arc::new(SketchStore::new()); + idx.register(meta_with_host_key(303)); + let p = idx.start_persistence(durable_cfg(tmp.path().to_path_buf())).unwrap(); + + // Append windows for series "a" across [0, 300_000). + for i in 0..10u64 { + let s = i * 30_000; + idx.append_sample(303, lv_host("a"), (s, s + 30_000), sample((i + 1) as u8)); + } + // Wait for everything to flush+evict from memory. + assert!( + wait_until( + || idx.approx_memory_bytes() == 0 && idx.list_sealed_epochs_len() == 0, + std::time::Duration::from_secs(5) + ), + "data never fully evicted from memory" + ); + // current_epoch may still hold the most-recent un-sealed window; + // query a range covering the EVICTED portion [0, 150_000). + let series = idx.query_range(303, 0, 150_000); + assert_eq!(series.len(), 1, "expected one series resolved from disk"); + let s = &series[0]; + assert_eq!(s.series_label_values, lv_host("a"), "label map rebuilt from disk"); + assert!( + !s.samples.is_empty(), + "query over evicted range returned no samples from disk" + ); + // Window-end 30_000 (window (0,30_000)) must be present from disk. + assert!( + s.samples.contains_key(&30_000), + "disk window (0,30000) missing from query result: {:?}", + s.samples.keys().collect::>() + ); + drop(p); + } + + #[test] + fn query_carry_in_full_base_lives_on_disk() { + let tmp = tempfile::TempDir::new().unwrap(); + let idx = Arc::new(SketchStore::new()); + idx.register(meta_with_host_key(404)); + let p = idx.start_persistence(durable_cfg(tmp.path().to_path_buf())).unwrap(); + + // A Full snapshot early (end=100_000), then delta windows later. + idx.append_sample(404, lv_host("a"), (70_000, 100_000), sample(1)); // Full base + for i in 0..6u64 { + let s = 100_000 + i * 30_000; + idx.append_sample(404, lv_host("a"), (s, s + 30_000), delta_sample((i + 2) as u8)); + } + // Flush+evict everything to disk. + assert!( + wait_until( + || idx.approx_memory_bytes() == 0 && idx.list_sealed_epochs_len() == 0, + std::time::Duration::from_secs(5) + ), + "data never fully evicted" + ); + + // Query a window that contains ONLY deltas; the Full base lives on + // disk before the window. The carry-in must splice it in. + let series = idx.query_range(404, 200_000, 280_000); + assert_eq!(series.len(), 1); + let s = &series[0]; + // A Full-encoded carry-in base (end < 200_000) must be present. + let has_full_base = s.samples.iter().any(|(w_end, smp)| { + *w_end < 200_000 + && matches!( + smp.encoding, + SketchEncoding::ProtoFull | SketchEncoding::MsgpackFull + ) + }); + assert!( + has_full_base, + "delta-only window did not get a disk-resident Full carry-in base: {:?}", + s.samples + .iter() + .map(|(k, v)| (*k, v.encoding)) + .collect::>() + ); + drop(p); + } + + #[test] + fn restart_recovery_makes_flushed_data_queryable() { + let tmp = tempfile::TempDir::new().unwrap(); + let disk = tmp.path().to_path_buf(); + { + let idx = Arc::new(SketchStore::new()); + idx.register(meta_with_host_key(505)); + let p = idx.start_persistence(durable_cfg(disk.clone())).unwrap(); + for i in 0..8u64 { + let s = i * 30_000; + idx.append_sample(505, lv_host("a"), (s, s + 30_000), sample((i + 1) as u8)); + } + assert!( + wait_until( + || !p.manifest.live_parts().is_empty() + && idx.list_sealed_epochs_len() == 0 + && idx.approx_memory_bytes() == 0, + std::time::Duration::from_secs(5) + ), + "data never flushed before restart" + ); + // Shutdown the flusher cleanly so the manifest is durable. + let mut p = p; + p.shutdown(); + } + + // "Restart": brand-new store + resolver on the SAME disk dir. The + // metadata is re-registered (the SeriesIdResolver WAL recovers + // sids in prod; here we re-register to model that), then + // persistence recovers the manifest+parts. + let idx2 = Arc::new(SketchStore::new()); + idx2.register(meta_with_host_key(505)); + let p2 = idx2.start_persistence(durable_cfg(disk.clone())).unwrap(); + assert!( + !p2.manifest.live_parts().is_empty(), + "recovery did not reload any parts" + ); + + let series = idx2.query_range(505, 0, 120_000); + assert_eq!(series.len(), 1, "recovered data not queryable"); + let s = &series[0]; + assert_eq!(s.series_label_values, lv_host("a")); + assert!( + s.samples.contains_key(&30_000), + "recovered disk window missing after restart" + ); + drop(p2); + } + + #[test] + fn persistence_disabled_keeps_327_retention_behavior() { + // Non-regression: with persistence OFF, query_range reads + // in-memory only and #327 retention still bounds memory. The + // #323–#326 overlap + carry-in shapes still pass (covered by the + // dedicated tests above); here we confirm the disk union is a + // no-op when no read handle is installed. + let idx = SketchStore::new(); + idx.register(meta_with_host_key(606)); + idx.append_sample(606, lv_host("a"), (0, 10), sample(1)); + idx.append_sample(606, lv_host("a"), (10, 20), sample(2)); + let series = idx.query_range(606, 0, 20); + assert_eq!(series.len(), 1); + assert_eq!(series[0].samples.len(), 2); + // No persistence handle → seal cadence disabled → no sealing. + assert!(idx.persistence_read.read().unwrap().is_none()); + } } // 2026-05 reorg: generic epoch-partitioned columnar storage lives diff --git a/data_plane/src/storage_engines/sketch_db/persistence/config.rs b/data_plane/src/storage_engines/sketch_db/persistence/config.rs index bb4d456a..13999a1a 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/config.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/config.rs @@ -48,6 +48,22 @@ pub struct SketchStorePersistenceConfig { /// query pays disk I/O). Default is `min(10% * memory_limit_bytes, /// 512 MiB)` — scale it with the write budget, not a fixed number. pub part_cache_bytes: u64, + + /// Seal cadence in DISTINCT WINDOWS. The per-sid hot `current_epoch` + /// is sealed into the (in-memory, pending-flush) sealed-epoch ring + /// once it accumulates this many distinct windows; the background + /// flusher then turns sealed epochs into durable disk parts and + /// evicts them. This is what makes sealing fire in production — the + /// in-memory-only deployment never seals (`epoch_capacity == None`). + /// + /// Sizing: the agent emits ~30s tumbling panes, so `20` windows is + /// ~10 min of one series per part — large enough to amortize the + /// per-part header/index overhead, small enough that the most-recent + /// fully-behind-`hot_window` data is actually sealed (and thus + /// flushable) rather than stuck un-sealed in `current_epoch`. + /// `0` disables cadence sealing (no durable tier even if a disk_path + /// is set). + pub seal_window_count: usize, } impl SketchStorePersistenceConfig { @@ -65,6 +81,7 @@ impl SketchStorePersistenceConfig { flush_interval: Duration::from_secs(1), disk_path, part_cache_bytes: cache, + seal_window_count: 20, // ~10 min of 30s panes per part } } } diff --git a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs index d1bb09e4..728f9521 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs @@ -506,6 +506,7 @@ mod tests { end_ts: max_ts, label: Some(KeyByLabelValues::new_with_labels(vec!["host".into()])), sketch_type_name: "SumAccumulator".into(), + encoding_tag: 0, sketch_bytes: b"dummy-payload".to_vec(), }], } @@ -521,6 +522,7 @@ mod tests { flush_interval: Duration::from_millis(10), disk_path, part_cache_bytes: 0, + seal_window_count: 20, } } diff --git a/data_plane/src/storage_engines/sketch_db/persistence/part.rs b/data_plane/src/storage_engines/sketch_db/persistence/part.rs index c9743d7c..b41d7b09 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/part.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/part.rs @@ -91,6 +91,22 @@ pub fn part_dir_path(parts_root: &Path, part_id: PartId) -> PathBuf { parts_root.join(part_dir_name(part_id)) } +/// Encoding-tag namespace for the per-entry `encoding` byte stored in +/// `data.bin` (the byte that v1 left as a reserved pad). Lets the disk +/// read-back path reconstruct the `SketchEncoding` so the +/// delta-stitching carry-in survives the in-mem/on-disk boundary. +/// +/// `0` is reserved for "unknown / treat as Full" so parts written by +/// the original v1 writer (which always wrote `0` into the pad) decode +/// as Full — the safe default for a carry-in base. +pub mod encoding_tag { + pub const UNKNOWN: u8 = 0; + pub const PROTO_FULL: u8 = 1; + pub const PROTO_DELTA: u8 = 2; + pub const MSGPACK_FULL: u8 = 3; + pub const MSGPACK_DELTA: u8 = 4; +} + /// One entry inside a decoded part. The `start_ts`/`end_ts`/`label` /// fields are resolved by the reader; the sketch payload stays as /// bytes so the query path can decide whether to decode lazily. @@ -101,6 +117,9 @@ pub struct SnapshotEntry { pub end_ts: u64, pub label: Option, pub sketch_type_name: String, + /// Wire-encoding tag — see [`encoding_tag`]. `0` for parts written + /// before encoding round-tripping landed (decode as Full). + pub encoding_tag: u8, pub sketch_bytes: Vec, } @@ -159,6 +178,7 @@ impl PartWriter { data_offset, label_bytes, type_name_bytes, + encoding_tag: e.encoding_tag, sketch_bytes: e.sketch_bytes.clone(), }); data_len += entry_size as u64; @@ -193,7 +213,11 @@ impl PartWriter { &mut data_crc, pe.type_name_bytes.len() as u16, )?; - write_u16(&mut data_file, &mut data_crc, 0)?; // _pad + // Repurposed v1 pad u16: low byte carries the encoding tag, + // high byte stays zero. v1 parts wrote 0 here → decode as + // `encoding_tag::UNKNOWN` (treat as Full), so old parts + // remain readable. + write_u16(&mut data_file, &mut data_crc, pe.encoding_tag as u16)?; write_u32(&mut data_file, &mut data_crc, 0)?; // _pad write_padded(&mut data_file, &mut data_crc, &pe.label_bytes, 8)?; write_padded(&mut data_file, &mut data_crc, &pe.type_name_bytes, 8)?; @@ -314,6 +338,7 @@ struct PlannedEntry { data_offset: u64, label_bytes: Vec, type_name_bytes: Vec, + encoding_tag: u8, sketch_bytes: Vec, } @@ -551,7 +576,8 @@ impl PartReader { u32::from_le_bytes(self.data_mmap[off + 4..off + 8].try_into().unwrap()) as usize; let type_name_len = u16::from_le_bytes(self.data_mmap[off + 8..off + 10].try_into().unwrap()) as usize; - // 10..12 pad, 12..16 pad + // 10..12 repurposed pad: low byte = encoding tag; 12..16 pad. + let encoding_tag = self.data_mmap[off + 10]; let mut cursor = off + 16; let label_padded = align_up(label_len, 8); let label_bytes = &self.data_mmap[cursor..cursor + label_len]; @@ -577,6 +603,7 @@ impl PartReader { end_ts: rec.end_ts, label, sketch_type_name: type_name, + encoding_tag, sketch_bytes, }) } @@ -613,6 +640,7 @@ mod tests { "api".into(), ])), sketch_type_name: "SumAccumulator".into(), + encoding_tag: encoding_tag::PROTO_FULL, sketch_bytes: b"opaque-sketch-1".to_vec(), }, EpochSnapshotEntry { @@ -620,6 +648,7 @@ mod tests { end_ts: 2_000, label: None, sketch_type_name: "DatasketchesKLLAccumulator".into(), + encoding_tag: encoding_tag::MSGPACK_DELTA, sketch_bytes: b"opaque-sketch-2-more-bytes".to_vec(), }, ], @@ -655,6 +684,7 @@ mod tests { let e0 = reader.load_entry(&recs[0]).expect("load_entry 0"); assert_eq!(e0.sketch_type_name, "SumAccumulator"); + assert_eq!(e0.encoding_tag, encoding_tag::PROTO_FULL); assert_eq!(e0.sketch_bytes, b"opaque-sketch-1"); assert_eq!( e0.label.as_ref().unwrap().labels, @@ -664,6 +694,7 @@ mod tests { let e1 = reader.load_entry(&recs[1]).expect("load_entry 1"); assert!(e1.label.is_none()); assert_eq!(e1.sketch_type_name, "DatasketchesKLLAccumulator"); + assert_eq!(e1.encoding_tag, encoding_tag::MSGPACK_DELTA); assert_eq!(e1.sketch_bytes, b"opaque-sketch-2-more-bytes"); } diff --git a/data_plane/src/storage_engines/sketch_db/persistence/recovery.rs b/data_plane/src/storage_engines/sketch_db/persistence/recovery.rs index 7810d90a..0428027f 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/recovery.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/recovery.rs @@ -138,6 +138,7 @@ mod tests { end_ts: 200, label: Some(KeyByLabelValues::new_with_labels(vec!["x".into()])), sketch_type_name: "SumAccumulator".into(), + encoding_tag: 0, sketch_bytes: b"payload".to_vec(), }], } diff --git a/data_plane/src/storage_engines/sketch_db/persistence/source.rs b/data_plane/src/storage_engines/sketch_db/persistence/source.rs index b993fae5..c1c4da85 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/source.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/source.rs @@ -67,6 +67,14 @@ pub struct EpochSnapshotEntry { /// so a future read-back path can dispatch to the right /// deserializer once the SketchStore-backed snapshot lands. pub sketch_type_name: String, + /// Wire-encoding tag for sketch payloads — see + /// [`encoding_tag`](crate::storage_engines::sketch_db::index::persistence::part::encoding_tag). + /// Carries the `SketchEncoding` (Full vs Delta) of the payload so + /// the disk read-back path can preserve the delta-stitching + /// carry-in semantics across the in-mem/on-disk boundary. `0` + /// (the default) means "unknown / treat as Full" — exact-agg + /// payloads use `0`. + pub encoding_tag: u8, /// Serialized sketch payload (opaque to the persistence layer). pub sketch_bytes: Vec, }