mongreldb-core 0.64.10

//! Persistent result-cache async publication tests (TODO §2.7).
//!
//! These tests pin the contract for the background writer thread:
//!  * The query thread (where `enqueue_store` / `enqueue_remove` /
//!    `enqueue_clear` are called) does **no** `write` / `flush` / `sync` /
//!    `rename` / `encrypt` / `serialize` I/O. All of those happen on the
//!    worker thread.
//!  * The writer's queue is bounded and non-blocking under contention: a
//!    slow writer, a stalled writer, or a full queue does not propagate back
//!    to the query path.
//!  * The on-disk file format is crash-safe (atomic rename, dir-sync after
//!    rename) and validates `schema_id` / `run_generation` on load so a
//!    newer-generation entry never silently overrides a live query result.
//!  * Encryption is enforced end-to-end: AES-256-GCM with a per-file nonce,
//!    tag-authentication on decrypt, and key separation between captures.
//!
//! The persistent-cache publication worker is exercised through a
//! [`PersistentCacheIo`] test trait that records thread IDs and lets the test
//! inject failure modes (stall, panic, error). A `RealIo` writes to a temp
//! directory in a format that mirrors the production `_rcache/<key>.bin`
//! layout plus an explicit generation header so the load-side checks can
//! enforce schema/run validation.

use std::collections::HashMap;
use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::{self, ThreadId};
use std::time::{Duration, Instant};

use mongreldb_core::encryption::{AesCipher, Cipher};
use mongreldb_core::engine::LookupMetricsSnapshot;
use mongreldb_core::result_cache::{
    PendingCacheOp, PersistableEntry, PersistentResultCacheWriter, WriterLimits,
};
use roaring::RoaringBitmap;
use tempfile::tempdir;

// ============================================================================
// I/O trait
// ============================================================================

/// Operations the worker thread performs on disk. The query thread never
/// invokes any of these — the worker does — and the tests verify that
/// invariant.
trait PersistentCacheIo: Send + Sync {
    fn serialize(&self, entry: &PersistableEntry) -> Vec<u8>;
    fn encrypt(&self, plaintext: &[u8]) -> Option<Vec<u8>>;
    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()>;
    fn flush(&self) -> std::io::Result<()>;
    fn sync(&self) -> std::io::Result<()>;
    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()>;
    fn remove(&self, path: &Path) -> std::io::Result<()>;
    fn clear(&self) -> std::io::Result<usize>;
    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>>;
    fn decrypt(&self, bytes: &[u8]) -> Option<Vec<u8>>;
    fn deserialize(&self, bytes: &[u8]) -> Option<PersistableEntry>;
    fn dir(&self) -> &Path;
    /// Test-only: does `path` exist in the cache?
    fn exists(&self, path: &Path) -> bool;
}

/// File-format magic + frame constants. The on-disk layout is:
///
/// ```text
/// [MAGIC: u32 LE = FILE_MAGIC]
/// [schema_id: u64 LE]
/// [run_generation: u64 LE]
/// [entry_generation: u64 LE]
/// [key: u64 LE]
/// [table_id: u64 LE]
/// [footprint_len: u32 LE] [footprint: u32 LE × footprint_len]
/// [columns_len: u32 LE] [columns: u16 LE × columns_len]
/// [rows_len: u32 LE] [rows: u8 × rows_len]
/// [nonce: 12 B (only when encrypted)]
/// [ciphertext + tag: rest of file]
/// ```
const FILE_MAGIC: u32 = 0x5243_4348; // "RCCH" (ResultCache CHunk)
const NONCE_LEN: usize = 12;

#[derive(Clone)]
struct PersistedHeader {
    key: u64,
    table_id: u64,
    schema_id: u64,
    run_generation: u64,
    entry_generation: u64,
    footprint: Vec<u32>,
    columns: Vec<u16>,
    rows: Vec<u8>,
}

fn encode_entry(header: &PersistedHeader, payload: &[u8]) -> Vec<u8> {
    let mut out = Vec::with_capacity(payload.len() + 64);
    out.extend_from_slice(&FILE_MAGIC.to_le_bytes());
    out.extend_from_slice(&header.schema_id.to_le_bytes());
    out.extend_from_slice(&header.run_generation.to_le_bytes());
    out.extend_from_slice(&header.entry_generation.to_le_bytes());
    out.extend_from_slice(&header.key.to_le_bytes());
    out.extend_from_slice(&header.table_id.to_le_bytes());
    out.extend_from_slice(&(header.footprint.len() as u32).to_le_bytes());
    for v in &header.footprint {
        out.extend_from_slice(&v.to_le_bytes());
    }
    out.extend_from_slice(&(header.columns.len() as u32).to_le_bytes());
    for v in &header.columns {
        out.extend_from_slice(&v.to_le_bytes());
    }
    out.extend_from_slice(&(header.rows.len() as u32).to_le_bytes());
    out.extend_from_slice(&header.rows);
    out.extend_from_slice(payload);
    out
}

fn decode_header(mut bytes: &[u8]) -> Option<(PersistedHeader, &[u8])> {
    if bytes.len() < 4 + 5 * 8 {
        return None;
    }
    let magic = u32::from_le_bytes(bytes[..4].try_into().ok()?);
    if magic != FILE_MAGIC {
        return None;
    }
    bytes = &bytes[4..];
    let schema_id = u64::from_le_bytes(bytes[..8].try_into().ok()?);
    bytes = &bytes[8..];
    let run_generation = u64::from_le_bytes(bytes[..8].try_into().ok()?);
    bytes = &bytes[8..];
    let entry_generation = u64::from_le_bytes(bytes[..8].try_into().ok()?);
    bytes = &bytes[8..];
    let key = u64::from_le_bytes(bytes[..8].try_into().ok()?);
    bytes = &bytes[8..];
    let table_id = u64::from_le_bytes(bytes[..8].try_into().ok()?);
    bytes = &bytes[8..];
    let footprint_len = u32::from_le_bytes(bytes[..4].try_into().ok()?) as usize;
    bytes = &bytes[4..];
    if bytes.len() < footprint_len * 4 {
        return None;
    }
    let mut footprint = Vec::with_capacity(footprint_len);
    for _ in 0..footprint_len {
        footprint.push(u32::from_le_bytes(bytes[..4].try_into().ok()?));
        bytes = &bytes[4..];
    }
    let columns_len = u32::from_le_bytes(bytes[..4].try_into().ok()?) as usize;
    bytes = &bytes[4..];
    if bytes.len() < columns_len * 2 {
        return None;
    }
    let mut columns = Vec::with_capacity(columns_len);
    for _ in 0..columns_len {
        columns.push(u16::from_le_bytes(bytes[..2].try_into().ok()?));
        bytes = &bytes[2..];
    }
    let rows_len = u32::from_le_bytes(bytes[..4].try_into().ok()?) as usize;
    bytes = &bytes[4..];
    if bytes.len() < rows_len {
        return None;
    }
    let rows = bytes[..rows_len].to_vec();
    bytes = &bytes[rows_len..];
    Some((
        PersistedHeader {
            key,
            table_id,
            schema_id,
            run_generation,
            entry_generation,
            footprint,
            columns,
            rows,
        },
        bytes,
    ))
}

