polyc-query 2026.8.3

//! The per-partition decode cache: a process-wide cache, held by
//! [`crate::authority::QueryAuthority`] and shared (the same `Arc`) with
//! every [`crate::authority::ScopedQuery`] it mints. Its whole purpose is to let a
//! second query over an unchanged partition skip both the replay AND the
//! decode `crate::engine::QueryEngine::build` would otherwise redo from
//! scratch every call.
//!
//! # The lookup protocol
//!
//! For each of a scope's own partitions, the caller
//! (`crate::authority::ScopedQuery::execute`) first reads the partition's
//! record count ([`crate::journal::PartitionJournal::partition_event_count`])
//! and calls [`DecodeCache::lookup`]:
//!
//! - **Same count → [`Lookup::Hit`].** Serve the cached
//!   [`crate::engine::PartitionTables`] as-is. Zero replay, zero decode.
//! - **Higher count → [`Lookup::Tail`].** The partition only grew (an
//!   ordinary append) since this entry was cached. The caller replays from
//!   the cached watermark
//!   ([`crate::journal::PartitionJournal::replay_from_with_positions_bounded`]),
//!   decodes ONLY that tail (`crate::engine::decode_partition_tables`), and
//!   appends it onto the cached tables
//!   ([`crate::engine::PartitionTables::concat`]) via
//!   [`DecodeCache::store_tail`].
//! - **Lower count, or no entry at all → [`Lookup::Miss`].** Full replay and
//!   a fresh decode of every event, via [`DecodeCache::store_full`].
//!
//! **The watermark [`DecodeCache::store_full`]/[`DecodeCache::store_tail`]
//! stores is derived from what the replay ACTUALLY returned — the position
//! immediately following the last replayed event — never from the
//! `event_count` the caller read to make the Hit/Tail/Miss decision above.**
//! Those two reads are separate round trips against a partition that can
//! grow in between: an append landing between the count read and the
//! replay itself makes the replay longer than the count implied. Keying the
//! stored entry to the smaller, stale count would leave a later
//! [`Lookup::Tail`] replay a range already inside the cached tables,
//! duplicating rows on the next [`crate::engine::PartitionTables::concat`]
//! — this is why the caller
//! (`crate::authority::ScopedQuery::resolve_partitions_cached`) always
//! computes the stored watermark from the replay's own last position, with
//! an empty replay keeping the entry's prior watermark unchanged.
//!
//! # Why an explicit invalidation, not just the event count, decides correctness
//!
//! A `(partition, record_count)` watermark alone is unsound: an in-place
//! rewrite can replace a record's payload at the same position, leaving the
//! count untouched — the #216/#860 erasure primitive. A count-only cache
//! would see an unchanged count after an erasure and serve the pre-erasure
//! bytes forever, with no signal that anything changed, which is the worst
//! failure mode this feature could have.
//!
//! What closes that gap is the report, not a counter read back off the
//! journal. Every mutating journal command this process issues — a destroy,
//! an excision, a repair — goes through
//! `crate::authority::QueryAuthority::invalidate_partition` on the way back,
//! driven by the wrapper the Container puts around the whole capability
//! (`crate::projection_signal::ReportingRepair`, #1565 chunk B6). It reports
//! only on the receipt, so a command that committed is reported and one that
//! did not is not, and the entry is dropped outright rather than left to be
//! caught by a later comparison.
//!
//! Everything that moves the count is still caught by the count itself,
//! re-read fresh on every lookup: a higher count is a tail, and a lower one
//! is a hard miss (the `Ordering::Less` arm of [`DecodeCache::lookup`]) —
//! never a watermark trusted ahead of reality. So the residual case is
//! exactly one: a rewrite that preserves the record count *and* whose report
//! was lost. That is bounded by the same assumption this cache already fails
//! closed on — one process writes a partition, and
//! [`CacheConfig::enabled`] is `true` only when the deployment positively
//! declares a single replica (see "The runtime kill switch" below).
//!
//! # Signer soundness across the cache's lifetime
//!
//! `crate::authority::ScopedQuery::trusted_signers` (the deployment's
//! approval-signer public-key allow-list, threaded into the `payments`/
//! `approvals`/`grant_replays` decode folds) is fixed for a
//! [`crate::authority::QueryAuthority`]'s entire lifetime — it comes from
//! the platform approval-signing key, loaded once at process startup and
//! never re-minted (`secret_store.rs`). Since this cache
//! lives exactly as long as the `QueryAuthority` that owns it, every decoded
//! batch it ever holds was decoded against the SAME signer set that would
//! decode it fresh today — there is no window where a cached `payments`/
//! `approvals`/`grant_replays` row reflects a signer key this deployment no
//! longer trusts. A future authority built with a different signer set
//! starts with a brand-new, empty cache (this type carries no cross-process
//! persistence), so there is nothing stale to key by signer generation at
//! all.
//!
//! # Reference data is never cached
//!
//! `crate::engine::ReferenceData` (`personas`/`participations`/
//! `persona_identities`) is Fleet-only, non-journal reference data from
//! `PersonaHost` — a different substrate with its own freshness model, not
//! this cache's per-partition record-count protocol. It is resolved fresh
//! on every Fleet query today (`crate::authority::ScopedQuery::resolve_reference_data`)
//! and this cache does not touch it.
//!
//! # The runtime kill switch
//!
//! This cache's whole design assumes exactly one process ever writes (and
//! therefore ever decodes) a given partition — a second replica racing a
//! DIFFERENT process's writes would let this process serve a watermark that
//! is stale relative to what that OTHER writer just committed, with no
//! invalidation to catch it (the report is this process's own, so it is
//! blind to a peer process's writes by construction).
//! [`CacheConfig::enabled`] is this cache's runtime kill switch, and
//! [`CacheConfig::new`] is deliberately fail-closed: it is `true` only when
//! the caller positively supplies `replicas == Some(1)`, never by a default.
//!
//! **Mechanism chosen**: an explicit `POLYCHROME_REPLICAS` deployment
//! declaration (`crates/control-plane/src/lib.rs` reads it, `None` on
//! missing/unparsable), rather than a live Kubernetes API read at
//! `QueryAuthority::new_state_backed`. This was a deliberate proportionality call, not an
//! oversight: `polyc-query` is a plain (non-foundation) Component
//! (`crates/query/Cargo.toml`), and wiring a Kubernetes client dependency
//! through it (or through this crate's constructor signature) purely to
//! answer a question the deployment's own manifest already answers
//! statelessly would be a real layering cost for a value that is static for
//! the container's entire lifetime — a running pod's replica count is not
//! something that changes under it the way, say, a feature flag does.
//! `scripts/check_control_plane_replicas.py` is extended to fail CI if
//! `manifests/base/control-plane-deployment.yaml`'s `POLYCHROME_REPLICAS`
//! env value ever disagrees with that same manifest's `spec.replicas` — so
//! the two cannot silently drift apart the way #1109's `minReplicas: 2` HPA
//! once did undetected.
//!
//! **Residual risk**: this is a deployment-time declaration, not a live
//! runtime read of the actual running replica count — an operator who
//! manually scales the Deployment out-of-band (`kubectl scale`) without
//! touching the manifest (and therefore without CI ever seeing the change)
//! defeats this guard silently, the same class of gap #1109 traced back to
//! the HPA. The CI cross-check closes the "manifest drift" failure mode, not
//! the "someone bypassed the manifest entirely" one; a live Downward-API/kube
//! read would close that residual gap too, at the layering cost described
//! above. A non-Kubernetes/local run must set `POLYCHROME_REPLICAS=1`
//! explicitly to opt in — the cache defaults OFF ([`CacheConfig::disabled`])
//! for any run that never sets the env var at all, satisfying "explicitly
//! opt-in for non-Kubernetes runs" as directly as "explicit opt-in for a
//! Kubernetes one."
//!
//! # Eviction: memory hygiene, not correctness
//!
//! `~12` Arrow tables per partition, `messages_raw`/`tool_calls_raw`
//! included (full message/tool-call content, uncapped), inside this
//! deployment's 1Gi container (`manifests/base/control-plane-deployment.yaml`)
//! means an unbounded cache is a real risk, not a theoretical one. This
//! cache tracks each entry's Arrow array memory
//! (`RecordBatch::get_array_memory_size`, `crate::engine::PartitionTables::memory_bytes`)
//! and evicts the least-recently-used partition once the total crosses
//! [`CacheConfig::max_bytes`] — see that field's own doc for the default.
//! Size-driven eviction is memory hygiene ONLY: dropping an entry can only
//! cost a rebuild, never an answer, so an eviction policy bug can waste
//! memory or hurt the hit rate but can never serve stale or erased data.

