reddb_server/storage/cache/

sweeper.rs

//! Bounded Blob Cache maintenance for L1 expiry and namespace flushing.
//!
//! `sweep_expired` consumes due records from per-shard L1 indexes and the
//! store's persisted ordered L2 deadline projection, so ordinary TTL work
//! depends on the number of deadlines due rather than the total cache size. A
//! separate rotating scan runs only while a namespace generation flush needs
//! physical reclamation.
//! The runtime invokes it every 250 ms on a dedicated maintenance thread.
//!
//! # Operations
//!
//! - [`BlobCacheSweeper::sweep_expired`] — bounded sweep of expired L1/L2 entries.
//! - [`BlobCacheSweeper::reclaim_orphans`] — bounded reclamation of L2 blob
//!   chains. This operation remains scaffolding until the L2 allocation walk
//!   is exposed; it reports zero work rather than claiming reclamation.
//! - [`BlobCacheSweeper::flush_namespace`] — foreground-fast namespace flush.
//!   O(1): bumps the per-namespace generation counter; physical reclamation
//!   happens lazily on next access or via [`BlobCacheSweeper::sweep_expired`].
//!
//! # Bounding
//!
//! Sweep operations are bounded by [`SweepLimit`]:
//!
//! - `Entries(N)` — hard cap on the number of entries scanned.
//! - `Millis(N)` — hard cap on wall-clock time. Checked at every iteration so
//!   the cap is honored within a few microseconds of overrun.
//! - `Either { entries, millis }` — first cap to fire wins.
//!
//! When a sweep terminates because it hit a limit (instead of running to
//! completion) the report's `truncated_due_to_limit` flag is set so admin
//! callers can decide whether to schedule a follow-up sweep.
//!
//! # Concurrency contract
//!
//! Operations are safe to call while concurrent readers
//! (`BlobCache::get`, `BlobCache::exists`) and writers
//! (`BlobCache::put`) are in flight:
//!
//! - The expiry sweep takes one shard write lock at a time and releases it
//!   before advancing. Readers touching other shards continue normally.
//! - `flush_namespace` only bumps a generation counter under a brief
//!   write-lock. Concurrent reads against the same namespace either see the
//!   old generation (returning a hit if the entry is still alive) or the new
//!   generation (treating any cached entry as stale). Either is correct.
//! - Each invocation has entry/time bounds and never holds a global lock.

use std::time::Instant;

use super::blob::BlobCache;

// ---------------------------------------------------------------------------
// Public types
// ---------------------------------------------------------------------------

/// Bound for a single sweeper invocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SweepLimit {
    /// Hard cap on deadline/generation records inspected.
    Entries(usize),
    /// Hard cap on wall-clock milliseconds. The sweep stops as soon as the
    /// elapsed time crosses this threshold.
    Millis(u32),
    /// First-wins composite. Stops when either bound is hit.
    Either { entries: usize, millis: u32 },
}

impl SweepLimit {
    fn entries_cap(self) -> Option<usize> {
        match self {
            SweepLimit::Entries(n) => Some(n),
            SweepLimit::Either { entries, .. } => Some(entries),
            SweepLimit::Millis(_) => None,
        }
    }

    fn millis_cap(self) -> Option<u32> {
        match self {
            SweepLimit::Millis(n) => Some(n),
            SweepLimit::Either { millis, .. } => Some(millis),
            SweepLimit::Entries(_) => None,
        }
    }
}

/// Per-namespace breakdown of a [`SweepReport`].
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct NamespaceSweepStats {
    pub entries_scanned: usize,
    pub entries_evicted: usize,
    pub bytes_reclaimed: u64,
}

/// Outcome of [`BlobCacheSweeper::sweep_expired`].
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SweepReport {
    pub entries_scanned: usize,
    pub entries_evicted: usize,
    pub bytes_reclaimed: u64,
    pub elapsed_ms: u32,
    /// `true` iff the sweep terminated because it hit [`SweepLimit`] before
    /// finishing the full set of candidates. Admin callers should treat this
    /// as "schedule another sweep soon".
    pub truncated_due_to_limit: bool,
    pub by_namespace: Vec<(String, NamespaceSweepStats)>,
}

