reddb-io-server 1.33.3

use std::cmp::Ordering as CmpOrdering;
use std::collections::{BinaryHeap, HashMap, HashSet};
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};

use indexmap::IndexMap;

use super::cache::{BlobCacheHit, BlobCacheKey};
use super::entry::Entry;
use crate::storage::cache::extended_ttl::{EffectiveExpiry, ExpiryDecision, ExtendedTtlPolicy};

/// Blob Cache L1 shard.
///
/// Decision #218 documents this as the tiered-cache exception to the
/// page-cache SIEVE invariant: priority may bias victim choice, but the shard
/// still falls back to the normal `visited` sweep.
#[derive(Debug)]
pub(super) struct Shard {
    entries: IndexMap<BlobCacheKey, Entry>,
    l2_second_hit_markers: HashSet<BlobCacheKey>,
    slots: Vec<Option<BlobCacheKey>>,
    free_slots: Vec<usize>,
    hand: usize,
    sweep_hand: usize,
    stale_sweep_epoch: u64,
    stale_sweep_completed_epoch: u64,
    stale_sweep_slots_remaining: usize,
    expiry_heap: BinaryHeap<ExpiryRecord>,
    expiry_sequence: u64,
    pub(super) bytes: usize,
}

#[derive(Debug, Eq, PartialEq)]
struct ExpiryRecord {
    deadline_unix_ms: u64,
    sequence: u64,
    key: BlobCacheKey,
}

impl Ord for ExpiryRecord {
    fn cmp(&self, other: &Self) -> CmpOrdering {
        // BinaryHeap is a max-heap. Reverse deadline/sequence ordering so the
        // earliest deadline is at the top without requiring BlobCacheKey: Ord.
        other
            .deadline_unix_ms
            .cmp(&self.deadline_unix_ms)
            .then_with(|| other.sequence.cmp(&self.sequence))
    }
}

impl PartialOrd for ExpiryRecord {
    fn partial_cmp(&self, other: &Self) -> Option<CmpOrdering> {
        Some(self.cmp(other))
    }
}

impl Shard {
    pub(super) fn new() -> Self {
        Self {
            entries: IndexMap::new(),
            l2_second_hit_markers: HashSet::new(),
            slots: Vec::new(),
            free_slots: Vec::new(),
            hand: 0,
            sweep_hand: 0,
            stale_sweep_epoch: 0,
            stale_sweep_completed_epoch: 0,
            stale_sweep_slots_remaining: 0,
            expiry_heap: BinaryHeap::new(),
            expiry_sequence: 0,
            bytes: 0,
        }
    }

    pub(super) fn get(
        &mut self,
        key: &BlobCacheKey,
        now_ms: u64,
        namespace_generation: u64,
    ) -> Lookup {
        self.get_by_parts(&key.namespace, &key.key, now_ms, namespace_generation)
    }

    /// Common hot hits only touch SIEVE's advisory visited bit. Idle/stale
    /// policies and removal still require the exclusive, revalidated path.
    pub(super) fn shared_hit(
        &self,
        namespace: &str,
        key: &str,
        now_ms: u64,
        namespace_generation: u64,
    ) -> Option<BlobCacheHit> {
        let entry = self.entries.get(&BlobCacheKey::borrowed(namespace, key))?;
        if entry.namespace_generation != namespace_generation
            || entry.extended.is_active()
            || entry.is_expired_at(now_ms)
        {
            return None;
        }
        entry.visited.store(true, Ordering::Relaxed);
        // last_access is used only by immutable extended-TTL policies. An
        // ordinary hit need not mutate it while holding a shared shard guard.
        Some(entry.hit())
    }