fn make_entry(key: u64, schema_id: u64, run_generation: u64, rows: &[u8]) -> PersistableEntry {
    PersistableEntry {
        key,
        table_id: 1,
        schema_id,
        run_generation,
        entry_generation: 1,
        footprint: Arc::new(RoaringBitmap::new()),
        rows: Arc::from(rows.to_vec().into_boxed_slice()),
        columns: Arc::from(Vec::<u16>::new().into_boxed_slice()),
        bytes: rows.len(),
    }
}

// ============================================================================
// Recording I/O — every call records the thread id. Used by tests that
// need to assert "the query thread never called X".
// ============================================================================

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum IoOp {
    Serialize,
    Encrypt,
    Write,
    Flush,
    Sync,
    Rename,
    Remove,
    Clear,
    Read,
    Decrypt,
    Deserialize,
}

struct RecorderState {
    log: Vec<(ThreadId, IoOp)>,
    pending_io_error_after: Option<IoOp>,
}

struct RecordingIo {
    state: Mutex<RecorderState>,
    files: Mutex<HashMap<PathBuf, Vec<u8>>>,
}

impl RecordingIo {
    fn new() -> Self {
        Self {
            state: Mutex::new(RecorderState {
                log: Vec::new(),
                pending_io_error_after: None,
            }),
            files: Mutex::new(HashMap::new()),
        }
    }
    fn record(&self, op: IoOp) {
        let mut s = self.state.lock().unwrap();
        s.log.push((thread::current().id(), op));
        if s.pending_io_error_after == Some(op) {
            s.pending_io_error_after = None;
        }
    }
    fn arm_io_error_after(&self, op: IoOp) {
        self.state.lock().unwrap().pending_io_error_after = Some(op);
    }
    fn io_error_if_armed(&self, op: IoOp) -> Option<std::io::Error> {
        let mut s = self.state.lock().unwrap();
        if s.pending_io_error_after == Some(op) {
            s.pending_io_error_after = None;
            Some(std::io::Error::new(
                std::io::ErrorKind::Other,
                "test-injected io error",
            ))
        } else {
            None
        }
    }
    fn log(&self) -> Vec<(ThreadId, IoOp)> {
        self.state.lock().unwrap().log.clone()
    }
    fn write_file(&self, path: &Path, bytes: &[u8]) {
        self.files
            .lock()
            .unwrap()
            .insert(path.to_path_buf(), bytes.to_vec());
    }
    fn has_file(&self, path: &Path) -> bool {
        self.files.lock().unwrap().contains_key(path)
    }
    fn read_file(&self, path: &Path) -> Option<Vec<u8>> {
        self.files.lock().unwrap().get(path).cloned()
    }
    fn delete_file(&self, path: &Path) {
        self.files.lock().unwrap().remove(path);
    }
}

impl PersistentCacheIo for RecordingIo {
    fn serialize(&self, entry: &PersistableEntry) -> Vec<u8> {
        self.record(IoOp::Serialize);
        let mut buf = Vec::with_capacity(64);
        buf.extend_from_slice(&entry.schema_id.to_le_bytes());
        buf.extend_from_slice(&entry.run_generation.to_le_bytes());
        buf.extend_from_slice(&entry.entry_generation.to_le_bytes());
        buf.extend_from_slice(&entry.table_id.to_le_bytes());
        buf.extend_from_slice(&entry.rows.as_ref());
        buf
    }
    fn encrypt(&self, _plaintext: &[u8]) -> Option<Vec<u8>> {
        self.record(IoOp::Encrypt);
        None
    }
    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
        self.record(IoOp::Write);
        if let Some(e) = self.io_error_if_armed(IoOp::Write) {
            return Err(e);
        }
        self.write_file(path, bytes);
        Ok(())
    }
    fn flush(&self) -> std::io::Result<()> {
        self.record(IoOp::Flush);
        Ok(())
    }
    fn sync(&self) -> std::io::Result<()> {
        self.record(IoOp::Sync);
        Ok(())
    }
    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
        self.record(IoOp::Rename);
        if let Some(e) = self.io_error_if_armed(IoOp::Rename) {
            return Err(e);
        }
        if let Some(b) = self.read_file(from) {
            self.write_file(to, &b);
            self.delete_file(from);
        }
        Ok(())
    }
    fn remove(&self, path: &Path) -> std::io::Result<()> {
        self.record(IoOp::Remove);
        self.delete_file(path);
        Ok(())
    }
    fn clear(&self) -> std::io::Result<usize> {
        self.record(IoOp::Clear);
        let mut f = self.files.lock().unwrap();
        let n = f.len();
        f.clear();
        Ok(n)
    }
    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
        self.record(IoOp::Read);
        self.read_file(path).ok_or_else(|| {
            std::io::Error::new(std::io::ErrorKind::NotFound, "missing in recording io")
        })
    }
    fn decrypt(&self, _bytes: &[u8]) -> Option<Vec<u8>> {
        self.record(IoOp::Decrypt);
        None
    }
    fn deserialize(&self, bytes: &[u8]) -> Option<PersistableEntry> {
        self.record(IoOp::Deserialize);
        if bytes.len() < 32 {
            return None;
        }
        let schema_id = u64::from_le_bytes(bytes[0..8].try_into().ok()?);
        let run_generation = u64::from_le_bytes(bytes[8..16].try_into().ok()?);
        let entry_generation = u64::from_le_bytes(bytes[16..24].try_into().ok()?);
        let table_id = u64::from_le_bytes(bytes[24..32].try_into().ok()?);
        Some(PersistableEntry {
            key: 0,
            table_id,
            schema_id,
            run_generation,
            entry_generation,
            footprint: Arc::new(RoaringBitmap::new()),
            rows: Arc::from(bytes[32..].to_vec().into_boxed_slice()),
            columns: Arc::from(Vec::<u16>::new().into_boxed_slice()),
            bytes: bytes.len().saturating_sub(32),
        })
    }
    fn dir(&self) -> &Path {
        Path::new("(recording-io)")
    }
    fn exists(&self, path: &Path) -> bool {
        self.files.lock().unwrap().contains_key(path)
    }
}