use std::collections::HashMap;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};

use crate::engine::PartitionTables;

/// Runtime configuration for this crate's own (sealed) decode cache.
///
/// The kill switch plus its resource ceilings. `pub`:
/// [`crate::authority::QueryAuthority::new_state_backed`]'s caller (the control plane)
/// must be able to name this type.
#[derive(Debug, Clone)]
pub struct CacheConfig {
    /// The runtime kill switch (see this module's own doc). `false` disables
    /// the cache entirely: every lookup reports a miss unconditionally, and
    /// every store call is a no-op — the exact behavior this crate had
    /// before this cache existed.
    pub enabled: bool,
    /// Total Arrow array memory (`RecordBatch::get_array_memory_size`,
    /// summed across every cached partition's decoded tables) this cache
    /// may hold before its LRU eviction reclaims the least-recently-used
    /// partition. Ignored when `enabled` is `false`.
    ///
    /// Default (via [`CacheConfig::new`]): 64 MiB — a conservative fraction
    /// of the roughly 832 MiB of headroom
    /// `crate::engine::QueryLimits::memory_bytes`'s own doc models below this
    /// container's 1Gi limit (1024 − 192 MiB `FairSpillPool`), leaving room
    /// for that headroom's other steady-state consumers (replay buffers, the
    /// decode fan-out itself, response serialization). Pilot-oriented, not a
    /// measured optimum — raise it once this cache's own hit-rate metrics
    /// (`crate::metrics`) show real pressure.
    pub max_bytes: u64,
    /// Ceiling on the total EFFECTIVE event volume (cached rows plus any
    /// freshly-replayed tail, summed across a scope's own partitions) handed
    /// to `crate::engine::QueryEngine::build_from_tables` — the cached-scan
    /// volume bound `crate::authority::ScopedQuery::enforce_cached_volume_budget`
    /// enforces. Exists because a cache hit reports near-zero REPLAYED
    /// events to `crate::engine::QueryLimits::max_source_events` (replay was
    /// skipped entirely) while `DataFusion` still scans the FULL cached
    /// table — this field closes that gap independently of the replay-based
    /// budget. Ignored when `enabled` is `false` (that budget check does not
    /// run at all in that case; see
    /// `crate::authority::ScopedQuery::enforce_cached_volume_budget`'s doc).
    ///
    /// Default (via [`CacheConfig::new`]): 1,000,000 — double
    /// `crate::engine::QueryLimits::max_source_events`'s own 500,000 default.
    /// A cache hit's cost to `QueryEngine::build_from_tables` is decode-free
    /// (no fan-out, no replay I/O — just registering already-built Arrow
    /// batches), so this ceiling can afford to sit above the replay-based
    /// backstop while still bounding the one real remaining cost: how much
    /// data `DataFusion` itself must plan and scan.
    pub max_cached_source_events: usize,
}