    pub(super) fn get_by_parts(
        &mut self,
        namespace: &str,
        key: &str,
        now_ms: u64,
        namespace_generation: u64,
    ) -> Lookup {
        let borrowed = BlobCacheKey::borrowed(namespace, key);
        let remove = {
            let Some(entry) = self.entries.get_mut(&borrowed) else {
                return Lookup::Miss;
            };
            if entry.namespace_generation != namespace_generation {
                Some(RemovalReason::Stale)
            } else if !entry.extended.is_active() {
                if entry.is_expired_at(now_ms) {
                    Some(RemovalReason::Expired)
                } else {
                    entry.visited.store(true, Ordering::Relaxed);
                    entry.last_access_unix_ms = now_ms;
                    return Lookup::Hit(entry.hit());
                }
            } else {
                #[cfg(test)]
                super::cache::EFFECTIVE_EXPIRY_COMPUTE_CALLS.with(|c| c.set(c.get() + 1));
                match EffectiveExpiry::compute(
                    entry.expires_at_unix_ms,
                    now_ms,
                    entry.last_access_unix_ms,
                    &entry.extended,
                ) {
                    ExpiryDecision::Fresh => {
                        entry.visited.store(true, Ordering::Relaxed);
                        entry.last_access_unix_ms = now_ms;
                        return Lookup::Hit(entry.hit());
                    }
                    ExpiryDecision::Stale {
                        window_remaining_ms,
                    } => {
                        entry.visited.store(true, Ordering::Relaxed);
                        entry.last_access_unix_ms = now_ms;
                        return Lookup::Hit(entry.hit_stale(window_remaining_ms));
                    }
                    ExpiryDecision::Expired => {
                        if entry.is_expired_at(now_ms) {
                            Some(RemovalReason::Expired)
                        } else {
                            Some(RemovalReason::IdleEvicted)
                        }
                    }
                }
            }
        };

        let key = BlobCacheKey::new(namespace, key);
        let removed = self.remove(&key).expect("entry exists");
        match remove.expect("removal reason") {
            RemovalReason::Expired => Lookup::Expired(removed),
            RemovalReason::IdleEvicted => Lookup::IdleEvicted(removed),
            RemovalReason::Stale => Lookup::Stale(removed),
        }
    }

    pub(super) fn contains(
        &mut self,
        key: &BlobCacheKey,
        now_ms: u64,
        namespace_generation: u64,
    ) -> Lookup {
        self.contains_by_parts(&key.namespace, &key.key, now_ms, namespace_generation)
    }

    pub(super) fn contains_by_parts(
        &mut self,
        namespace: &str,
        key: &str,
        now_ms: u64,
        namespace_generation: u64,
    ) -> Lookup {
        let borrowed = BlobCacheKey::borrowed(namespace, key);
        let remove = {
            let Some(entry) = self.entries.get_mut(&borrowed) else {
                return Lookup::Miss;
            };
            if entry.namespace_generation != namespace_generation {
                Some(RemovalReason::Stale)
            } else if entry.is_expired_at(now_ms) {
                Some(RemovalReason::Expired)
            } else {
                entry.visited.store(true, Ordering::Relaxed);
                return Lookup::Present;
            }
        };

        let key = BlobCacheKey::new(namespace, key);
        let removed = self.remove(&key).expect("entry exists");
        match remove.expect("removal reason") {
            RemovalReason::Expired => Lookup::Expired(removed),
            RemovalReason::Stale => Lookup::Stale(removed),
            RemovalReason::IdleEvicted => unreachable!("contains does not evaluate idle TTL"),
        }
    }

    pub(super) fn existing_version(
        &self,
        key: &BlobCacheKey,
        namespace_generation: u64,
    ) -> Option<u64> {
        self.entries.get(key).and_then(|entry| {
            if entry.namespace_generation == namespace_generation {
                entry.version
            } else {
                None
            }
        })
    }

    pub(super) fn can_update_expiry(
        &self,
        key: &BlobCacheKey,
        now_ms: u64,
        namespace_generation: u64,
    ) -> bool {
        self.entries.get(key).is_some_and(|entry| {
            if entry.namespace_generation != namespace_generation {
                return false;
            }
            if entry.extended.is_active() {
                !matches!(
                    EffectiveExpiry::compute(
                        entry.expires_at_unix_ms,
                        now_ms,
                        entry.last_access_unix_ms,
                        &entry.extended,
                    ),
                    ExpiryDecision::Expired
                )
            } else {
                !entry.is_expired_at(now_ms)
            }
        })
    }

    pub(super) fn update_expiry(
        &mut self,
        key: &BlobCacheKey,
        expires_at_unix_ms: Option<u64>,
        now_ms: u64,
    ) {
        let entry = self
            .entries
            .get_mut(key)
            .expect("live expiry target exists");
        entry.expires_at_unix_ms = expires_at_unix_ms;
        entry.last_access_unix_ms = now_ms;
        self.track_expiry(key);
    }