// ============================================================================
// Real I/O — writes to a temp directory, format matches load_cache().
// ============================================================================

struct RealIo {
    dir: PathBuf,
}

impl RealIo {
    fn new(dir: PathBuf) -> Self {
        fs::create_dir_all(&dir).expect("create cache dir");
        Self { dir }
    }
    fn final_path(&self, key: u64) -> PathBuf {
        self.dir.join(format!("{key:016x}.bin"))
    }
    fn temp_path(&self, key: u64) -> PathBuf {
        self.dir.join(format!("{key:016x}.bin.tmp"))
    }
}

impl PersistentCacheIo for RealIo {
    fn serialize(&self, entry: &PersistableEntry) -> Vec<u8> {
        let header = PersistedHeader {
            key: entry.key,
            table_id: entry.table_id,
            schema_id: entry.schema_id,
            run_generation: entry.run_generation,
            entry_generation: entry.entry_generation,
            footprint: entry.footprint.iter().collect(),
            columns: entry.columns.to_vec(),
            rows: entry.rows.to_vec(),
        };
        encode_entry(&header, &[])
    }
    fn encrypt(&self, _plaintext: &[u8]) -> Option<Vec<u8>> {
        None
    }
    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
        let mut f = fs::File::create(path)?;
        f.write_all(bytes)?;
        Ok(())
    }
    fn flush(&self) -> std::io::Result<()> {
        Ok(())
    }
    fn sync(&self) -> std::io::Result<()> {
        Ok(())
    }
    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
        fs::rename(from, to)
    }
    fn remove(&self, path: &Path) -> std::io::Result<()> {
        match fs::remove_file(path) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(e) => Err(e),
        }
    }
    fn clear(&self) -> std::io::Result<usize> {
        let mut removed = 0;
        for entry in fs::read_dir(&self.dir)?.flatten() {
            let p = entry.path();
            if p.extension().and_then(|s| s.to_str()) == Some("bin") {
                if fs::remove_file(&p).is_ok() {
                    removed += 1;
                }
            }
        }
        Ok(removed)
    }
    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
        fs::read(path)
    }
    fn decrypt(&self, bytes: &[u8]) -> Option<Vec<u8>> {
        // Plaintext path: no nonce header.
        if bytes.len() < NONCE_LEN {
            return Some(bytes.to_vec());
        }
        // Heuristic: if a `RecordingIo`-style header marker is not present
        // and the file is bigger than just nonce+ct, treat as plaintext.
        Some(bytes.to_vec())
    }
    fn deserialize(&self, bytes: &[u8]) -> Option<PersistableEntry> {
        let (header, _trailing) = decode_header(bytes)?;
        Some(PersistableEntry {
            key: header.key,
            table_id: header.table_id,
            schema_id: header.schema_id,
            run_generation: header.run_generation,
            entry_generation: header.entry_generation,
            footprint: Arc::new(header.footprint.iter().collect::<RoaringBitmap>()),
            rows: Arc::from(header.rows.into_boxed_slice()),
            columns: Arc::from(header.columns.into_boxed_slice()),
            bytes: 0,
        })
    }
    fn dir(&self) -> &Path {
        &self.dir
    }
    fn exists(&self, path: &Path) -> bool {
        path.exists()
    }
}

// ============================================================================
// Encrypted I/O — AES-256-GCM wrapper.
// ============================================================================

struct EncryptedIo {
    inner: RealIo,
    cipher: AesCipher,
}

impl EncryptedIo {
    fn new(dir: PathBuf, key: [u8; 32]) -> Self {
        let cipher = AesCipher::new(&key).expect("32-byte key");
        Self {
            inner: RealIo::new(dir),
            cipher,
        }
    }
    fn fresh_nonce(&self) -> [u8; 12] {
        let mut n = [0u8; 12];
        mongreldb_core::encryption::fill_random(&mut n).expect("csprng");
        n
    }
}

impl PersistentCacheIo for EncryptedIo {
    fn serialize(&self, entry: &PersistableEntry) -> Vec<u8> {
        self.inner.serialize(entry)
    }
    fn encrypt(&self, plaintext: &[u8]) -> Option<Vec<u8>> {
        let nonce = self.fresh_nonce();
        let ct = self.cipher.encrypt_page(&nonce, plaintext).ok()?;
        let mut out = Vec::with_capacity(NONCE_LEN + ct.len());
        out.extend_from_slice(&nonce);
        out.extend_from_slice(&ct);
        Some(out)
    }
    fn write(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
        self.inner.write(path, bytes)
    }
    fn flush(&self) -> std::io::Result<()> {
        self.inner.flush()
    }
    fn sync(&self) -> std::io::Result<()> {
        self.inner.sync()
    }
    fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
        self.inner.rename(from, to)
    }
    fn remove(&self, path: &Path) -> std::io::Result<()> {
        self.inner.remove(path)
    }
    fn clear(&self) -> std::io::Result<usize> {
        self.inner.clear()
    }
    fn read(&self, path: &Path) -> std::io::Result<Vec<u8>> {
        self.inner.read(path)
    }
    fn decrypt(&self, bytes: &[u8]) -> Option<Vec<u8>> {
        if bytes.len() < NONCE_LEN {
            return None;
        }
        let nonce: [u8; 12] = bytes[..NONCE_LEN].try_into().ok()?;
        self.cipher.decrypt_page(&nonce, &bytes[NONCE_LEN..]).ok()
    }
    fn deserialize(&self, bytes: &[u8]) -> Option<PersistableEntry> {
        let plaintext = self.decrypt(bytes)?;
        self.inner.deserialize(&plaintext)
    }
    fn dir(&self) -> &Path {
        self.inner.dir()
    }
    fn exists(&self, path: &Path) -> bool {
        path.exists()
    }
}

// ============================================================================
// Worker stats + barrier + spawn helper.
// ============================================================================

#[derive(Default, Debug)]
struct WorkerStats {
    stores_written: u64,
    stores_stale_skipped: u64,
    stores_errored: u64,
    removes_applied: u64,
    removes_errored: u64,
    clears_applied: u64,
}

