kevy-store 7.0.0

//! Transparent tiering — the store core.
//!
//! Cold values live in a per-shard [`kevy_vlog::Vlog`]; each leaves a
//! 24-byte [`ColdRef`] stub *in the map* (`Value::Cold`), so every raw
//! probe (SET fast path, DEL, RENAME, FLUSHALL, SCAN's single-table
//! sweep, the reaper) works unchanged by construction. The two-stage
//! funnel keeps the ~196 downstream `Value` matches Cold-free:
//!
//! - **Stage 1 (zero IO)**: existence / NX / XX / EXPIRE-family answer
//!   from the `Entry`; `TYPE` from the stub's tag; a WRONGTYPE refusal
//!   never pays a pread ([`Store::tier_resolve`] / [`Store::tier_serve`]
//!   check the tag before touching disk).
//! - **Stage 2 (materialize)**: write paths promote in place; read
//!   paths run the promotion gate — the FIRST materializing access
//!   serves decoded bytes without installing (probation `touched`
//!   mark), the SECOND promotes. Bulk/`&self` shared-lane reads never
//!   promote and never set the mark.
//!
//! Demotion/promotion are dedicated in-place primitives — NOT
//! `insert_entry`/`remove_entry`, which would clear hash field-TTLs,
//! capture `new` events and drift the `expires` counter. They emit
//! zero keyspace notifications, never bump WATCH versions, and
//! preserve `lru_clock` (LFU history survives a round trip).
//!
//! This module is compiled only with `std` off-wasm (the vlog needs a
//! real filesystem); a sibling `cfg(not(...))` block provides funnel
//! passthroughs so call sites stay cfg-free.

#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
mod enabled {
    use std::io;
    use std::path::{Path, PathBuf};

    use kevy_vlog::{Vlog, VlogRef};

    use crate::value::{ColdRef, Value};
    use crate::{EvictionPolicy, SmallBytes, Store};

    pub(crate) use crate::tier_state::TierState;

    pub use crate::tier_stats::TierStats;

    impl ColdRef {
        #[inline]
        pub(crate) fn vref(self) -> VlogRef {
            VlogRef::new(self.file_id, self.offset, self.len)
        }
    }

    impl Store {
        /// Turn tiering on: open (wiping) the vlog under `dir` and set
        /// the RAM `budget` the demotion watermark works against.
        /// Callers own the dir choice (`<data>/tier/` by convention).
        pub fn enable_tiering(&mut self, dir: &Path, budget: u64) -> io::Result<()> {
            let dir: PathBuf = dir.to_path_buf();
            let vlog = Vlog::open(&dir, kevy_vlog::DEFAULT_ROTATE_BYTES)?;
            self.cold_backing = true;
            self.tier = Some(TierState {
                vlog,
                budget,
                policy: EvictionPolicy::AllKeysLru,
                demotions_total: 0,
                tick_wait: 0,
                tick_skip: 0,
                hand: 0,
                promotions_total: 0,
                preads_total: 0,
                peek_preads_total: 0,
                batch_submissions_total: 0,
                cold_keys: 0,
                cold_bytes: 0,
                max_spill: 0,
                reserved_bytes: 0,
                overhead_bytes: 0,
                growth_reserve: 0,
                stub_bytes: 0,
                renames: std::collections::HashMap::new(),
            });
            Ok(())
        }

        /// Live-update the tiering budget (auto/percent re-resolution on
        /// the shard tick, `CONFIG SET` — the maxmemory reapply
        /// precedent). Touches nothing but the number: the vlog, the
        /// stubs and every counter stay as they are. No-op when tiering
        /// is off.
        #[inline]
        pub fn set_tier_budget(&mut self, bytes: u64) {
            if let Some(t) = &mut self.tier {
                t.budget = bytes;
            }
        }

        /// Cap the largest spillable value (0 = unlimited). Embedded
        /// sets 256 KiB by default (RFC §7) to bound cold-read
        /// lock-hold time; the server leaves it unlimited. No-op when
        /// tiering is off.
        #[inline]
        pub fn set_tier_max_spill(&mut self, bytes: u64) {
            if let Some(t) = &mut self.tier {
                t.max_spill = bytes;
            }
        }