    pub(super) fn insert(&mut self, key: BlobCacheKey, mut entry: Entry) -> InsertOutcome {
        self.l2_second_hit_markers.remove(&key);
        let old_entry = if let Some(old) = self.entries.shift_remove(&key) {
            let slot_index = old.slot_index;
            self.bytes = self.bytes.saturating_sub(old.l1_charge_bytes);
            entry.slot_index = slot_index;
            Some(old)
        } else {
            let slot_index = self.free_slots.pop().unwrap_or_else(|| {
                self.slots.push(None);
                self.slots.len() - 1
            });
            entry.slot_index = slot_index;
            None
        };

        self.bytes = self.bytes.saturating_add(entry.l1_charge_bytes);
        self.slots[entry.slot_index] = Some(key.clone());
        self.entries.insert(key.clone(), entry);
        self.track_expiry(&key);
        InsertOutcome {
            old_entry,
            admitted: true,
        }
    }

    pub(super) fn evict_one(&mut self) -> Option<(BlobCacheKey, Entry)> {
        if self.entries.is_empty() {
            self.hand = 0;
            return None;
        }
        let max_sweeps = self.slots.len().saturating_mul(2).max(1);
        for _ in 0..max_sweeps {
            if self.entries.is_empty() {
                self.hand = 0;
                return None;
            }
            if self.hand >= self.slots.len() {
                self.hand = 0;
            }
            let Some(candidate) = self.slots[self.hand].clone() else {
                self.advance_hand();
                continue;
            };
            let Some(entry) = self.entries.get(&candidate) else {
                self.slots[self.hand] = None;
                self.free_slots.push(self.hand);
                self.advance_hand();
                continue;
            };
            if entry.visited.load(Ordering::Relaxed) {
                if let Some(entry) = self.entries.get_mut(&candidate) {
                    entry.visited.store(false, Ordering::Relaxed);
                }
                self.advance_hand();
                continue;
            }

            if self.has_lower_priority_unvisited(entry.priority) {
                self.advance_hand();
                continue;
            }

            let removed = self
                .entries
                .shift_remove(&candidate)
                .expect("candidate exists");
            self.bytes = self.bytes.saturating_sub(removed.l1_charge_bytes);
            self.slots[self.hand] = None;
            self.free_slots.push(self.hand);
            self.advance_hand();
            return Some((candidate, removed));
        }
        None
    }

    pub(super) fn len(&self) -> usize {
        self.entries.len()
    }

    pub(super) fn remove(&mut self, key: &BlobCacheKey) -> Option<Entry> {
        self.l2_second_hit_markers.remove(key);
        let removed = self.entries.shift_remove(key)?;
        self.bytes = self.bytes.saturating_sub(removed.l1_charge_bytes);
        self.slots[removed.slot_index] = None;
        self.free_slots.push(removed.slot_index);
        if self.entries.is_empty() || self.hand >= self.slots.len() {
            self.hand = 0;
        }
        self.maybe_rebuild_expiry_index();
        Some(removed)
    }

    pub(super) fn clear_l2_hit_marker(&mut self, key: &BlobCacheKey) {
        self.l2_second_hit_markers.remove(key);
    }

    pub(super) fn l2_hit_should_promote_on_second_hit(&mut self, key: &BlobCacheKey) -> bool {
        if self.l2_second_hit_markers.remove(key) {
            true
        } else {
            self.l2_second_hit_markers.insert(key.clone());
            false
        }
    }

    pub(super) fn keys_matching(
        &self,
        mut predicate: impl FnMut(&BlobCacheKey) -> bool,
    ) -> Vec<BlobCacheKey> {
        self.entries
            .keys()
            .filter(|key| predicate(key))
            .cloned()
            .collect()
    }

    pub(super) fn remove_one_from_namespace(
        &mut self,
        namespace: &str,
    ) -> Option<(BlobCacheKey, Entry)> {
        let key = self
            .entries
            .keys()
            .find(|key| key.namespace == namespace)
            .cloned()?;
        let entry = self.remove(&key).expect("selected namespace entry exists");
        Some((key, entry))
    }