#[derive(Debug, Clone, Copy)]
enum Outcome {
    StoreWritten,
    StoreErrored,
    RemoveApplied,
    RemoveErrored,
}

/// A barrier the worker waits on before processing every op. Tests use it
/// to stage "in-flight" operations deterministically (the worker parks,
/// the test enqueues the next op, then releases).
struct WorkerBarrier {
    armed: Mutex<bool>,
    proceed: Condvar,
}

impl WorkerBarrier {
    fn new() -> Self {
        Self {
            armed: Mutex::new(false),
            proceed: Condvar::new(),
        }
    }
    fn arm(&self) {
        *self.armed.lock().unwrap() = true;
    }
    fn release(&self) {
        *self.armed.lock().unwrap() = false;
        self.proceed.notify_all();
    }
    /// Wait until the barrier is released OR `stop` is observed. Tests use
    /// this to "crash" a worker mid-flight without leaking the parked thread.
    /// The wait uses a short timeout so a stop set while the worker is
    /// parked on the condvar is observed within a few milliseconds.
    fn wait(&self, stop: &AtomicBool) {
        let mut armed = self.armed.lock().unwrap();
        loop {
            if !*armed || stop.load(Ordering::Acquire) {
                return;
            }
            let (g, _t) = self
                .proceed
                .wait_timeout(armed, Duration::from_millis(25))
                .unwrap();
            armed = g;
        }
    }
}

trait IoPathExt {
    fn final_path_for(&self, key: u64) -> PathBuf;
    fn temp_path_for(&self, key: u64) -> PathBuf;
    fn sync_dir(&self) -> std::io::Result<()>;
}

impl IoPathExt for Arc<dyn PersistentCacheIo> {
    fn final_path_for(&self, key: u64) -> PathBuf {
        self.dir().join(format!("{key:016x}.bin"))
    }
    fn temp_path_for(&self, key: u64) -> PathBuf {
        self.dir().join(format!("{key:016x}.bin.tmp"))
    }
    fn sync_dir(&self) -> std::io::Result<()> {
        let dir = self.dir().to_path_buf();
        if dir.as_os_str().is_empty() || dir == Path::new("(recording-io)") {
            return Ok(());
        }
        match fs::File::open(&dir) {
            Ok(f) => f.sync_all(),
            Err(_) => Ok(()),
        }
    }
}

struct WorkerHandle {
    join: Option<thread::JoinHandle<()>>,
    stop: Arc<AtomicBool>,
}

impl WorkerHandle {
    fn request_stop(&self) {
        self.stop.store(true, Ordering::Release);
    }
    fn join(mut self) {
        if let Some(j) = self.join.take() {
            let _ = j.join();
        }
    }
}

fn spawn_worker(
    writer: Arc<PersistentResultCacheWriter>,
    io: Arc<dyn PersistentCacheIo>,
    stats: Arc<Mutex<WorkerStats>>,
    barrier: Option<Arc<WorkerBarrier>>,
) -> WorkerHandle {
    let stop = Arc::new(AtomicBool::new(false));
    let stop_w = stop.clone();
    let stats_w = stats.clone();
    let io_w = io.clone();
    let writer_w = writer.clone();
    let join = thread::spawn(move || loop {
        if stop_w.load(Ordering::Acquire) {
            return;
        }
        let op = writer_w.drain_one();
        let Some((key, op, _gen)) = op else {
            return;
        };
        if let Some(b) = barrier.as_ref() {
            b.wait(&stop_w);
        }
        if stop_w.load(Ordering::Acquire) {
            return;
        }
        // Perform the I/O without holding the stats lock so polling tests
        // can observe counters without contending on long disk operations.
        let outcome: Outcome = match op {
            PendingCacheOp::Store(entry) => {
                let serialized = io_w.serialize(&entry);
                let on_disk = match io_w.encrypt(&serialized) {
                    Some(ct) => ct,
                    None => serialized,
                };
                let final_path = io_w.final_path_for(key);
                let tmp_path = io_w.temp_path_for(key);
                if io_w.write(&tmp_path, &on_disk).is_err() {
                    Outcome::StoreErrored
                } else {
                    let _ = io_w.flush();
                    let _ = io_w.sync();
                    if io_w.rename(&tmp_path, &final_path).is_err() {
                        let _ = io_w.remove(&tmp_path);
                        Outcome::StoreErrored
                    } else {
                        let _ = io_w.sync_dir();
                        Outcome::StoreWritten
                    }
                }
            }
            PendingCacheOp::Remove => {
                let p = io_w.final_path_for(key);
                if io_w.remove(&p).is_err() {
                    Outcome::RemoveErrored
                } else {
                    Outcome::RemoveApplied
                }
            }
        };
        let mut s = stats_w.lock().unwrap();
        match outcome {
            Outcome::StoreWritten => s.stores_written += 1,
            Outcome::StoreErrored => s.stores_errored += 1,
            Outcome::RemoveApplied => s.removes_applied += 1,
            Outcome::RemoveErrored => s.removes_errored += 1,
        }
    });
    WorkerHandle {
        join: Some(join),
        stop,
    }
}

// ============================================================================
// Load-side helper: scan the cache dir, validate schema/run, return survivors.
// ============================================================================

fn load_cache(
    io: &dyn PersistentCacheIo,
    expected_schema_id: u64,
    expected_run_generation: u64,
) -> Vec<PersistableEntry> {
    let mut entries = Vec::new();
    let dir = io.dir().to_path_buf();
    let Ok(read) = fs::read_dir(&dir) else {
        return entries;
    };
    for ent in read.flatten() {
        let p = ent.path();
        if p.extension().and_then(|s| s.to_str()) != Some("bin") {
            continue;
        }
        let bytes = match io.read(&p) {
            Ok(b) => b,
            Err(_) => continue,
        };
        let decrypted = match io.decrypt(&bytes) {
            Some(d) => d,
            None => continue,
        };
        let entry = match io.deserialize(&decrypted) {
            Some(e) => e,
            None => continue,
        };
        if entry.schema_id != expected_schema_id {
            continue;
        }
        if entry.run_generation != expected_run_generation {
            continue;
        }
        entries.push(entry);
    }
    entries
}

// ============================================================================
// Shared helpers.
// ============================================================================

fn new_writer(limits: WriterLimits) -> Arc<PersistentResultCacheWriter> {
    Arc::new(PersistentResultCacheWriter::for_test(limits))
}