/// Outcome of [`BlobCacheSweeper::reclaim_orphans`].
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct OrphanReport {
    pub blob_chains_scanned: usize,
    pub blob_chains_reclaimed: usize,
    pub bytes_reclaimed: u64,
    pub elapsed_ms: u32,
    pub truncated_due_to_limit: bool,
}

/// Outcome of [`BlobCacheSweeper::flush_namespace`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NamespaceFlushReport {
    pub namespace: String,
    pub generation_before: u64,
    pub generation_after: u64,
    /// Foreground-fast contract target: <100 µs typical.
    pub elapsed_micros: u32,
}

// ---------------------------------------------------------------------------
// Sweeper
// ---------------------------------------------------------------------------

/// Stateless namespace for sweeper operations against a [`BlobCache`].
///
/// The sweeper holds no state of its own; every method is a free-function
/// over `&BlobCache`. This keeps the surface easy to call from admin
/// handlers, the runtime scheduler, and tests without owning a sweeper
/// instance.
pub struct BlobCacheSweeper;

impl BlobCacheSweeper {
    /// Bounded sweep of expired L1 and L2 entries.
    ///
    /// Pops due L1 deadlines across shards, revalidates them against the
    /// current entry, and evicts only entries whose effective expiry elapsed.
    /// Honors `limit` precisely.
    ///
    /// # Concurrency
    ///
    /// One shard write lock is held at a time. A read on that shard may wait
    /// briefly; readers on other shards continue unimpeded. Persistent L2
    /// deletion is published as one bounded WAL batch after the L1 locks are
    /// released.
    ///
    pub fn sweep_expired(cache: &BlobCache, limit: SweepLimit) -> SweepReport {
        let started = Instant::now();
        let mut outcome = cache.sweep_expired_l1(limit.entries_cap(), limit.millis_cap());
        let entries_remaining = limit.entries_cap().map_or(usize::MAX, |cap| {
            cap.saturating_sub(outcome.entries_scanned)
        });
        let millis_remaining = limit.millis_cap().map(|cap| {
            let elapsed = u32::try_from(started.elapsed().as_millis()).unwrap_or(u32::MAX);
            cap.saturating_sub(elapsed)
        });
        let l2 = cache.sweep_expired_l2(entries_remaining, millis_remaining);
        outcome.entries_scanned = outcome.entries_scanned.saturating_add(l2.entries_scanned);
        outcome.entries_evicted = outcome.entries_evicted.saturating_add(l2.entries_evicted);
        outcome.expirations = outcome.expirations.saturating_add(l2.expirations);
        outcome.bytes_reclaimed = outcome.bytes_reclaimed.saturating_add(l2.bytes_reclaimed);
        outcome.completed &= l2.completed;
        outcome.elapsed_ms = u32::try_from(started.elapsed().as_millis()).unwrap_or(u32::MAX);
        for (namespace, stats) in l2.by_namespace {
            let aggregate = outcome.by_namespace.entry(namespace).or_default();
            aggregate.entries_scanned = aggregate
                .entries_scanned
                .saturating_add(stats.entries_scanned);
            aggregate.entries_evicted = aggregate
                .entries_evicted
                .saturating_add(stats.entries_evicted);
            aggregate.bytes_reclaimed = aggregate
                .bytes_reclaimed
                .saturating_add(stats.bytes_reclaimed);
        }
        SweepReport {
            entries_scanned: outcome.entries_scanned,
            entries_evicted: outcome.entries_evicted,
            bytes_reclaimed: outcome.bytes_reclaimed,
            elapsed_ms: outcome.elapsed_ms,
            truncated_due_to_limit: !outcome.completed,
            by_namespace: outcome
                .by_namespace
                .into_iter()
                .map(|(namespace, stats)| {
                    (
                        namespace,
                        NamespaceSweepStats {
                            entries_scanned: stats.entries_scanned,
                            entries_evicted: stats.entries_evicted,
                            bytes_reclaimed: stats.bytes_reclaimed,
                        },
                    )
                })
                .collect(),
        }
    }