impl CacheConfig {
    /// Cache fully off. The safe default for any caller that has not
    /// positively established single-replica — see this module's own doc.
    #[must_use]
    pub const fn disabled() -> Self {
        Self {
            enabled: false,
            max_bytes: 0,
            max_cached_source_events: 0,
        }
    }

    /// Resolve the kill switch: `replicas == Some(1)` enables the cache
    /// (at this module's own documented default ceilings); anything else —
    /// `None` (missing/unparsable `POLYCHROME_REPLICAS`) or any value other
    /// than exactly `1` — disables it. Deliberately fail-closed: there is no
    /// branch here that enables the cache by omission.
    #[must_use]
    pub const fn new(replicas: Option<u32>) -> Self {
        match replicas {
            Some(1) => Self {
                enabled: true,
                max_bytes: 64 * 1024 * 1024,
                max_cached_source_events: 1_000_000,
            },
            _ => Self::disabled(),
        }
    }
}

/// One partition's cached decode state.
struct CacheEntry {
    /// Total record count this entry's [`Self::tables`] represents — compared
    /// against a fresh
    /// [`crate::journal::PartitionJournal::partition_event_count`] read on
    /// every lookup. Set by [`DecodeCache::store_full`]/[`DecodeCache::store_tail`]
    /// from what the caller's own replay actually returned (the position
    /// immediately following the last replayed event), never from a
    /// `partition_event_count` snapshot taken before that replay ran — see
    /// either store method's own doc for why.
    watermark: u64,
    /// The decoded tables themselves.
    tables: PartitionTables,
    /// Recency stamp ([`DecodeCache::touch`]'s monotonic counter) — the LRU
    /// eviction key.
    last_used: u64,
}