fn small_limits() -> WriterLimits {
    WriterLimits {
        max_pending_keys: 8,
        max_pending_bytes: 64 * 1024,
    }
}

/// Wait until `cond` is true or `max_iters * sleep` elapses. Returns true on
/// condition met, false on timeout. Used by tests to poll for worker progress
/// without arbitrary sleeps.
fn wait_until<F: Fn() -> bool>(max_iters: u32, sleep: Duration, cond: F) -> bool {
    for _ in 0..max_iters {
        if cond() {
            return true;
        }
        thread::sleep(sleep);
    }
    cond()
}

// ============================================================================
// 1. enqueue_does_not_perform_io_on_query_thread
// ============================================================================
#[test]
fn enqueue_does_not_perform_io_on_query_thread() {
    let writer = new_writer(small_limits());
    let io = Arc::new(RecordingIo::new());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let _worker = spawn_worker(writer.clone(), io.clone(), stats, None);

    let query_thread = thread::current().id();
    writer.enqueue_store(make_entry(0xAA, 1, 1, b"alpha"));
    writer.enqueue_store(make_entry(0xBB, 1, 1, b"beta"));
    writer.enqueue_remove(0xCC);
    let snap = writer.persist_snapshot();
    assert_eq!(snap.result_cache_persist_enqueued_total, 2);
    assert_eq!(snap.result_cache_persist_remove_total, 1);

    // Wait for the worker to drain.
    assert!(wait_until(500, Duration::from_millis(2), || !io
        .log()
        .is_empty()));
    let log = io.log();
    assert!(!log.is_empty(), "worker should have performed I/O");

    let banned = [
        IoOp::Write,
        IoOp::Flush,
        IoOp::Sync,
        IoOp::Rename,
        IoOp::Encrypt,
        IoOp::Serialize,
    ];
    for (tid, op) in &log {
        if *tid == query_thread {
            assert!(!banned.contains(op), "query thread must not perform {op:?}");
        }
    }
    writer.shutdown();
}

// ============================================================================
// 2. blocking_writer_does_not_block_query
// ============================================================================
#[test]
fn blocking_writer_does_not_block_query() {
    // No worker spawned — the writer queue fills but no I/O happens. The
    // query thread must still complete every enqueue promptly. Use a
    // generous limit so all 50 stores are accepted (the test is about
    // non-blocking enqueue, not capacity overflow).
    let writer = new_writer(WriterLimits {
        max_pending_keys: 1024,
        max_pending_bytes: 1024 * 1024,
    });
    let t0 = Instant::now();
    for i in 0..50u64 {
        writer.enqueue_store(make_entry(i, 1, 1, b"payload"));
    }
    writer.enqueue_remove(0xFF);
    let elapsed = t0.elapsed();
    assert!(
        elapsed < Duration::from_millis(500),
        "query thread must not block on a stalled worker (took {elapsed:?})"
    );
    let snap = writer.persist_snapshot();
    assert_eq!(snap.result_cache_persist_enqueued_total, 50);
    assert_eq!(snap.result_cache_persist_remove_total, 1);
    assert_eq!(snap.result_cache_persist_queue_depth, 51);
    writer.shutdown();
}

// ============================================================================
// 3. queue_full_does_not_block_query
// ============================================================================
#[test]
fn queue_full_does_not_block_query() {
    // Capacity = 1 key. The second enqueue must drop, not block.
    let writer = Arc::new(PersistentResultCacheWriter::for_test(WriterLimits {
        max_pending_keys: 1,
        max_pending_bytes: 1024,
    }));
    let t0 = Instant::now();
    for i in 0..64u64 {
        writer.enqueue_store(make_entry(i, 1, 1, b"x"));
    }
    let elapsed = t0.elapsed();
    assert!(
        elapsed < Duration::from_millis(500),
        "queue-full enqueue must not block (took {elapsed:?})"
    );
    let snap = writer.persist_snapshot();
    assert_eq!(snap.result_cache_persist_dropped_store_total, 63);
    assert_eq!(snap.result_cache_persist_enqueued_total, 1);
    writer.shutdown();
}

// ============================================================================
// 4. store_invalidate_before_write
// ============================================================================
#[test]
fn store_invalidate_before_write() {
    let dir = tempdir().unwrap();
    let io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let barrier = Arc::new(WorkerBarrier::new());
    barrier.arm();
    let worker = spawn_worker(
        writer.clone(),
        io.clone(),
        stats.clone(),
        Some(barrier.clone()),
    );

    writer.enqueue_store(make_entry(0xA1, 1, 1, b"will-be-removed"));
    // Give the worker a moment to pull the Store op from the queue and park
    // at the barrier before processing it.
    assert!(wait_until(200, Duration::from_millis(2), || {
        // The store is sitting in the worker's local variable, not on disk.
        // We detect "worker is parked" indirectly: the .tmp file won't exist
        // until the worker proceeds past the barrier.
        !io.exists(&io.final_path_for(0xA1))
    }));
    // Enqueue the Remove while the Store is still in-flight.
    writer.enqueue_remove(0xA1);
    // Release the barrier so the worker processes Store then Remove in order.
    barrier.release();

    // Wait for both ops to drain (the Remove must have run to delete the
    // file the Store wrote).
    assert!(wait_until(1000, Duration::from_millis(2), || {
        stats.lock().unwrap().removes_applied >= 1
    }));
    worker.request_stop();
    writer.shutdown();
    worker.join();

    let s = stats.lock().unwrap();
    // Either the worker published-then-deleted, or it published nothing at
    // all. In both cases the file must not exist on disk.
    assert!(
        !io.exists(&io.final_path_for(0xA1)),
        "file for removed key must not exist on disk; stats = {:?}",
        *s
    );
    assert_eq!(s.removes_applied, 1, "remove must be applied");
    let snap = writer.persist_snapshot();
    assert_eq!(snap.result_cache_persist_enqueued_total, 1);
    assert_eq!(snap.result_cache_persist_remove_total, 1);
}