    /// Remove due entries from the deadline index. Stale heap records are
    /// validated against the current entry and discarded lazily.
    pub(super) fn sweep_expired(
        &mut self,
        now_ms: u64,
        entries_limit: usize,
        started: Instant,
        time_limit: Option<Duration>,
    ) -> ShardSweep {
        let mut sweep = ShardSweep::default();
        if entries_limit == 0 {
            return sweep;
        }

        while sweep.entries_scanned < entries_limit {
            if time_limit.is_some_and(|limit| started.elapsed() >= limit) {
                break;
            }
            let Some(record) = self.expiry_heap.peek() else {
                sweep.completed_cycle = true;
                break;
            };
            if record.deadline_unix_ms > now_ms {
                sweep.completed_cycle = true;
                break;
            }
            let record = self.expiry_heap.pop().expect("peeked expiry record exists");
            sweep.entries_scanned += 1;
            *sweep
                .scanned_by_namespace
                .entry(record.key.namespace.clone())
                .or_default() += 1;

            let Some(entry) = self.entries.get(&record.key) else {
                continue;
            };
            let expired = if entry.extended.is_active() {
                matches!(
                    EffectiveExpiry::compute(
                        entry.expires_at_unix_ms,
                        now_ms,
                        entry.last_access_unix_ms,
                        &entry.extended,
                    ),
                    ExpiryDecision::Expired
                )
            } else {
                entry.is_expired_at(now_ms)
            };
            if expired {
                let entry = self
                    .remove(&record.key)
                    .expect("indexed expiry candidate exists");
                sweep.removed.push((record.key, entry, true));
            } else if let Some(deadline_unix_ms) = self
                .entries
                .get(&record.key)
                .and_then(Self::effective_removal_deadline)
            {
                self.push_expiry_record(record.key, deadline_unix_ms);
            }
        }
        sweep
    }

    /// Reclaim entries hidden by an O(1) namespace generation flush. This is
    /// deliberately separate from TTL expiry so ordinary sweeps do not scan
    /// the cache when no namespace flush is pending.
    pub(super) fn sweep_stale_generations(
        &mut self,
        namespace_generations: &HashMap<String, u64>,
        epoch: u64,
        entries_limit: usize,
        started: Instant,
        time_limit: Option<Duration>,
    ) -> ShardSweep {
        let mut sweep = ShardSweep::default();
        if self.stale_sweep_epoch != epoch {
            self.stale_sweep_epoch = epoch;
            self.sweep_hand = 0;
            self.stale_sweep_slots_remaining = self.slots.len();
        }
        if self.entries.is_empty() || entries_limit == 0 {
            if self.entries.is_empty() {
                self.stale_sweep_slots_remaining = 0;
                self.stale_sweep_completed_epoch = epoch;
                sweep.completed_cycle = true;
            }
            return sweep;
        }

        while self.stale_sweep_slots_remaining > 0 && sweep.entries_scanned < entries_limit {
            if time_limit.is_some_and(|limit| started.elapsed() >= limit) {
                break;
            }
            if self.sweep_hand >= self.slots.len() {
                self.sweep_hand = 0;
            }
            let slot = self.sweep_hand;
            self.sweep_hand = (self.sweep_hand + 1) % self.slots.len();
            self.stale_sweep_slots_remaining -= 1;

            let Some(key) = self.slots[slot].clone() else {
                continue;
            };
            let Some(entry) = self.entries.get(&key) else {
                continue;
            };
            sweep.entries_scanned += 1;
            *sweep
                .scanned_by_namespace
                .entry(key.namespace.clone())
                .or_default() += 1;
            let stale_generation = entry.namespace_generation
                != namespace_generations
                    .get(&key.namespace)
                    .copied()
                    .unwrap_or(0);
            if stale_generation {
                let entry = self.remove(&key).expect("stale sweep candidate exists");
                sweep.removed.push((key, entry, false));
            }
        }
        sweep.completed_cycle = self.stale_sweep_slots_remaining == 0;
        if sweep.completed_cycle {
            self.stale_sweep_completed_epoch = epoch;
        }
        sweep
    }