    /// Bounded reclamation of L2 orphan blob chains.
    ///
    /// Orphan chains arise when a writer flushes the blob bytes to L2 pages
    /// but is killed before the metadata B+ tree commit (see the WAL-ordering
    /// note in `docs/perf/blob-cache-l2-spike.md`). Recovery cannot tell
    /// these pages apart from a successful write without cross-checking the
    /// metadata catalog, so they accumulate as wasted L2 capacity until the
    /// sweeper reclaims them.
    ///
    /// The algorithm (once accessor #2 lands):
    /// 1. Walk the L2 free-list of allocated blob chains
    ///    (`cache.l2_blob_chains()` — to-be-added).
    /// 2. For each chain root, look up the metadata B+ tree
    ///    (`cache.for_each_l2_record`) and check whether any record points
    ///    at this root.
    /// 3. Chains with no metadata reference are orphans — free their pages
    ///    via the L2 API and accumulate `bytes_reclaimed`.
    ///
    /// # Concurrency
    ///
    /// Same contract as [`BlobCacheSweeper::sweep_expired`].
    pub fn reclaim_orphans(cache: &BlobCache, limit: SweepLimit) -> OrphanReport {
        let started = Instant::now();
        let budget = Budget::new(limit, started);

        // Same "observable interaction" rationale as in `sweep_expired`.
        let _ = cache.stats();

        let mut report = OrphanReport::default();

        // Bounded scaffolding: real L2 chain walk awaits flag #2.
        budget.observe(&mut report.elapsed_ms);
        report.truncated_due_to_limit = false;
        report
    }

    /// Foreground-fast namespace flush.
    ///
    /// Delegates to [`BlobCache::invalidate_namespace`], which is O(1): it
    /// only bumps a per-namespace generation counter. Cached entries with
    /// the old generation become invisible immediately and are physically
    /// removed by later cache access or by [`BlobCacheSweeper::sweep_expired`].
    ///
    /// # Foreground-fast contract
    ///
    /// Returns within ~100 µs typical. The `elapsed_micros` field on the
    /// returned [`NamespaceFlushReport`] makes the contract observable so
    /// admin endpoints can alert on regressions.
    ///
    /// # Generation reporting
    ///
    /// `generation_before` and `generation_after` are reported on a
    /// best-effort basis. Because `BlobCache` does not expose a public
    /// generation accessor, the sweeper reports them as `(0, 0)` when the
    /// flush call signals "namespace did not exist" and `(0, 0)` when it
    /// signals success — placeholder values until a `current_generation`
    /// public accessor is added (see flag #2). The `namespace` and
    /// `elapsed_micros` fields are accurate today.
    pub fn flush_namespace(cache: &BlobCache, namespace: &str) -> NamespaceFlushReport {
        let started = Instant::now();
        let generation_before = cache.current_generation(namespace);
        let _flushed = cache.invalidate_namespace(namespace);
        let generation_after = cache.current_generation(namespace);
        let elapsed = started.elapsed();
        // Saturate at u32::MAX; under the foreground-fast contract this
        // bound is never approached.
        let elapsed_micros = u32::try_from(elapsed.as_micros()).unwrap_or(u32::MAX);
        NamespaceFlushReport {
            namespace: namespace.to_string(),
            generation_before,
            generation_after,
            elapsed_micros,
        }
    }
}

// ---------------------------------------------------------------------------
// Budget — internal bounded-work accounting
// ---------------------------------------------------------------------------

/// Internal helper: tracks elapsed time and entries-scanned against a
/// [`SweepLimit`]. Centralises the "first-bound-wins" logic so the three
/// public sweeper entry-points stay short.
struct Budget {
    started: Instant,
    entries_cap: Option<usize>,
    millis_cap: Option<u32>,
    entries_seen: usize,
}

impl Budget {
    fn new(limit: SweepLimit, started: Instant) -> Self {
        Self {
            started,
            entries_cap: limit.entries_cap(),
            millis_cap: limit.millis_cap(),
            entries_seen: 0,
        }
    }