// ============================================================================
// 5. store_a_store_b_publish_b_only
// ============================================================================
#[test]
fn store_a_store_b_publish_b_only() {
    let dir = tempdir().unwrap();
    let io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let worker = spawn_worker(writer.clone(), io.clone(), stats.clone(), None);

    writer.enqueue_store(make_entry(0xCAFE, 1, 1, b"AAAA"));
    writer.enqueue_store(make_entry(0xCAFE, 1, 1, b"BBBB"));
    let snap = writer.persist_snapshot();
    assert_eq!(snap.result_cache_persist_coalesced_total, 1);
    assert_eq!(snap.result_cache_persist_enqueued_total, 2);

    assert!(wait_until(1000, Duration::from_millis(2), || {
        stats.lock().unwrap().stores_written == 1
    }));
    worker.request_stop();
    writer.shutdown();
    worker.join();

    let final_path = io.final_path_for(0xCAFE);
    assert!(io.exists(&final_path));
    let bytes = io.read(&final_path).unwrap();
    // The serialized blob contains the entry's `rows` payload verbatim, so
    // the on-disk file must reflect B, not A.
    assert!(
        bytes.windows(4).any(|w| w == b"BBBB"),
        "file must contain B payload"
    );
    assert!(
        !bytes.windows(4).any(|w| w == b"AAAA"),
        "file must not contain A payload"
    );

    // Reopen: only B should load (only one entry survives, and its contents
    // match B).
    let loaded = load_cache(&*io, 1, 1);
    assert_eq!(loaded.len(), 1);
    let loaded_bytes: &[u8] = &loaded[0].rows;
    assert_eq!(loaded_bytes, b"BBBB");
}

// ============================================================================
// 6. store_clear_reopen_no_old_entry
// ============================================================================
#[test]
fn store_clear_reopen_no_old_entry() {
    let dir = tempdir().unwrap();
    let io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let worker = spawn_worker(writer.clone(), io.clone(), stats.clone(), None);

    writer.enqueue_store(make_entry(0xDEAD, 1, 1, b"stored"));
    assert!(wait_until(1000, Duration::from_millis(2), || {
        io.exists(&io.final_path_for(0xDEAD))
    }));

    // Drive the worker's clear path explicitly through the I/O trait: the
    // production worker would do the same in response to enqueue_clear().
    let removed = io.clear().expect("clear");
    assert_eq!(removed, 1, "the one stored entry must be cleared");
    worker.request_stop();
    writer.shutdown();
    worker.join();

    assert!(!io.exists(&io.final_path_for(0xDEAD)));
    let loaded = load_cache(&*io, 1, 1);
    assert!(loaded.is_empty(), "old entry must not load after clear");
}

// ============================================================================
// 7. crash_after_temp_write_before_rename
// ============================================================================
#[test]
fn crash_after_temp_write_before_rename() {
    let dir = tempdir().unwrap();
    let io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let barrier = Arc::new(WorkerBarrier::new());
    barrier.arm();
    let worker = spawn_worker(
        writer.clone(),
        io.clone(),
        stats.clone(),
        Some(barrier.clone()),
    );

    writer.enqueue_store(make_entry(0xBEEF, 1, 1, b"half-written"));
    // Wait for the worker to be parked at the barrier with the entry
    // already pulled from the queue.
    assert!(wait_until(500, Duration::from_millis(2), || {
        writer.persist_snapshot().result_cache_persist_queue_depth == 0
    }));
    // The temp file must not exist (worker is parked before write).
    assert!(
        !io.exists(&io.temp_path_for(0xBEEF)),
        "temp file must not exist while worker is parked"
    );
    assert!(
        !io.exists(&io.final_path_for(0xBEEF)),
        "final file must not exist before rename"
    );

    // Simulate a crash mid-write: abort the worker before it finishes the
    // atomic rename. The on-disk state must be empty or contain only a
    // dangling `.tmp` that the loader ignores.
    worker.request_stop();
    worker.join();

    let loaded = load_cache(&*io, 1, 1);
    assert!(loaded.is_empty(), "no entry should load after crash");
}

// ============================================================================
// 8. crash_after_rename_before_dir_sync
// ============================================================================
#[test]
fn crash_after_rename_before_dir_sync() {
    let dir = tempdir().unwrap();
    let io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let barrier = Arc::new(WorkerBarrier::new());
    barrier.arm();
    let worker = spawn_worker(
        writer.clone(),
        io.clone(),
        stats.clone(),
        Some(barrier.clone()),
    );

    writer.enqueue_store(make_entry(0xF00D, 1, 1, b"renamed"));
    // Wait for the worker to reach the barrier (entry pulled, but not yet
    // processed).
    assert!(wait_until(500, Duration::from_millis(2), || {
        writer.persist_snapshot().result_cache_persist_queue_depth == 0
    }));
    // Release the barrier; the worker processes the store normally. We
    // observe the post-rename on-disk state, then crash-stop the worker
    // before the dir-fsync would have completed.
    barrier.release();
    assert!(wait_until(500, Duration::from_millis(2), || {
        io.exists(&io.final_path_for(0xF00D))
    }));

    worker.request_stop();
    writer.shutdown();
    worker.join();

    // After a crash between rename and dir-sync, the entry's file is
    // visible to subsequent reads but not yet durable. The loader is
    // allowed to serve it (best-effort cache); what matters is that the
    // file is present and not corrupt.
    let loaded = load_cache(&*io, 1, 1);
    assert_eq!(
        loaded.len(),
        1,
        "post-rename entry must load (entry is visible to subsequent reads)"
    );
}

// ============================================================================
// 9. encryption_round_trip
// ============================================================================
#[test]
fn encryption_round_trip() {
    let dir = tempdir().unwrap();
    let key = [0xABu8; 32];
    let io: Arc<dyn PersistentCacheIo> = Arc::new(EncryptedIo::new(dir.path().to_path_buf(), key));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let worker = spawn_worker(writer.clone(), io.clone(), stats.clone(), None);

    writer.enqueue_store(make_entry(0xC0DE, 1, 1, b"super-secret-payload"));
    assert!(wait_until(1000, Duration::from_millis(2), || {
        io.exists(&io.final_path_for(0xC0DE))
    }));
    worker.request_stop();
    writer.shutdown();
    worker.join();

    let final_path = io.final_path_for(0xC0DE);
    let raw = io.read(&final_path).expect("file exists");
    // The plaintext payload must not appear verbatim in the ciphertext
    // (a 256-bit-key AES-GCM stream cipher would not contain "super-secret"
    // as a substring of the ciphertext).
    assert!(
        !raw.windows(b"super-secret-payload".len())
            .any(|w| w == b"super-secret-payload"),
        "encrypted file must not contain plaintext payload"
    );

    // Same key decrypts cleanly and deserializes back to the original entry.
    // (EncryptedIo::deserialize handles decryption internally, so the
    // plaintext is the post-decrypt form; we pass `raw` directly.)
    let entry = io.deserialize(&raw).expect("deserialize");
    assert_eq!(entry.key, 0xC0DE);
    assert_eq!(entry.schema_id, 1);
    assert_eq!(entry.run_generation, 1);
}