/// What [`DecodeCache::lookup`] found for one partition — see the module
/// doc's "lookup protocol" section.
pub(crate) enum Lookup {
    /// Cache disabled, no entry for this partition, or a count lower than the
    /// one it holds: full replay and a fresh decode of every event.
    Miss,
    /// Higher record count: replay only from `from` (the cached
    /// watermark) and decode only that tail, then
    /// [`crate::engine::PartitionTables::concat`] it onto `base`.
    Tail {
        /// The cached state to concatenate the freshly-decoded tail onto.
        base: PartitionTables,
        /// Journal position to resume replay from
        /// ([`crate::journal::PartitionJournal::replay_from_with_positions_bounded`]).
        from: u64,
    },
    /// The same record count: serve as-is. Zero replay, zero decode.
    Hit(PartitionTables),
}

/// The process-wide per-partition decode cache — see the module doc.
pub(crate) struct DecodeCache {
    config: CacheConfig,
    entries: Mutex<HashMap<String, CacheEntry>>,
    recency: AtomicU64,
}

impl DecodeCache {
    /// Build a cache under `config`. A disabled config still builds a valid,
    /// empty instance — every method degrades to a no-op/`Lookup::Miss`
    /// rather than the caller needing to branch on `config.enabled` itself.
    pub(crate) fn new(config: CacheConfig) -> Self {
        Self {
            config,
            entries: Mutex::new(HashMap::new()),
            recency: AtomicU64::new(0),
        }
    }

    /// The kill switch's current value — `crate::authority::ScopedQuery`
    /// branches its whole per-partition resolution loop on this once per
    /// query, rather than on `config.enabled` directly, so the cache's own
    /// disabled-ness is the single source of truth for "does this query even
    /// attempt a lookup."
    pub(crate) const fn enabled(&self) -> bool {
        self.config.enabled
    }

    /// [`CacheConfig::max_cached_source_events`], for
    /// `crate::authority::ScopedQuery::enforce_cached_volume_budget`.
    pub(crate) const fn max_cached_source_events(&self) -> usize {
        self.config.max_cached_source_events
    }

