data-plane: honest SketchStore memory diagnostic + idle-sid eviction

data_plane/src/main.rs

@@ -196,6 +196,18 @@ struct Args {
     #[arg(long)]
     schema_eviction_dry_run: bool,
 
+    /// Idle-sid eviction (memory reclaim). Drop the in-memory state of any
+    /// sketch sid with no writes for this many seconds AND whose state is
+    /// fully flushed to disk, keeping its queryable metadata — the series
+    /// stays answerable from the durable tier and rehydrates on the next
+    /// write. Bounds resident registry memory when series churn / go stale
+    /// (without it, stale sketch sids are pinned in RAM until config-driven
+    /// retirement). 0 disables. Effective horizon is
+    /// max(this, --persistence-hot-window-secs), since eviction waits for
+    /// the sid's windows to seal+flush first.
+    #[arg(long, env = "ASAP_IDLE_SID_EVICT_SECS", default_value = "0")]
+    idle_sid_evict_secs: u64,
+
     // ---- SketchStore persistence ----
     //
     // When --persistence-enabled is set, the store is constructed via
@@ -502,6 +514,36 @@ async fn main() -> Result<()> {
     // via the shared `SketchStore` (already passed in above).
     let engine = Arc::new(engine);
 
+    // Idle-sid eviction sweep (memory reclaim) — opt-in via
+    // --idle-sid-evict-secs. Drops the in-memory `SidStoreData` for
+    // write-idle, fully-flushed sketch sids while keeping their queryable
+    // metadata, bounding resident registry memory under series churn.
+    if args.idle_sid_evict_secs > 0 {
+        let evict_index = sketch_index.clone();
+        let idle_ms = args.idle_sid_evict_secs.saturating_mul(1000);
+        // Sweep a few times per idle horizon, clamped to a sane cadence.
+        let sweep = std::time::Duration::from_secs(args.idle_sid_evict_secs.clamp(10, 60));
+        info!(
+            "Idle-sid eviction enabled: idle threshold {}s, sweep every {}s",
+            args.idle_sid_evict_secs,
+            sweep.as_secs()
+        );
+        tokio::spawn(async move {
+            let mut interval = tokio::time::interval(sweep);
+            loop {
+                interval.tick().await;
+                let n = evict_index.evict_idle_series(idle_ms);
+                if n > 0 {
+                    info!(
+                        "[IDLE_EVICT] evicted {} idle sid(s) from memory \
+                         (still queryable from disk; rehydrate on next write)",
+                        n
+                    );
+                }
+            }
+        });
+    }
+
     // Setup OTLP receiver (after precompute engine so it can share the ingest state)
     // Issue #46 ⑥ — freshness-probe last-value cache. Shared between
     // the OTLP receiver (write path) and the HTTP query handler (read
@@ -833,6 +875,23 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
+/// Best-effort process resident-set size (RSS) in bytes, read from
+/// `/proc/self/statm` (field 2 = resident pages × page size). Returns 0 if
+/// unreadable (non-Linux / sandboxed) so the diagnostic degrades gracefully
+/// rather than failing. This is the ground-truth counterpart to the
+/// store's structural estimates in the memory diagnostic.
+fn process_resident_bytes() -> usize {
+    let Ok(statm) = std::fs::read_to_string("/proc/self/statm") else {
+        return 0;
+    };
+    let Some(resident_pages) = statm.split_whitespace().nth(1) else {
+        return 0;
+    };
+    let pages: usize = resident_pages.parse().unwrap_or(0);
+    // `sysconf(_SC_PAGESIZE)` is 4 KiB on every platform this runs on.
+    pages * 4096
+}
+
 /// Periodic memory diagnostics logger — runs every 30 seconds.
 async fn spawn_memory_diagnostics(
     sketch_index: Arc<data_plane::storage_engines::sketch_db::index::SketchStore>,
@@ -849,12 +908,26 @@ async fn spawn_memory_diagnostics(
         //    pre-M2.3 per-agg_id SketchStore::diagnostic_info).
         let instance_count = sketch_index.instance_count();
         let series_count = sketch_index.series_len();
-        let approx_bytes = sketch_index.approx_memory_bytes();
+        // `approx_memory_bytes` is the flusher's EVICTABLE-payload gauge:
+        // it counts only live sketch payloads (current_epoch + sealed), so
+        // it correctly reads ~0 once everything has been flushed to disk.
+        // On its own it badly misrepresents the store's footprint — the
+        // per-sid registry + intern caches stay resident and are not
+        // flushable. Report all three: evictable payload, the structural
+        // resident estimate, and the process RSS ground truth.
+        let payload_bytes = sketch_index.approx_memory_bytes();
+        let resident_bytes = sketch_index.approx_resident_bytes();
+        let rss_bytes = process_resident_bytes();
         info!(
-            "[MEMORY_DIAG] SketchStore: {} instance(s), {} sid(s) with state, {:.2} KB approx in-memory bytes (hot current_epoch + sealed)",
+            "[MEMORY_DIAG] SketchStore: {} instance(s), {} sid(s) with state, \
+             payload={:.2} KB (evictable, flusher gauge), \
+             registry+intern\u{2248}{:.2} MB (resident, not flushable), \
+             process RSS={:.1} MB",
             instance_count,
             series_count,
-            approx_bytes as f64 / 1024.0,
+            payload_bytes as f64 / 1024.0,
+            resident_bytes as f64 / (1024.0 * 1024.0),
+            rss_bytes as f64 / (1024.0 * 1024.0),
         );
 
         // 2. Worker diagnostics (precompute engine only)

data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs

@@ -104,6 +104,37 @@ impl<K: Eq + std::hash::Hash + Clone> Default for InternTable<K> {
     }
 }
 
+impl InternTable<BTreeMap<String, String>> {
+    /// Approximate RESIDENT heap bytes of the interned label maps for the
+    /// concrete `BTreeMap<String, String>` key the SketchStore uses.
+    ///
+    /// Both `id_to_label` (the `Vec`) and `label_to_id` (the `HashMap`)
+    /// retain a clone of every interned key, so each distinct label map is
+    /// counted twice, plus backing-store slot capacity. This is the
+    /// dominant per-sid resident cost once sketch payloads have been
+    /// flushed to disk — and it is exactly the cost the payload-only
+    /// [`crate::storage_engines::sketch_db::index::persistence::EpochSource::approx_memory_bytes`]
+    /// (the flusher's eviction gauge) does NOT see, which is why the memory
+    /// diagnostic read ~0 KB while RSS sat in the hundreds of MB.
+    pub fn approx_heap_bytes(&self) -> usize {
+        let mut key_bytes = 0usize;
+        for m in &self.id_to_label {
+            for (k, v) in m.iter() {
+                // string bytes + the two `String` headers + a BTree node.
+                key_bytes += k.len() + v.len() + 2 * std::mem::size_of::<String>() + 32;
+            }
+            key_bytes += std::mem::size_of::<BTreeMap<String, String>>();
+        }
+        // ×2 for the cloned copy held by `label_to_id`, plus the backing
+        // Vec / HashMap slot capacity.
+        key_bytes * 2
+            + self.id_to_label.capacity() * std::mem::size_of::<BTreeMap<String, String>>()
+            + self.label_to_id.capacity()
+                * (std::mem::size_of::<BTreeMap<String, String>>()
+                    + std::mem::size_of::<LabelValuesId>())
+    }
+}
+
 /// Active (mutable) epoch: append-only insert, O(1) amortized.
 ///
 /// Three parallel columns (Opt 5) — windows / label-id / payload —
@@ -816,6 +847,15 @@ pub struct SidStoreData<K: Eq + std::hash::Hash + Clone, P> {
     /// 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,
+    /// Wall-clock millis of the most recent write (append) into this sid.
+    /// `0` means "never written" / freshly (re)hydrated. Drives idle-sid
+    /// eviction: a sid with no writes for the idle threshold whose state
+    /// is fully durable on disk can have this whole `SidStoreData` dropped
+    /// from memory while its queryable `SketchInstanceMetadata` is kept
+    /// (the series stays answerable from the disk tier and rehydrates on
+    /// the next write). Updated under the per-sid write lock the append
+    /// path already holds, so it costs nothing extra on the hot path.
+    pub last_write_unix_ms: u64,
 }
 
 /// Default in-memory retention horizon (ms) for the WARM sketch store.
@@ -858,6 +898,7 @@ impl<K: Eq + std::hash::Hash + Clone, P> SidStoreData<K, P> {
             retention_horizon_ms: default_retention_horizon_ms(),
             seal_window_count: None,
             persistence_enabled: false,
+            last_write_unix_ms: 0,
         }
     }