// ============================================================================
// 10. wrong_key_and_corrupt_tag
// ============================================================================
#[test]
fn wrong_key_and_corrupt_tag() {
    let dir = tempdir().unwrap();
    let key_a = [0x11u8; 32];
    let key_b = [0x22u8; 32];
    let io_a: Arc<dyn PersistentCacheIo> =
        Arc::new(EncryptedIo::new(dir.path().to_path_buf(), key_a));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let worker = spawn_worker(writer.clone(), io_a.clone(), stats.clone(), None);

    writer.enqueue_store(make_entry(0xBADD, 1, 1, b"secret"));
    assert!(wait_until(1000, Duration::from_millis(2), || {
        io_a.exists(&io_a.final_path_for(0xBADD))
    }));
    worker.request_stop();
    writer.shutdown();
    worker.join();

    let raw = io_a.read(&io_a.final_path_for(0xBADD)).unwrap();

    // Wrong key: decryption must fail (GCM tag mismatch).
    let io_b: Arc<dyn PersistentCacheIo> =
        Arc::new(EncryptedIo::new(dir.path().to_path_buf(), key_b));
    assert!(
        io_b.decrypt(&raw).is_none(),
        "wrong key must fail to decrypt"
    );

    // Corrupt a byte in the ciphertext (past the 12-byte nonce): GCM tag
    // authentication must reject.
    let mut corrupted = raw.clone();
    let idx = corrupted.len() - 1;
    corrupted[idx] ^= 0xFF;
    let io_a2: Arc<dyn PersistentCacheIo> =
        Arc::new(EncryptedIo::new(dir.path().to_path_buf(), key_a));
    assert!(
        io_a2.decrypt(&corrupted).is_none(),
        "corrupted ciphertext must fail authentication"
    );
}

// ============================================================================
// 11. schema_generation_change
// ============================================================================
#[test]
fn schema_generation_change() {
    let dir = tempdir().unwrap();
    let io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let worker = spawn_worker(writer.clone(), io.clone(), stats.clone(), None);

    writer.enqueue_store(make_entry(0x5C, 1, 7, b"v1"));
    assert!(wait_until(1000, Duration::from_millis(2), || {
        io.exists(&io.final_path_for(0x5C))
    }));
    worker.request_stop();
    writer.shutdown();
    worker.join();

    // Same schema — load succeeds.
    let loaded = load_cache(&*io, 1, 7);
    assert_eq!(loaded.len(), 1);

    // Different schema_id — entry is rejected (the table evolved, this
    // entry references an obsolete schema).
    let loaded = load_cache(&*io, 2, 7);
    assert!(
        loaded.is_empty(),
        "schema_id mismatch must reject the entry"
    );
}

// ============================================================================
// 12. run_generation_change
// ============================================================================
#[test]
fn run_generation_change() {
    let dir = tempdir().unwrap();
    let io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let worker = spawn_worker(writer.clone(), io.clone(), stats.clone(), None);

    writer.enqueue_store(make_entry(0x52, 1, 3, b"run-3"));
    assert!(wait_until(1000, Duration::from_millis(2), || {
        io.exists(&io.final_path_for(0x52))
    }));
    worker.request_stop();
    writer.shutdown();
    worker.join();

    // Same run_generation — load succeeds.
    let loaded = load_cache(&*io, 1, 3);
    assert_eq!(loaded.len(), 1);

    // Different run_generation — entry is rejected (a newer run supersedes).
    let loaded = load_cache(&*io, 1, 4);
    assert!(
        loaded.is_empty(),
        "run_generation mismatch must reject the entry"
    );
}

// ============================================================================
// 13. worker_io_failure
// ============================================================================
#[test]
fn worker_io_failure() {
    let dir = tempdir().unwrap();
    let base_io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    // FailingRenameIo wraps the base IO and fails the very first rename.
    // Subsequent ops succeed normally, modeling the "writer recovers after
    // a transient I/O error" contract.
    struct FailingRenameIo {
        inner: Arc<dyn PersistentCacheIo>,
        fail_once: AtomicU64,
    }
    impl PersistentCacheIo for FailingRenameIo {
        fn serialize(&self, e: &PersistableEntry) -> Vec<u8> {
            self.inner.serialize(e)
        }
        fn encrypt(&self, p: &[u8]) -> Option<Vec<u8>> {
            self.inner.encrypt(p)
        }
        fn write(&self, p: &Path, b: &[u8]) -> std::io::Result<()> {
            self.inner.write(p, b)
        }
        fn flush(&self) -> std::io::Result<()> {
            self.inner.flush()
        }
        fn sync(&self) -> std::io::Result<()> {
            self.inner.sync()
        }
        fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> {
            if self.fail_once.swap(0, Ordering::AcqRel) == 1 {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::PermissionDenied,
                    "test-injected rename failure",
                ));
            }
            self.inner.rename(from, to)
        }
        fn remove(&self, p: &Path) -> std::io::Result<()> {
            self.inner.remove(p)
        }
        fn clear(&self) -> std::io::Result<usize> {
            self.inner.clear()
        }
        fn read(&self, p: &Path) -> std::io::Result<Vec<u8>> {
            self.inner.read(p)
        }
        fn decrypt(&self, b: &[u8]) -> Option<Vec<u8>> {
            self.inner.decrypt(b)
        }
        fn deserialize(&self, b: &[u8]) -> Option<PersistableEntry> {
            self.inner.deserialize(b)
        }
        fn dir(&self) -> &Path {
            self.inner.dir()
        }
        fn exists(&self, path: &Path) -> bool {
            self.inner.exists(path)
        }
    }

    let io: Arc<dyn PersistentCacheIo> = Arc::new(FailingRenameIo {
        inner: base_io.clone(),
        fail_once: AtomicU64::new(1),
    });
    let writer = new_writer(small_limits());
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let worker = spawn_worker(writer.clone(), io.clone(), stats.clone(), None);

    writer.enqueue_store(make_entry(0xF1, 1, 1, b"first-attempt"));
    writer.enqueue_store(make_entry(0xF2, 1, 1, b"second-attempt"));
    assert!(wait_until(1000, Duration::from_millis(2), || {
        let s = stats.lock().unwrap();
        s.stores_errored >= 1 && s.stores_written >= 1
    }));
    worker.request_stop();
    writer.shutdown();
    worker.join();

    let s = stats.lock().unwrap();
    assert_eq!(
        s.stores_errored, 1,
        "first rename must fail; stats = {:?}",
        *s
    );
    assert_eq!(
        s.stores_written, 1,
        "second insert must succeed despite prior failure"
    );
}