        /// Feed the index/view memory floor (Σ segment `approx_bytes`
        /// on this shard) into the unified watermark. Called per shard
        /// tick by the serving layer. No-op when tiering is off.
        #[inline]
        pub fn set_tier_reserved(&mut self, bytes: u64) {
            if let Some(t) = &mut self.tier {
                t.reserved_bytes = bytes;
            }
        }

        /// Feed this shard's share of the memory the process holds live
        /// outside `used_memory` and the index floor — what a serving
        /// layer measures from the allocator. It lowers the demote target,
        /// so the budget bounds the process and not only what the store
        /// accounts for. No-op when tiering is off.
        ///
        /// ```
        /// use kevy_store::Store;
        /// # let dir = std::env::temp_dir().join(format!("kevy-doc-overhead-{}", std::process::id()));
        /// let mut s = Store::new();
        /// s.enable_tiering(&dir, 1 << 20)?;
        /// s.set_tier_overhead(1000);
        /// assert_eq!(s.tier_stats().effective_target, (1 << 20) * 19 / 20 - 1000);
        /// # std::fs::remove_dir_all(&dir)?;
        /// # Ok::<(), Box<dyn std::error::Error>>(())
        /// ```
        #[inline]
        pub fn set_tier_overhead(&mut self, bytes: u64) {
            if let Some(t) = &mut self.tier {
                t.overhead_bytes = bytes;
            }
        }

        /// While the keyspace table is within an eighth of its next growth,
        /// set aside the bytes that growth will add, so demotion makes room
        /// before the bigger table lands instead of after it — by then the
        /// process holds both. Called from the shard tick; no-op when
        /// tiering is off.
        ///
        /// ```
        /// use kevy_store::{SetCondition, Store};
        /// # let dir = std::env::temp_dir().join(format!("kevy-doc-growth-{}", std::process::id()));
        /// let mut s = Store::new();
        /// s.enable_tiering(&dir, 1 << 30)?;
        /// let full = s.tier_stats().effective_target;
        /// // sixteen slots hold fourteen keys; the fourteenth leaves no room
        /// for i in 0..14 {
        ///     s.set(format!("k{i}").as_bytes(), b"v".to_vec(), None, SetCondition::Always);
        /// }
        /// s.tier_reserve_growth();
        /// assert!(s.tier_stats().effective_target < full, "the next table is set aside");
        /// s.set(b"k14", b"v".to_vec(), None, SetCondition::Always); // grows
        /// s.tier_reserve_growth();
        /// assert_eq!(s.tier_stats().effective_target, full, "and released once it is charged");
        /// # std::fs::remove_dir_all(&dir)?;
        /// # Ok::<(), Box<dyn std::error::Error>>(())
        /// ```
        pub fn tier_reserve_growth(&mut self) {
            let Some(t) = &mut self.tier else { return };
            let cap = self.map.capacity();
            t.growth_reserve = if cap > 0 && self.map.room() <= cap / 8 {
                (self.map.grown_footprint() as u64).saturating_sub(self.keyspace_bytes)
            } else {
                0
            };
        }

        /// Whether the index/view floor (`reserved_bytes + extra`)
        /// already exhausts the tier's demotable headroom — the
        /// IDX.CREATE refusal predicate. What demotion can never reclaim
        /// is the index floor plus the part of `used_memory` that stays
        /// when every value is cold: the keyspace table and the cold
        /// keys' own bytes. `false` when tiering is off.
        pub fn tier_index_floor_blocked(&self, extra: u64) -> bool {
            match &self.tier {
                Some(t) => {
                    let cold_key_heap = t
                        .stub_bytes
                        .saturating_sub(t.cold_keys.saturating_mul(crate::value::ENTRY_OVERHEAD));
                    let fixed = self
                        .keyspace_bytes
                        .saturating_add(cold_key_heap)
                        .saturating_add(t.overhead_bytes)
                        .saturating_add(t.growth_reserve);
                    t.reserved_bytes.saturating_add(extra).saturating_add(fixed)
                        >= crate::tier_demote::watermark(t.budget)
                }
                None => false,
            }
        }