    /// Look up `partition` against its CALLER-SUPPLIED, freshly-read
    /// `event_count` (a read the caller took just before this call) — see the
    /// module doc's "lookup protocol" section.
    pub(crate) fn lookup(&self, partition: &str, event_count: u64) -> Lookup {
        if !self.config.enabled {
            return Lookup::Miss;
        }
        let touch = self.touch();
        let mut entries = self.entries.lock().expect("poison");
        let Some(entry) = entries.get(partition) else {
            drop(entries);
            return Lookup::Miss;
        };
        // Copy out the owned state this decision needs before touching
        // `entries` again, so the branches below (including a `remove`) are
        // never fighting an outstanding borrow from `entry` itself.
        let watermark = entry.watermark;
        let tables = entry.tables.clone();

        if let Some(entry) = entries.get_mut(partition) {
            entry.last_used = touch;
        }

        match event_count.cmp(&watermark) {
            std::cmp::Ordering::Equal => {
                drop(entries);
                crate::metrics::record_cache_hit();
                Lookup::Hit(tables)
            }
            std::cmp::Ordering::Greater => {
                drop(entries);
                crate::metrics::record_cache_tail();
                Lookup::Tail {
                    base: tables,
                    from: watermark,
                }
            }
            // A lower count means content went away without this cache being
            // told — fail safe by treating it as a miss rather than trusting a
            // watermark ahead of reality.
            std::cmp::Ordering::Less => {
                entries.remove(partition);
                drop(entries);
                crate::metrics::record_cache_full_rebuild();
                Lookup::Miss
            }
        }
    }

    /// Store a freshly-decoded, full replay of `partition` — the
    /// [`Lookup::Miss`] path's result. No-op when disabled.
    ///
    /// `watermark` MUST be derived from what `tables` actually represents —
    /// the position immediately following the last event this call's own
    /// replay actually returned (`0` for a genuinely empty replay) — never
    /// from a `partition_event_count` snapshot read in a
    /// separate round trip before that replay ran. The caller
    /// (`crate::authority::ScopedQuery::resolve_partitions_cached`) reads
    /// that count only to DECIDE Hit/Tail/Miss via [`Self::lookup`]; storing
    /// it back here as the watermark is exactly the stale-watermark race a
    /// prior version of this cache had — an append landing between that
    /// count read and this call's own replay made the replay longer than the
    /// snapshot, so keying the entry to the smaller, stale count left a
    /// later [`Lookup::Tail`] re-replay a range this entry's `tables`
    /// already covered, duplicating rows on concat. See that method's own
    /// doc for the exact derivation.
    pub(crate) fn store_full(&self, partition: &str, watermark: u64, tables: PartitionTables) {
        self.insert(partition, watermark, tables);
    }

    /// Store a merged (cached base + freshly-decoded tail) state — the
    /// [`Lookup::Tail`] path's result, after the caller has already called
    /// [`crate::engine::PartitionTables::concat`]. No-op when disabled.
    ///
    /// `watermark` carries the identical derivation requirement as
    /// [`Self::store_full`]'s own `watermark` parameter — see that doc.
    pub(crate) fn store_tail(&self, partition: &str, watermark: u64, merged: PartitionTables) {
        self.insert(partition, watermark, merged);
    }

    fn insert(&self, partition: &str, watermark: u64, tables: PartitionTables) {
        if !self.config.enabled {
            return;
        }
        let touch = self.touch();
        {
            let mut entries = self.entries.lock().expect("poison");
            entries.insert(
                partition.to_owned(),
                CacheEntry {
                    watermark,
                    tables,
                    last_used: touch,
                },
            );
        }
        self.evict_over_budget();
    }