    /// Returns `true` if either bound has been crossed.
    #[allow(dead_code)] // Used by the not-yet-wired walk loops; keep for clarity.
    fn exhausted(&self) -> bool {
        if let Some(cap) = self.entries_cap {
            if self.entries_seen >= cap {
                return true;
            }
        }
        if let Some(cap) = self.millis_cap {
            if self.elapsed_ms_capped() >= cap {
                return true;
            }
        }
        false
    }

    /// Records that one more entry was scanned. Returns the post-tick
    /// `exhausted()` result.
    #[allow(dead_code)] // Used by the not-yet-wired walk loops; keep for clarity.
    fn tick(&mut self) -> bool {
        self.entries_seen = self.entries_seen.saturating_add(1);
        self.exhausted()
    }

    fn elapsed_ms_capped(&self) -> u32 {
        u32::try_from(self.started.elapsed().as_millis()).unwrap_or(u32::MAX)
    }

    /// Stamps the elapsed-ms field of a report at the end of a sweep.
    fn observe(self, elapsed_ms_field: &mut u32) {
        *elapsed_ms_field = self.elapsed_ms_capped();
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use std::thread;
    use std::time::{Duration, Instant};

    use super::*;
    use crate::storage::cache::blob::{BlobCache, BlobCacheConfig, BlobCachePolicy, BlobCachePut};

    fn cache() -> BlobCache {
        BlobCache::new(
            BlobCacheConfig::default()
                .with_l1_bytes_max(64 * 1024)
                .with_shard_count(4)
                .with_max_namespaces(8),
        )
    }

    // -- SweepLimit cap-extraction ------------------------------------------

    #[test]
    fn sweep_limit_entries_only_caps_entries() {
        let limit = SweepLimit::Entries(42);
        assert_eq!(limit.entries_cap(), Some(42));
        assert_eq!(limit.millis_cap(), None);
    }

    #[test]
    fn sweep_limit_millis_only_caps_millis() {
        let limit = SweepLimit::Millis(7);
        assert_eq!(limit.entries_cap(), None);
        assert_eq!(limit.millis_cap(), Some(7));
    }

    #[test]
    fn sweep_limit_either_caps_both() {
        let limit = SweepLimit::Either {
            entries: 100,
            millis: 5,
        };
        assert_eq!(limit.entries_cap(), Some(100));
        assert_eq!(limit.millis_cap(), Some(5));
    }

    // -- Budget first-bound-wins --------------------------------------------

    #[test]
    fn budget_entries_bound_fires_first() {
        let mut budget = Budget::new(SweepLimit::Entries(3), Instant::now());
        assert!(!budget.exhausted());
        assert!(!budget.tick());
        assert!(!budget.tick());
        // Third tick crosses the cap.
        assert!(budget.tick());
        assert!(budget.exhausted());
    }

    #[test]
    fn budget_either_uses_first_bound_to_fire() {
        let mut budget = Budget::new(
            SweepLimit::Either {
                entries: 1_000,
                millis: 2,
            },
            Instant::now(),
        );
        // Entries cap not yet reached.
        budget.tick();
        // Sleep just past the millis cap.
        thread::sleep(Duration::from_millis(5));
        assert!(budget.exhausted(), "millis bound should fire first");
    }

    // -- sweep_expired contract ---------------------------------------------

    #[test]
    fn sweep_expired_with_entries_limit_returns_report_within_bound() {
        let cache = cache();
        cache
            .put("n", "k", BlobCachePut::new(b"v".to_vec()))
            .unwrap();

        let report = BlobCacheSweeper::sweep_expired(&cache, SweepLimit::Entries(10));

        assert_eq!(report.entries_scanned, 0);
        assert_eq!(report.entries_evicted, 0);
        assert_eq!(report.bytes_reclaimed, 0);
        assert!(!report.truncated_due_to_limit);
        assert!(report.by_namespace.is_empty());
    }

    #[test]
    fn sweep_expired_with_millis_limit_honors_wall_clock_bound() {
        let cache = cache();
        let limit_ms = 50u32;
        let started = Instant::now();
        let report = BlobCacheSweeper::sweep_expired(&cache, SweepLimit::Millis(limit_ms));
        let observed_ms = started.elapsed().as_millis() as u32;

        // The sweeper itself must respect the bound. Allow a generous slop
        // for CI scheduling jitter.
        assert!(
            observed_ms <= limit_ms + 50,
            "sweep_expired should not block beyond limit; observed {observed_ms}ms vs cap {limit_ms}ms",
        );
        assert!(
            report.elapsed_ms <= limit_ms + 50,
            "report.elapsed_ms ({}) should be near or under cap ({})",
            report.elapsed_ms,
            limit_ms,
        );
    }

    #[test]
    fn sweep_expired_does_not_remove_unexpired_entries() {
        let cache = cache();
        cache
            .put("alive", "k", BlobCachePut::new(b"v".to_vec()))
            .unwrap();

        let _ = BlobCacheSweeper::sweep_expired(&cache, SweepLimit::Entries(100));

        // Entry must still be retrievable after a sweep when it has no TTL.
        let hit = cache.get("alive", "k").expect("entry survives sweep");
        assert_eq!(hit.value(), b"v");
    }

    #[test]
    fn sweep_expired_evicts_expired_but_not_unexpired() {
        let cache = cache();
        let policy = BlobCachePolicy::default().expires_at_unix_ms(1);
        cache
            .put(
                "n",
                "expired",
                BlobCachePut::new(b"x".to_vec()).with_policy(policy),
            )
            .unwrap();
        cache
            .put("n", "alive", BlobCachePut::new(b"y".to_vec()))
            .unwrap();

        let report = BlobCacheSweeper::sweep_expired(&cache, SweepLimit::Entries(10));

        assert_eq!(report.entries_evicted, 1);
        assert_eq!(report.entries_scanned, 1);
        assert!(report.bytes_reclaimed > 1);
        assert!(cache.get("n", "expired").is_none());
        assert!(cache.get("n", "alive").is_some());
    }

    // -- reclaim_orphans contract -------------------------------------------

    #[test]
    fn reclaim_orphans_returns_well_formed_report() {
        let cache = cache();
        let report = BlobCacheSweeper::reclaim_orphans(&cache, SweepLimit::Entries(10));
        assert_eq!(report.blob_chains_scanned, 0);
        assert_eq!(report.blob_chains_reclaimed, 0);
        assert_eq!(report.bytes_reclaimed, 0);
        assert!(!report.truncated_due_to_limit);
    }

    // -- flush_namespace ----------------------------------------------------

    #[test]
    fn flush_namespace_returns_within_foreground_fast_bound_and_bumps_generation() {
        let cache = cache();
        cache
            .put("ns", "k", BlobCachePut::new(b"v".to_vec()))
            .unwrap();

        let started = Instant::now();
        let report = BlobCacheSweeper::flush_namespace(&cache, "ns");
        let observed = started.elapsed();

        assert_eq!(report.namespace, "ns");
        assert_eq!(report.generation_before, 0);
        assert_eq!(report.generation_after, 1);

        // Foreground-fast contract: <100 µs typical. Allow generous slop for
        // CI noise (debug builds + virtualised runners can be 10× slower).
        assert!(
            observed < Duration::from_millis(5),
            "flush_namespace should be foreground-fast; observed {observed:?}",
        );
        assert!(
            report.elapsed_micros < 5_000,
            "report.elapsed_micros should be <5ms; observed {}µs",
            report.elapsed_micros,
        );

        // The flush bumped the generation: previously-stored entry is gone.
        assert!(
            cache.get("ns", "k").is_none(),
            "entry should be invisible after generation bump",
        );
    }

    #[test]
    fn flush_namespace_on_unknown_namespace_still_returns_well_formed_report() {
        let cache = cache();
        let report = BlobCacheSweeper::flush_namespace(&cache, "never-existed");
        assert_eq!(report.namespace, "never-existed");
        assert_eq!(report.generation_before, 0);
        assert_eq!(report.generation_after, 0);
    }

    // -- Concurrency property test ------------------------------------------

    /// 8 reader threads + 1 sweeper thread. Readers must complete a burst
    /// of `cache.get` calls within a tight time budget while the sweeper
    /// runs concurrently. Empirically verifies the
    /// "concurrent reads never block during sweep" contract documented in
    /// the module-level docstring.
    ///
    /// The test is intentionally short-running (≈250 ms total) so it stays
    /// in the default test suite without slowing CI.
    #[test]
    fn concurrent_reads_never_block_during_sweep() {
        const READER_THREADS: usize = 8;
        const READS_PER_THREAD: usize = 5_000;
        // Per-thread soft cap. If readers were ever serialised behind the
        // sweeper, individual reads would queue and total time would
        // explode well past this bound.
        const READER_SOFT_CAP: Duration = Duration::from_millis(500);

        let cache = Arc::new(cache());
        // Pre-populate with enough entries that reads exercise multiple
        // shards.
        for i in 0..64 {
            cache
                .put("ns", &format!("k{i}"), BlobCachePut::new(vec![i as u8; 32]))
                .unwrap();
        }

        let stop = Arc::new(AtomicBool::new(false));

        // Sweeper thread: keep invoking sweep_expired with a small per-call
        // bound, simulating the runtime scheduler's background pulse.
        let sweeper_cache = Arc::clone(&cache);
        let sweeper_stop = Arc::clone(&stop);
        let sweeper = thread::spawn(move || {
            while !sweeper_stop.load(Ordering::Relaxed) {
                let _ = BlobCacheSweeper::sweep_expired(
                    &sweeper_cache,
                    SweepLimit::Either {
                        entries: 1_000,
                        millis: 5,
                    },
                );
                let _ = BlobCacheSweeper::reclaim_orphans(&sweeper_cache, SweepLimit::Millis(5));
            }
        });

        // Reader threads: each runs a burst of gets and reports its
        // wall-clock time so the assertion is "no reader was starved".
        let reader_handles: Vec<_> = (0..READER_THREADS)
            .map(|tid| {
                let reader_cache = Arc::clone(&cache);
                thread::spawn(move || {
                    let started = Instant::now();
                    for i in 0..READS_PER_THREAD {
                        let key = format!("k{}", (tid * 7 + i) % 64);
                        let _ = reader_cache.get("ns", &key);
                    }
                    started.elapsed()
                })
            })
            .collect();

        let elapsed_per_reader: Vec<Duration> = reader_handles
            .into_iter()
            .map(|h| h.join().expect("reader thread panicked"))
            .collect();

        stop.store(true, Ordering::Relaxed);
        sweeper.join().expect("sweeper thread panicked");

        for (tid, elapsed) in elapsed_per_reader.iter().enumerate() {
            assert!(
                *elapsed < READER_SOFT_CAP,
                "reader {tid} took {elapsed:?}, exceeding soft cap {READER_SOFT_CAP:?} \
                 — sweeper appears to be blocking reads",
            );
        }
    }

    /// Companion property test: while a `flush_namespace` storm runs,
    /// readers from a different namespace must remain unaffected.
    #[test]
    fn flush_namespace_storm_does_not_block_other_namespace_reads() {
        let cache = Arc::new(cache());
        cache
            .put("readers", "k", BlobCachePut::new(b"hello".to_vec()))
            .unwrap();
        // Touch the flush-target namespace once so it exists.
        cache
            .put("flushed", "k", BlobCachePut::new(b"x".to_vec()))
            .unwrap();

        let stop = Arc::new(AtomicBool::new(false));
        let flush_cache = Arc::clone(&cache);
        let flush_stop = Arc::clone(&stop);
        let flusher = thread::spawn(move || {
            while !flush_stop.load(Ordering::Relaxed) {
                let _ = BlobCacheSweeper::flush_namespace(&flush_cache, "flushed");
            }
        });

        let started = Instant::now();
        for _ in 0..10_000 {
            let hit = cache.get("readers", "k").expect("reader namespace alive");
            assert_eq!(hit.value(), b"hello");
        }
        let elapsed = started.elapsed();

        stop.store(true, Ordering::Relaxed);
        flusher.join().expect("flusher panicked");

        assert!(
            elapsed < Duration::from_millis(500),
            "10k reads on a quiet namespace took {elapsed:?} — flush storm appears to block reads",
        );
    }
}