        /// Whether tiering is on for this shard.
        #[inline]
        pub fn tier_enabled(&self) -> bool {
            self.tier.is_some()
        }

        /// Tiering gauges — zeros when tiering is off.
        pub fn tier_stats(&self) -> TierStats {
            match &self.tier {
                None => TierStats::default(),
                Some(t) => {
                    let v = t.vlog.stats();
                    TierStats {
                        budget: t.budget,
                        effective_target: crate::tier_demote::effective_target(t),
                        reserved_bytes: t.reserved_bytes,
                        stub_bytes: t.stub_bytes,
                        demotions_total: t.demotions_total,
                        promotions_total: t.promotions_total,
                        preads_total: t.preads_total,
                        peek_preads_total: t.peek_preads_total,
                        batch_submissions_total: t.batch_submissions_total,
                        cold_keys: t.cold_keys,
                        cold_bytes: t.cold_bytes,
                        vlog_files: v.files as u64,
                        vlog_bytes: v.bytes,
                        vlog_live_bytes: v.live_bytes,
                        vlog_epoch: v.epoch,
                    }
                }
            }
        }

        /// The value log's compression accounting (see
        /// [`kevy_vlog::CompressionStats`]); all zero when tiering is off.
        pub fn tier_compression(&self) -> kevy_vlog::CompressionStats {
            self.tier.as_ref().map(|t| t.vlog.compression()).unwrap_or_default()
        }

        /// Whether the LRU/LFU access clock must advance: eviction
        /// (`maxmemory > 0`) or tiering (demotion scoring) needs it.
        /// Same single-branch cost as the old `maxmemory > 0` test.
        #[inline]
        pub(crate) fn clock_on(&self) -> bool {
            self.maxmemory > 0 || self.tier.is_some()
        }

        /// The policy access-touches score under: eviction's when
        /// enabled, else the tier's (tiered-lru default).
        #[inline]
        pub(crate) fn touch_policy(&self) -> EvictionPolicy {
            if self.maxmemory > 0 {
                return self.eviction_policy;
            }
            match &self.tier {
                Some(t) => t.policy,
                None => self.eviction_policy,
            }
        }

        /// Pin every current vlog file (view pinning): a
        /// snapshot view / rewrite plan captured from a tiered store
        /// carries these so its frozen [`ColdRef`]s stay readable on the
        /// serializer thread across compaction — a retired file is
        /// unlinked only when the last pin drops. Empty when tiering is
        /// off.
        pub fn tier_pins(&self) -> Vec<std::sync::Arc<kevy_vlog::VlogFile>> {
            match &self.tier {
                Some(t) => t.vlog.pin_all(),
                None => Vec::new(),
            }
        }

        /// Serialization-side cold materialization: decode `v`'s
        /// record into a fresh owned hot value WITHOUT installing,
        /// promoting, or setting the probation mark — persistence is a
        /// bulk path and never promotes. `None` when `v` is hot.
        pub fn materialize_cold(&self, key: &[u8], v: &Value) -> Option<Value> {
            self.tier_peek_value(key, v)
        }

        /// A cold stub is being discarded (DEL / overwrite / expiry /
        /// FLUSH of the key): credit its record's bytes as dead so the
        /// compaction trigger sees them, and release the stub's RAM
        /// cost from `stub_bytes`. `key_heap` is the heap-byte cost of
        /// the key the stub lived under (part of the stub cost —
        /// callers pass `key_heap_bytes_for(key)` / `key.heap_bytes()`
        /// since some sites have already moved the key into the map).
        /// No-op for hot values.
        pub(crate) fn tier_note_dead(&mut self, key_heap: u64, v: &Value) {
            let Value::Cold(c) = v else { return };
            if c.is_seg() {
                self.segrow_note_dead(*c);
                return;
            }
            if let Some(t) = &mut self.tier {
                t.vlog.note_dead(c.vref());
                t.cold_keys = t.cold_keys.saturating_sub(1);
                t.cold_bytes = t.cold_bytes.saturating_sub(u64::from(c.weight));
                t.stub_bytes = t.stub_bytes.saturating_sub(crate::value::ENTRY_OVERHEAD + key_heap);
                t.renames.remove(&(c.file_id, c.offset));
            }
        }