    pub(super) fn stale_cleanup_completed(&self, epoch: u64) -> bool {
        self.stale_sweep_completed_epoch >= epoch
    }

    pub(super) fn entry_has_any_tag(&self, key: &BlobCacheKey, labels: &HashSet<String>) -> bool {
        self.entries
            .get(key)
            .is_some_and(|entry| labels.iter().any(|label| entry.tags.contains(label)))
    }

    pub(super) fn entry_has_any_dependency(
        &self,
        key: &BlobCacheKey,
        labels: &HashSet<String>,
    ) -> bool {
        self.entries.get(key).is_some_and(|entry| {
            labels
                .iter()
                .any(|label| entry.dependencies.contains(label))
        })
    }

    fn has_lower_priority_unvisited(&self, priority: u8) -> bool {
        self.entries
            .values()
            .any(|entry| !entry.visited.load(Ordering::Relaxed) && entry.priority < priority)
    }

    fn advance_hand(&mut self) {
        if self.slots.is_empty() {
            self.hand = 0;
        } else {
            self.hand = (self.hand + 1) % self.slots.len();
        }
    }

    fn track_expiry(&mut self, key: &BlobCacheKey) {
        if let Some(deadline_unix_ms) = self
            .entries
            .get(key)
            .and_then(Self::effective_removal_deadline)
        {
            self.push_expiry_record(key.clone(), deadline_unix_ms);
        }
        self.maybe_rebuild_expiry_index();
    }

    fn maybe_rebuild_expiry_index(&mut self) {
        let rebuild_threshold = self.entries.len().saturating_mul(2);
        if (self.entries.is_empty() && !self.expiry_heap.is_empty())
            || self.expiry_heap.len() > rebuild_threshold
        {
            self.rebuild_expiry_index();
        }
    }

    fn push_expiry_record(&mut self, key: BlobCacheKey, deadline_unix_ms: u64) {
        let sequence = self.expiry_sequence;
        self.expiry_sequence = self.expiry_sequence.wrapping_add(1);
        self.expiry_heap.push(ExpiryRecord {
            deadline_unix_ms,
            sequence,
            key,
        });
    }

    fn rebuild_expiry_index(&mut self) {
        let deadlines: Vec<_> = self
            .entries
            .iter()
            .filter_map(|(key, entry)| {
                Self::effective_removal_deadline(entry)
                    .map(|deadline_unix_ms| (key.clone(), deadline_unix_ms))
            })
            .collect();
        self.expiry_heap = BinaryHeap::with_capacity(deadlines.len());
        for (key, deadline_unix_ms) in deadlines {
            self.push_expiry_record(key, deadline_unix_ms);
        }
    }

    fn effective_removal_deadline(entry: &Entry) -> Option<u64> {
        if !entry.extended.is_active() {
            return entry.expires_at_unix_ms;
        }
        let idle_deadline = entry.extended.idle_ttl_ms.map(|idle_ttl_ms| {
            entry
                .last_access_unix_ms
                .saturating_add(idle_ttl_ms)
                .saturating_add(1)
        });
        let hard_deadline = entry.expires_at_unix_ms.map(|hard_expiry| {
            hard_expiry
                .saturating_add(entry.extended.stale_serve_ms.unwrap_or(0))
                .saturating_add(1)
        });
        match (idle_deadline, hard_deadline) {
            (Some(idle), Some(hard)) => Some(idle.min(hard)),
            (Some(idle), None) => Some(idle),
            (None, Some(hard)) => Some(hard),
            (None, None) => None,
        }
    }
}

#[derive(Default)]
pub(super) struct ShardSweep {
    pub(super) entries_scanned: usize,
    pub(super) scanned_by_namespace: std::collections::HashMap<String, usize>,
    pub(super) removed: Vec<(BlobCacheKey, Entry, bool)>,
    pub(super) completed_cycle: bool,
}

pub(super) enum Lookup {
    Hit(BlobCacheHit),
    Present,
    Expired(Entry),
    /// Entry was killed by the idle TTL gate of an `ExtendedTtlPolicy`.
    /// Distinguished from [`Expired`] so the cache-level counter for
    /// idle evictions stays separate from hard-TTL expirations.
    IdleEvicted(Entry),
    Stale(Entry),
    Miss,
}