    /// Evict the least-recently-used partition(s) until the total cached
    /// Arrow array memory is at or under [`CacheConfig::max_bytes`], or only
    /// one entry remains (never evict a cache down to nothing just because
    /// its sole entry alone exceeds the budget — that would make the cache
    /// pointless for a single large conversation; see the module doc's
    /// "Eviction: memory hygiene, not correctness" section for why this
    /// bound is a hygiene ceiling, not a hard memory guarantee).
    #[allow(
        clippy::significant_drop_tightening,
        reason = "the lock must stay held across the whole re-check-then-evict loop (each \
                  iteration's total-bytes check depends on the previous iteration's removal) — \
                  metrics recording is deferred until after the block specifically so the lock \
                  is not held during it, but the loop itself cannot tighten further"
    )]
    fn evict_over_budget(&self) {
        let evicted = {
            let mut entries = self.entries.lock().expect("poison");
            let mut evicted = 0_u32;
            loop {
                let total_bytes: u64 = entries
                    .values()
                    .map(|entry| entry.tables.memory_bytes() as u64)
                    .sum();
                if total_bytes <= self.config.max_bytes || entries.len() <= 1 {
                    break;
                }
                let Some(victim) = entries
                    .iter()
                    .min_by_key(|(_, entry)| entry.last_used)
                    .map(|(partition, _)| partition.clone())
                else {
                    break;
                };
                entries.remove(&victim);
                evicted += 1;
            }
            evicted
        };
        for _ in 0..evicted {
            crate::metrics::record_cache_eviction();
        }
    }

    /// Drop `partition`'s cached state outright, if any.
    ///
    /// Proactive hygiene on a destroy, an excision, a repair, or a migration —
    /// reached from
    /// [`crate::authority::QueryAuthority::invalidate_partition`] once the
    /// command that caused it returned its receipt. Evicting the named
    /// partition unconditionally covers every mutation shape: a destroy, an
    /// excision, or a repair names the partition that changed directly, and a
    /// migration's source and destination are each named by their own command
    /// result.
    pub(crate) fn evict(&self, partition: &str) {
        let mut entries = self.entries.lock().expect("poison");
        if entries.remove(partition).is_some() {
            crate::metrics::record_cache_eviction();
        }
    }

    fn touch(&self) -> u64 {
        self.recency.fetch_add(1, Ordering::Relaxed)
    }

    /// Total Arrow array memory this cache currently holds, across every
    /// cached partition — test-only introspection for the eviction-budget
    /// tests.
    #[cfg(test)]
    fn total_bytes(&self) -> u64 {
        self.entries
            .lock()
            .expect("poison")
            .values()
            .map(|entry| entry.tables.memory_bytes() as u64)
            .sum()
    }

    /// Whether `partition` currently has a cached entry — test-only
    /// introspection.
    #[cfg(test)]
    fn contains(&self, partition: &str) -> bool {
        self.entries.lock().expect("poison").contains_key(partition)
    }
}

#[cfg(test)]
mod tests {
    use polyc_eventlog::Event;
    use polyc_proto::kinds;
    use uuid::Uuid;

    use super::*;
    use crate::engine::decode_partition_tables;

    fn sample_tables(partition: &str, n: usize) -> PartitionTables {
        let events: Vec<(u64, Event)> = (0..n)
            .map(|i| {
                let turn = Uuid::now_v7();
                (
                    i as u64,
                    Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                )
            })
            .collect();
        decode_partition_tables(
            partition,
            &events,
            &[],
            &crate::engine::fixture_handoff_trust(),
        )
        .expect("decode")
    }

    #[test]
    fn disabled_cache_always_misses_and_never_stores() {
        let cache = DecodeCache::new(CacheConfig::disabled());
        assert!(matches!(cache.lookup("conv-a", 3), Lookup::Miss));
        cache.store_full("conv-a", 3, sample_tables("conv-a", 3));
        assert!(!cache.contains("conv-a"));
        assert!(matches!(cache.lookup("conv-a", 3), Lookup::Miss));
    }

    #[test]
    fn the_same_count_is_a_hit() {
        let cache = DecodeCache::new(CacheConfig::new(Some(1)));
        cache.store_full("conv-a", 3, sample_tables("conv-a", 3));
        assert!(matches!(cache.lookup("conv-a", 3), Lookup::Hit(_)));
    }

    #[test]
    fn a_higher_count_is_a_tail() {
        let cache = DecodeCache::new(CacheConfig::new(Some(1)));
        cache.store_full("conv-a", 3, sample_tables("conv-a", 3));
        match cache.lookup("conv-a", 5) {
            Lookup::Tail { from, .. } => assert_eq!(from, 3),
            _ => panic!("expected a tail lookup"),
        }
    }