        /// RENAME moved a cold stub from `src` to `dst` without reading
        /// it — the record's embedded key is now stale; register the
        /// forward pointer compaction resolves through, and re-account
        /// the stub cost for the new key's heap bytes.
        pub(crate) fn tier_note_renamed(&mut self, v: &Value, src: &[u8], dst: &[u8]) {
            let Value::Cold(c) = v else { return };
            if let Some(t) = &mut self.tier {
                t.stub_bytes = t
                    .stub_bytes
                    .saturating_sub(crate::key_heap_bytes_for(src))
                    .saturating_add(crate::key_heap_bytes_for(dst));
                t.renames.insert((c.file_id, c.offset), SmallBytes::from_slice(dst));
            }
        }

        /// FLUSHALL: every stub died with the map — mark the whole log
        /// dead (sealed files drop scan-free at the next compaction).
        pub(crate) fn tier_on_flushall(&mut self) {
            if let Some(t) = &mut self.tier {
                t.vlog.mark_all_dead();
                t.renames.clear();
                t.cold_keys = 0;
                t.cold_bytes = 0;
                t.stub_bytes = 0;
            }
        }
    }
}

#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
pub(crate) use enabled::TierState;
#[cfg(all(feature = "std", not(target_arch = "wasm32")))]
pub use enabled::TierStats;

/// Funnel passthroughs for builds without the tier backend (no_std /
/// wasm): `Value::Cold` cannot be constructed there (no `enable_tiering`),
/// so the funnel degenerates to `live_entry` and no-ops.
#[cfg(not(all(feature = "std", not(target_arch = "wasm32"))))]
mod disabled {
    use crate::value::Value;
    use crate::{EvictionPolicy, Store};

    impl Store {
        #[inline]
        pub(crate) fn clock_on(&self) -> bool {
            self.maxmemory > 0
        }

        #[inline]
        pub(crate) fn touch_policy(&self) -> EvictionPolicy {
            self.eviction_policy
        }

        /// No tier backend on this target — `Value::Cold` cannot exist.
        #[inline]
        pub fn materialize_cold(&self, _key: &[u8], _v: &Value) -> Option<Value> {
            None
        }

        /// No tier backend on this target — always 0.
        #[inline]
        pub fn demote_to_watermark(&mut self) -> usize {
            0
        }

        #[inline]
        pub(crate) fn tier_note_dead(&mut self, _key_heap: u64, _v: &Value) {}

        #[inline]
        pub(crate) fn tier_note_renamed(&mut self, _v: &Value, _src: &[u8], _dst: &[u8]) {}

        #[inline]
        pub(crate) fn tier_on_flushall(&mut self) {}

        /// No tier backend on this target — no-op.
        #[inline]
        pub fn set_tier_budget(&mut self, _bytes: u64) {}

        /// No tier backend on this target — no-op.
        #[inline]
        pub fn set_tier_reserved(&mut self, _bytes: u64) {}

        /// No tier backend on this target — no-op.
        #[inline]
        pub fn set_tier_overhead(&mut self, _bytes: u64) {}

        /// No tier backend on this target — no-op.
        #[inline]
        pub fn tier_reserve_growth(&mut self) {}

        /// No tier backend on this target — always false.
        #[inline]
        pub fn tier_index_floor_blocked(&self, _extra: u64) -> bool {
            false
        }

        #[inline]
        pub(crate) fn promote_in_place(&mut self, _key: &[u8]) -> bool {
            false
        }

        /// No tier backend on this target — always 0.
        #[inline]
        pub fn try_demote_after_write(&mut self) -> usize {
            0
        }

        /// No tier backend on this target — always 0.
        #[inline]
        pub fn demote_step(&mut self) -> usize {
            0
        }

        /// No tier backend on this target — always 0.
        #[inline]
        pub fn tier_compact_tick(&mut self) -> usize {
            0
        }

        /// No tier backend on this target — always false.
        #[doc(hidden)]
        pub fn debug_force_demote(&mut self, _key: &[u8]) -> bool {
            false
        }

        /// No tier backend on this target — always false.
        #[inline]
        pub fn tier_enabled(&self) -> bool {
            false
        }
    }
}