enum RemovalReason {
    Expired,
    IdleEvicted,
    Stale,
}

pub(super) struct InsertOutcome {
    pub(super) old_entry: Option<Entry>,
    pub(super) admitted: bool,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::cache::blob::{BlobCachePolicy, CachePresence};

    fn key(name: &str) -> BlobCacheKey {
        BlobCacheKey::new("n", name)
    }

    fn entry(bytes: &[u8], priority: u8) -> Entry {
        Entry::new(
            bytes.to_vec(),
            Default::default(),
            Default::default(),
            Default::default(),
            BlobCachePolicy::default().priority(priority),
            0,
            1_000,
            "n",
            "k",
        )
    }

    #[test]
    fn insert_replaces_existing_key_and_tracks_bytes() {
        let mut shard = Shard::new();
        let k = key("same");

        let first = shard.insert(k.clone(), entry(b"abc", 128));
        assert!(first.old_entry.is_none());
        assert_eq!(
            shard.bytes,
            shard.entries.get(&k).expect("entry").l1_charge_bytes
        );

        let second = shard.insert(k.clone(), entry(b"xy", 128));
        assert_eq!(second.old_entry.expect("old entry").size, 3);
        assert_eq!(shard.len(), 1);
        assert_eq!(
            shard.bytes,
            shard.entries.get(&k).expect("entry").l1_charge_bytes
        );
        assert!(matches!(shard.contains(&k, 1_000, 0), Lookup::Present));
    }

    #[test]
    fn insert_replaces_existing_key_in_the_same_slot() {
        let mut shard = Shard::new();
        let k = key("same");

        shard.insert(k.clone(), entry(b"abc", 128));
        let first_slot = shard.entries.get(&k).expect("entry").slot_index;

        shard.insert(k.clone(), entry(b"xy", 128));

        let replaced = shard.entries.get(&k).expect("entry");
        assert_eq!(replaced.slot_index, first_slot);
        assert_eq!(shard.slots[first_slot].as_ref(), Some(&k));
    }

    #[test]
    fn remove_empties_entry_slot_without_shifting_survivors() {
        let mut shard = Shard::new();
        let a = key("a");
        let b = key("b");
        let c = key("c");
        shard.insert(a.clone(), entry(b"aa", 128));
        shard.insert(b.clone(), entry(b"bb", 128));
        shard.insert(c.clone(), entry(b"cc", 128));
        let b_slot = shard.entries.get(&b).expect("b").slot_index;
        let c_slot = shard.entries.get(&c).expect("c").slot_index;

        shard.remove(&b).expect("removed");

        assert!(shard.slots[b_slot].is_none());
        assert_eq!(shard.entries.get(&c).expect("c").slot_index, c_slot);
        assert_eq!(shard.slots[c_slot].as_ref(), Some(&c));
    }

    #[test]
    fn remove_keeps_hand_valid_after_current_slot() {
        let mut shard = Shard::new();
        let a = key("a");
        let b = key("b");
        shard.insert(a.clone(), entry(b"aa", 128));
        shard.insert(b.clone(), entry(b"bb", 128));

        let removed = shard.remove(&a).expect("removed");
        assert_eq!(removed.size, 2);
        assert_eq!(shard.len(), 1);
        assert_eq!(
            shard.bytes,
            shard.entries.get(&b).expect("entry").l1_charge_bytes
        );
        let evicted = shard.evict_one().expect("remaining entry evicted");
        assert_eq!(evicted.0, b);
    }

    #[test]
    fn priority_biases_eviction_before_falling_back_to_sieve() {
        let mut shard = Shard::new();
        let low = key("low");
        let high = key("high");
        shard.insert(low.clone(), entry(b"low", 1));
        shard.insert(high.clone(), entry(b"hi", 250));

        assert!(matches!(shard.contains(&high, 1_000, 0), Lookup::Present));
        let (evicted_key, _) = shard.evict_one().expect("victim");
        assert_eq!(evicted_key, low);
        assert!(matches!(shard.contains(&high, 1_000, 0), Lookup::Present));
    }