    /// A count that went DOWN is a hard miss, never a watermark trusted ahead
    /// of reality: content went away without this cache being told.
    #[test]
    fn a_lower_count_is_a_miss_and_drops_the_entry() {
        let cache = DecodeCache::new(CacheConfig::new(Some(1)));
        cache.store_full("conv-a", 3, sample_tables("conv-a", 3));
        assert!(matches!(cache.lookup("conv-a", 2), Lookup::Miss));
        // The stale entry is dropped, not merely bypassed.
        assert!(!cache.contains("conv-a"));
    }

    #[test]
    fn no_entry_is_a_miss() {
        let cache = DecodeCache::new(CacheConfig::new(Some(1)));
        assert!(matches!(cache.lookup("conv-never-seen", 0), Lookup::Miss));
    }

    #[test]
    fn invalidation_evicts_the_named_partition() {
        let cache = DecodeCache::new(CacheConfig::new(Some(1)));
        cache.store_full("conv-a", 3, sample_tables("conv-a", 3));
        assert!(cache.contains("conv-a"));

        cache.evict("conv-a");

        assert!(!cache.contains("conv-a"));
        assert!(matches!(cache.lookup("conv-a", 3), Lookup::Miss));
    }

    /// The erasure case, and the one the report is load-bearing for: an
    /// in-place rewrite leaves the record count untouched, so the count alone
    /// would serve the pre-erasure bytes forever. What refuses them is the
    /// invalidation the mutating command reports on its receipt
    /// (`crate::projection_signal::ReportingRepair` in the Container).
    ///
    /// Two assertions in one test on purpose: the first is what makes the
    /// second mean something, because a cache that missed anyway would pass the
    /// second for the wrong reason.
    #[test]
    fn an_in_place_rewrite_is_refused_by_the_report_the_count_cannot_give() {
        let cache = DecodeCache::new(CacheConfig::new(Some(1)));
        cache.store_full("conv-a", 3, sample_tables("conv-a", 3));
        assert!(
            matches!(cache.lookup("conv-a", 3), Lookup::Hit(_)),
            "an unchanged count is a hit, which is exactly why the report matters"
        );

        cache.evict("conv-a");

        assert!(
            matches!(cache.lookup("conv-a", 3), Lookup::Miss),
            "the reported mutation is what refuses the pre-erasure bytes"
        );
    }

    /// LRU eviction reclaims the least-recently-touched partition once the
    /// total cached bytes cross the configured cap, keeping the total under
    /// (or at) that cap.
    #[test]
    fn eviction_keeps_total_bytes_under_the_cap_once_a_third_partition_is_stored() {
        let one = sample_tables("conv-a", 50);
        let per_entry_bytes = one.memory_bytes() as u64;
        // Room for two full entries but not a third, so inserting a third
        // partition must evict the least-recently-used one of the first two
        // rather than the just-inserted third.
        let max_bytes = per_entry_bytes * 5 / 2;
        let cache = DecodeCache::new(CacheConfig {
            enabled: true,
            max_bytes,
            max_cached_source_events: 1_000_000,
        });

        cache.store_full("conv-a", 50, sample_tables("conv-a", 50));
        cache.store_full("conv-b", 50, sample_tables("conv-b", 50));
        // Touch `conv-b` so `conv-a` is the least-recently-used entry.
        assert!(matches!(cache.lookup("conv-b", 50), Lookup::Hit(_)));
        cache.store_full("conv-c", 50, sample_tables("conv-c", 50));

        assert!(
            cache.total_bytes() <= max_bytes,
            "total cached bytes must stay under the configured cap after eviction"
        );
        assert!(!cache.contains("conv-a"), "the LRU entry must be evicted");
        assert!(cache.contains("conv-b"), "the touched entry must survive");
        assert!(
            cache.contains("conv-c"),
            "the just-inserted entry must survive"
        );
    }

    #[test]
    fn cache_config_new_enables_only_at_exactly_one_replica() {
        assert!(CacheConfig::new(Some(1)).enabled);
        assert!(!CacheConfig::new(Some(2)).enabled);
        assert!(!CacheConfig::new(None).enabled);
        assert!(!CacheConfig::new(Some(0)).enabled);
    }
}