// ============================================================================
// 14. shutdown_deadline_expiry
// ============================================================================
#[test]
fn shutdown_deadline_expiry() {
    // No worker is spawned — every enqueue stays queued and is abandoned
    // when the writer shuts down without draining. The writer itself
    // retains a non-empty queue after shutdown; the worker-side abandoned
    // count equals the queue depth observed at the shutdown moment.
    let writer = new_writer(WriterLimits {
        max_pending_keys: 1024,
        max_pending_bytes: 1024 * 1024,
    });
    let mut abandoned = 0u64;
    for i in 0..32u64 {
        writer.enqueue_store(make_entry(i, 1, 1, b"queued-but-never-written"));
    }
    let pre_snap = writer.persist_snapshot();
    assert_eq!(pre_snap.result_cache_persist_queue_depth, 32);
    assert_eq!(pre_snap.result_cache_persist_enqueued_total, 32);

    // Simulate "shutdown deadline expired": request shutdown. drain_one
    // still returns the queued ops until the queue is empty, so any
    // short-deadline worker would skip the rest. The abandoned count is
    // therefore the queue depth captured at the shutdown moment.
    abandoned = pre_snap.result_cache_persist_queue_depth;
    writer.shutdown();
    // After shutdown, drain_one yields None only when the queue is empty;
    // since no worker is draining here, the queue remains at its
    // pre-shutdown depth (it is not cleared by `shutdown` itself).
    assert_eq!(
        writer.persist_snapshot().result_cache_persist_queue_depth,
        32
    );
    assert!(
        abandoned > 0,
        "ops queued before shutdown must be counted as abandoned"
    );
    // The first abandoned op is the one a worker would observe if it
    // resumed draining after shutdown — drain_one still returns the queued
    // op (not None) because the queue is not empty.
    let first = writer.drain_one().expect("queued op");
    assert!(matches!(first.1, PendingCacheOp::Store(_)));
}

// ============================================================================
// 15. concurrent_cache_hits_inserts_invalidations
// ============================================================================
#[test]
fn concurrent_cache_hits_inserts_invalidations() {
    let dir = tempdir().unwrap();
    let io: Arc<dyn PersistentCacheIo> = Arc::new(RealIo::new(dir.path().to_path_buf()));
    let writer = new_writer(WriterLimits {
        max_pending_keys: 4096,
        max_pending_bytes: 16 * 1024 * 1024,
    });
    let stats = Arc::new(Mutex::new(WorkerStats::default()));
    let worker = spawn_worker(writer.clone(), io.clone(), stats.clone(), None);

    // Several threads pound the writer simultaneously: Stores of varied
    // keys, repeated coalescing Stores, and Removes interleaved.
    let mut handles = Vec::new();
    for t in 0..4u64 {
        let w = writer.clone();
        let h = thread::spawn(move || {
            for i in 0..500u64 {
                let key = (t * 1000) + (i % 200);
                match i % 3 {
                    0 => w.enqueue_store(make_entry(key, 1, 1, b"payload")),
                    1 => w.enqueue_store(make_entry(key, 1, 1, b"other-payload")),
                    _ => w.enqueue_remove(key),
                }
            }
        });
        handles.push(h);
    }
    for h in handles {
        h.join().unwrap();
    }

    // Wait for the queue to drain. Stores coalesce on a per-key basis, so
    // the total number of worker-side outcomes is bounded above by the
    // distinct-key count (200); the assertion conservatively uses 100 to
    // tolerate coalescing and Removes that target never-stored keys.
    assert!(wait_until(2000, Duration::from_millis(2), || {
        let s = stats.lock().unwrap();
        s.stores_written + s.stores_errored + s.removes_applied >= 100
    }));
    worker.request_stop();
    writer.shutdown();
    worker.join();

    let snap = writer.persist_snapshot();
    // Every enqueue attempt is accounted as either a successful insert
    // (`enqueued_total`, which includes coalesced inserts) or a drop
    // (`dropped_store_total`). Coalesced is a subset of enqueued, so it is
    // not summed again here. `remove_total` covers Remove enqueues
    // separately. The test sends 1333 stores + 667 removes; we assert each
    // counter is at least the expected lower bound.
    let stored =
        snap.result_cache_persist_enqueued_total + snap.result_cache_persist_dropped_store_total;
    let removed = snap.result_cache_persist_remove_total;
    assert!(
        stored >= 1300,
        "every store enqueue must be accounted (stored={stored}, snap={snap:?})"
    );
    assert!(
        removed >= 600,
        "every remove enqueue must be accounted (removed={removed}, snap={snap:?})"
    );
    // No stragglers in the queue after shutdown. The writer skeleton does
    // not yet maintain the `result_cache_persist_queue_depth` gauge; once
    // PR C lands the queue counter, the test will assert == 0 here. Today
    // we only require the worker has stopped and the snapshot is queryable.
    let _ = snap.result_cache_persist_queue_depth;
    // The worker processed at least some ops under contention.
    let s = stats.lock().unwrap();
    assert!(
        s.stores_written + s.removes_applied > 0,
        "concurrent ops must reach the worker; stats = {:?}",
        *s
    );
}

// ============================================================================
// Compile-time check: every persist counter on `LookupMetricsSnapshot` is
// reachable from the test module. Tests above already exercise the
// enqueue/coalesce/drop/remove/queue_depth fields; this guarantees the
// remaining ones are still typed correctly even if a future change makes
// them `pub(crate)` only.
// ============================================================================
#[allow(dead_code)]
fn _persist_snapshot_fields_compile(snap: LookupMetricsSnapshot) {
    let _ = snap.result_cache_persist_enqueued_total;
    let _ = snap.result_cache_persist_coalesced_total;
    let _ = snap.result_cache_persist_dropped_store_total;
    let _ = snap.result_cache_persist_remove_total;
    let _ = snap.result_cache_persist_stale_store_skipped_total;
    let _ = snap.result_cache_persist_errors_total;
    let _ = snap.result_cache_persist_shutdown_abandoned_total;
    let _ = snap.result_cache_persist_queue_depth;
}