    #[test]
    fn stale_generation_removes_entry() {
        let mut shard = Shard::new();
        let k = key("stale");
        shard.insert(k.clone(), entry(b"v", 128));

        assert!(matches!(shard.get(&k, 1_000, 1), Lookup::Stale(_)));
        assert!(matches!(shard.get(&k, 1_000, 1), Lookup::Miss));
    }

    #[test]
    fn expired_entry_reports_absent_after_removal() {
        let mut shard = Shard::new();
        let k = key("ttl");
        let mut e = entry(b"v", 128);
        e.expires_at_unix_ms = Some(1_000);
        shard.insert(k.clone(), e);

        assert!(matches!(shard.get(&k, 1_000, 0), Lookup::Expired(_)));
        assert_eq!(
            CachePresence::from(matches!(shard.get(&k, 1_000, 0), Lookup::Hit(_))),
            CachePresence::Absent
        );
    }

    #[test]
    fn expiry_sweep_reads_only_due_deadline_records() {
        let mut shard = Shard::new();
        for index in 0..10_000 {
            let key = key(&format!("permanent-{index}"));
            shard.insert(key, entry(b"v", 128));
        }
        let expiring_key = key("expiring");
        let mut expiring = entry(b"v", 128);
        expiring.expires_at_unix_ms = Some(1_000);
        shard.insert(expiring_key.clone(), expiring);

        let sweep = shard.sweep_expired(1_000, 16, Instant::now(), None);

        assert_eq!(sweep.entries_scanned, 1);
        assert_eq!(sweep.removed.len(), 1);
        assert_eq!(sweep.removed[0].0, expiring_key);
        assert_eq!(shard.len(), 10_000);
    }

    #[test]
    fn stale_expiry_record_cannot_remove_entry_with_new_deadline() {
        let mut shard = Shard::new();
        let cache_key = key("renewed");
        let mut expiring = entry(b"v", 128);
        expiring.expires_at_unix_ms = Some(1_000);
        shard.insert(cache_key.clone(), expiring);
        shard.update_expiry(&cache_key, Some(2_000), 500);

        let early = shard.sweep_expired(1_000, 16, Instant::now(), None);
        assert_eq!(early.entries_scanned, 1);
        assert!(early.removed.is_empty());
        assert!(matches!(shard.get(&cache_key, 1_500, 0), Lookup::Hit(_)));

        let due = shard.sweep_expired(2_000, 16, Instant::now(), None);
        assert_eq!(due.removed.len(), 1);
        assert_eq!(due.removed[0].0, cache_key);
    }

    #[test]
    fn stale_expiry_record_cannot_remove_persisted_entry() {
        let mut shard = Shard::new();
        let cache_key = key("persisted");
        let mut expiring = entry(b"v", 128);
        expiring.expires_at_unix_ms = Some(1_000);
        shard.insert(cache_key.clone(), expiring);
        shard.update_expiry(&cache_key, None, 500);

        let sweep = shard.sweep_expired(1_000, 16, Instant::now(), None);

        assert_eq!(sweep.entries_scanned, 1);
        assert!(sweep.removed.is_empty());
        assert!(matches!(shard.get(&cache_key, 2_000, 0), Lookup::Hit(_)));
    }

    #[test]
    fn repeated_expiry_updates_keep_lazy_index_bounded() {
        let mut shard = Shard::new();
        let cache_key = key("renewed-many-times");
        shard.insert(cache_key.clone(), entry(b"v", 128));

        for deadline in 1..=10_000 {
            shard.update_expiry(&cache_key, Some(deadline), 0);
        }

        assert!(shard.expiry_heap.len() <= 2);
        assert!(matches!(shard.get(&cache_key, 9_999, 0), Lookup::Hit(_)));
    }

    #[test]
    fn removing_last_entry_releases_future_expiry_index_storage() {
        let mut shard = Shard::new();
        let cache_key = key("removed-before-expiry");
        let mut expiring = entry(b"v", 128);
        expiring.expires_at_unix_ms = Some(u64::MAX);
        shard.insert(cache_key.clone(), expiring);
        assert_eq!(shard.expiry_heap.len(), 1);

        shard.remove(&cache_key).expect("entry removed");

        assert!(shard.expiry_heap.is_empty());
        assert_eq!(shard.expiry_heap.capacity(), 0);
    }
}