livekit-telemetry 0.2.0

// Copyright 2026 LiveKit, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::{
    cell::{Cell, RefCell},
    fs, io,
    path::{Path, PathBuf},
    sync::{
        atomic::{AtomicBool, Ordering},
        Mutex,
    },
};

/// Queue of encoded OTLP batches between the exporter (`Exporter`) and the transport.
///
/// The exporter writes every batch here *before* trying the network (write-ahead), uploads
/// oldest-first, and removes what the collector accepted or rejected. Two implementations ship:
/// [`MemoryCache`] (default) and [`FileCache`] (survives crashes and restarts); anything else —
/// a database, a platform store — plugs in through `Telemetry::with_cache`.
///
/// Ids are chosen by the exporter as `<unix ns>-<seq>-<records>-<l|t>-<session>-<project key>`.
/// They hold digits, `a` and `b` (a split's halves), hex or `process`, `l` or `t`, `-`, and a
/// routing key of lowercase letters, digits, `.`, `-`, `_`, `+`, and `%` followed by two uppercase
/// hex digits: characters every platform allows in a file name, Windows included. A batch without a
/// project ends with `-`, because its key is empty. Ids are sortable, so [`pending`](Self::pending)
/// is a plain sort, and prefixed with the creation time so an implementation can expire old batches
/// without touching file timestamps.
/// Implementations bound their own footprint by evicting the oldest batches and report what they
/// evicted, so every loss is counted. A failed write or delete is an `Err` and is never reported
/// as done, with one exception: [`FileCache`] counts an eviction once the file is gone, even when
/// the directory sync after it failed. A failed listing reads as empty, and the next pass tries
/// again.
pub trait BatchCache: Send + Sync {
    /// Store one encoded batch under `id`: once this returns `Ok` the batch is committed (for
    /// [`FileCache`], on disk). Returns the ids of older batches evicted to make room (the
    /// exporter counts their events as dropped).
    fn push(&self, id: &str, body: &[u8]) -> io::Result<Vec<String>>;
    /// Replace the batch `old` with `new` (a split): all of `new` is committed before `old`
    /// goes, and nothing is evicted in between, so a crash can duplicate records but never lose
    /// them. A missing `old` returns [`io::ErrorKind::NotFound`] without storing children;
    /// an error does not promise that a concurrently evicted parent still exists. Returns the
    /// ids evicted afterwards, excluding any parent already reported as evicted.
    fn replace(&self, old: &str, new: &[(String, Vec<u8>)]) -> io::Result<Vec<String>>;
    /// Ids of stored batches, oldest first.
    fn pending(&self) -> Vec<String>;
    /// The body stored under `id`: `ErrorKind::NotFound` when it is gone, another error when it
    /// exists but cannot be read right now (e.g. a locked device's file protection).
    fn read(&self, id: &str) -> io::Result<Vec<u8>>;
    /// Delete one batch; a batch already gone is not an error.
    fn remove(&self, id: &str) -> io::Result<()>;
    /// Discard everything (telemetry disabled: nothing may be replayed later); `Err` when
    /// something could not be deleted.
    fn clear(&self) -> io::Result<()>;
    /// Ids evicted before the exporter could see them (over the bounds when the cache
    /// was opened), handed over once so they are counted as lost.
    fn take_evicted(&self) -> Vec<String> {
        Vec::new()
    }
}

/// Batches a cache will hold regardless of their size. `max_cache_bytes` is the size policy;
/// this only stops a long offline stretch at a 1 s cadence from leaving thousands of tiny files
/// in the directory.
// ponytail: a flat cap rather than another config knob; make it one if a host ever needs
// buffering at a scale where the file count matters.
pub(crate) const MAX_BATCHES: usize = 512;

/// In-memory [`BatchCache`]: batches that could not be uploaded wait for the next attempt,
/// bounded by `max_bytes` and 512 batches (`MAX_BATCHES`, oldest evicted). The newest batch always
/// stays, even when it alone is larger than `max_bytes`. Lost with the process.
// ponytail: a Vec with remove(0) — a handful of small batches, and it shares Vec code the
// binary already has instead of pulling in VecDeque's ring-buffer instantiations.
pub struct MemoryCache {
    batches: Mutex<Vec<(String, Vec<u8>)>>,
    max_bytes: usize,
}

impl MemoryCache {
    /// An empty cache holding at most `max_bytes` of batches, apart from the newest batch, which
    /// always stays.
    pub fn new(max_bytes: u64) -> Self {
        Self {
            batches: Mutex::new(Vec::new()),
            max_bytes: usize::try_from(max_bytes).unwrap_or(usize::MAX),
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, Vec<(String, Vec<u8>)>> {
        self.batches.lock().unwrap_or_else(|e| e.into_inner())
    }

    /// Drop the oldest batches until the rest fit; returns their ids. The newest always stays.
    fn evict(&self, batches: &mut Vec<(String, Vec<u8>)>) -> Vec<String> {
        let mut total: usize = batches.iter().map(|(_, b)| b.len()).sum();
        let mut evicted = Vec::new();
        while (total > self.max_bytes || batches.len() > MAX_BATCHES) && batches.len() > 1 {
            let (id, body) = batches.remove(0);
            total -= body.len();
            evicted.push(id);
        }
        evicted
    }
}

impl BatchCache for MemoryCache {
    fn push(&self, id: &str, body: &[u8]) -> io::Result<Vec<String>> {
        let mut batches = self.lock();
        batches.push((id.to_owned(), body.to_vec()));
        batches.sort_by(|a, b| a.0.cmp(&b.0));
        Ok(self.evict(&mut batches))
    }

    fn replace(&self, old: &str, new: &[(String, Vec<u8>)]) -> io::Result<Vec<String>> {
        let mut batches = self.lock();
        if !batches.iter().any(|(id, _)| id == old) {
            return Err(io::ErrorKind::NotFound.into());
        }
        batches.retain(|(id, _)| id != old);
        batches.extend(new.iter().cloned());
        batches.sort_by(|a, b| a.0.cmp(&b.0));
        Ok(self.evict(&mut batches))
    }

    fn pending(&self) -> Vec<String> {
        self.lock().iter().map(|(id, _)| id.clone()).collect()
    }

    fn read(&self, id: &str) -> io::Result<Vec<u8>> {
        self.lock()
            .iter()
            .find(|(i, _)| i == id)
            .map(|(_, body)| body.clone())
            .ok_or_else(|| io::ErrorKind::NotFound.into())
    }

    fn remove(&self, id: &str) -> io::Result<()> {
        self.lock().retain(|(i, _)| i != id);
        Ok(())
    }

    fn clear(&self) -> io::Result<()> {
        self.lock().clear();
        Ok(())
    }
}

/// The extension of a committed batch file.
const BATCH_EXT: &str = "otlp";
/// The extension of a file being written; a crash leaves it behind, and `FileCache::open` removes
/// it.
const TMP_EXT: &str = "tmp";

/// On-disk [`BatchCache`]: one file per encoded batch, so a crash or an offline shutdown loses
/// nothing and the next launch replays what is left.
///
/// Files are written as `.tmp` and renamed into place, so a crash never leaves a half batch
/// readable; eviction is drop-oldest above `max_bytes` or 512 files (`MAX_BATCHES`), and the
/// newest batch always stays. Age is the exporter's call (24 h, judged by the timestamp in the id
/// rather than file metadata — an Apple "required reason" API).
/// Writes, recovery and clears serialize across instances in this process, including clear's
/// listing through deletion. A clear does not revoke future writes; its caller owns consent.
pub struct FileCache {
    dir: PathBuf,
    max_bytes: u64,
    /// File-count bound: [`MAX_BATCHES`] (lowered only by tests).
    max_batches: usize,
    /// What opening pruned, until [`BatchCache::take_evicted`] collects it.
    evicted: Mutex<Vec<String>>,
    /// Publishing a journaled split failed: listings stop retrying (and logging) until the cache
    /// writes again.
    recovery_failed: std::sync::atomic::AtomicBool,
    /// `recover` runs so far (tests only).
    #[cfg(test)]
    recoveries: std::sync::atomic::AtomicUsize,
    /// Test-only fault points around every storage step (see [`Step`]).
    #[cfg(test)]
    fault: Mutex<Option<Fault>>,
}

/// A storage step: the points a test can fail (return an error) or crash at (panic),
/// deterministically.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Step {
    /// Creating and writing a `.tmp` (the path is passed).
    Write,
    /// `fsync` of a `.tmp`.
    Sync,
    /// Renaming a `.tmp` into place, or into a replacement's journal (`.pend`).
    Rename,
    /// `fsync` of the directory.
    SyncDir,
    /// Deleting a batch (including a replaced one — a split's commit point).
    Delete,
    /// Publishing a committed replacement half (`.pend` → batch).
    Publish,
    /// Clear has taken its batch listing (tests only).
    #[cfg(test)]
    ClearListed,
}

#[cfg(test)]
type Fault = Box<dyn Fn(Step, &Path) -> io::Result<()> + Send + Sync>;

/// The extension of a journaled split half: `<parent>@<half>.pend`.
const JOURNAL_EXT: &str = "pend";
/// Separates parent and half in a journal file name. Batch ids never contain it (see the id
/// alphabet on [`BatchCache`]).
const JOURNAL_SEP: char = '@';

/// Held across a push's write, a split's journaled section and recovery, by every [`FileCache`]
/// in the process: a reconfigure opens a second cache on the directory while the replaced
/// pipeline still drains. Its open must never sweep away that pipeline's `.tmp` mid-write, and
/// its recovery must never see a split half-done — it would delete the journal of a split about
/// to commit, losing the halves.
// ponytail: process-wide rather than per directory. Every push and split holds it for a few
// fsyncs, and it contends only while two pipelines overlap in a reconfigure. One lock per
// directory if that ever matters.
static DIR_LOCK: Mutex<()> = Mutex::new(());

/// Lock order: spill admission, file operation, directory journal, instance state. No host
/// logger runs under these locks: `outside_gate` defers our diagnostics to the outer release.
/// Reentrant custom-cache calls into spill mutations fail with WouldBlock instead of waiting.
// ponytail: process-wide serialization; use per-directory barriers if measured contention warrants it.
static FILE_WRITE_LOCK: Mutex<()> = Mutex::new(());
static SPILL_WRITE_LOCK: Mutex<()> = Mutex::new(());

thread_local! {
    /// The current thread owns spill admission; a custom primary must not wait on itself.
    static SPILL_ACTIVE: Cell<bool> = const { Cell::new(false) };
}

/// Own spill admission until commit/clear completes, including unwinding custom primaries.
struct SpillGuard {
    _lock: std::sync::MutexGuard<'static, ()>,
}

impl Drop for SpillGuard {
    fn drop(&mut self) {
        SPILL_ACTIVE.with(|active| active.set(false));
    }
}

/// Refuse synchronous reentry before taking the process-wide admission barrier.
fn spill_gate() -> io::Result<SpillGuard> {
    if SPILL_ACTIVE.with(Cell::get) {
        return Err(io::Error::new(
            io::ErrorKind::WouldBlock,
            "reentrant telemetry cache mutation",
        ));
    }
    let lock = lock_commit(&SPILL_WRITE_LOCK);
    SPILL_ACTIVE.with(|active| active.set(true));
    Ok(SpillGuard { _lock: lock })
}

/// Acquire a commit barrier; tests observe actual contention rather than waiting a fixed time.
fn lock_commit(lock: &'static Mutex<()>) -> std::sync::MutexGuard<'static, ()> {
    #[cfg(test)]
    if matches!(lock.try_lock(), Err(std::sync::TryLockError::WouldBlock)) {
        tests::COMMIT_WAIT.with(|slot| {
            if let Some(waiting) = slot.borrow().as_ref() {
                waiting();
            }
        });
    }
    lock.lock().unwrap_or_else(|e| e.into_inner())
}

/// Serialize a whole file operation, including clear's listing, and release before diagnostics.
fn file_operation<T>(operation: impl FnOnce() -> T) -> T {
    outside_gate(|| {
        let _gate = lock_commit(&FILE_WRITE_LOCK);
        operation()
    })
}

fn lock_dir() -> std::sync::MutexGuard<'static, ()> {
    DIR_LOCK.lock().unwrap_or_else(|e| e.into_inner())
}

impl FileCache {
    /// Open the cache directory (created if missing; its parent must exist), finish or roll back
    /// a split interrupted by a crash — before anything is evicted — and discard stray or
    /// half-written files (only here: a replaced pipeline may still be writing its own).
    pub fn open(dir: impl Into<PathBuf>, max_bytes: u64) -> io::Result<Self> {
        Self::open_bounded(dir, max_bytes, MAX_BATCHES)
    }

    /// [`open`](Self::open) with the file-count bound in force from the start, so opening-time
    /// pruning applies it.
    fn open_bounded(
        dir: impl Into<PathBuf>,
        max_bytes: u64,
        max_batches: usize,
    ) -> io::Result<Self> {
        file_operation(|| Self::open_locked(dir.into(), max_bytes, max_batches))
    }

    /// Initialize under the file barrier so recovery and pruning cannot cross a clear.
    fn open_locked(dir: PathBuf, max_bytes: u64, max_batches: usize) -> io::Result<Self> {
        create_dir_synced(&dir)?;
        let cache = Self {
            dir,
            max_bytes,
            max_batches,
            evicted: Mutex::new(Vec::new()),
            recovery_failed: Default::default(),
            #[cfg(test)]
            recoveries: Default::default(),
            #[cfg(test)]
            fault: Mutex::new(None),
        };
        cache.recover()?;
        {
            // Under `DIR_LOCK` too: another cache's push or split in progress has `.tmp` files
            // here.
            let _locked = lock_dir();
            for entry in fs::read_dir(&cache.dir)?.flatten() {
                let path = entry.path();
                if !is_batch(&path) && path.extension().is_none_or(|e| e != JOURNAL_EXT) {
                    let _ = fs::remove_file(&path);
                }
            }
        }
        let evicted = cache.prune()?;
        *cache.evicted.lock().unwrap_or_else(|e| e.into_inner()) = evicted;
        Ok(cache)
    }

    /// A cache with a lower file-count bound (tests only).
    #[cfg(test)]
    pub(crate) fn open_with_max_batches(
        dir: impl Into<PathBuf>,
        max_bytes: u64,
        max_batches: usize,
    ) -> io::Result<Self> {
        Self::open_bounded(dir, max_bytes, max_batches)
    }

    /// Fail or crash at storage steps from now on (tests only).
    #[cfg(test)]
    pub(crate) fn inject(
        &self,
        fault: impl Fn(Step, &Path) -> io::Result<()> + Send + Sync + 'static,
    ) {
        *self.fault.lock().unwrap_or_else(|e| e.into_inner()) = Some(Box::new(fault));
    }

    /// Run before a storage step: in tests the injected fault can fail it, otherwise a no-op.
    #[cfg_attr(not(test), expect(unused_variables, reason = "only the test fault hook reads them"))]
    fn step(&self, step: Step, path: &Path) -> io::Result<()> {
        #[cfg(test)]
        if let Some(fault) = self.fault.lock().unwrap_or_else(|e| e.into_inner()).as_ref() {
            return fault(step, path);
        }
        Ok(())
    }

    fn path(&self, id: &str, ext: &str) -> PathBuf {
        self.dir.join(format!("{id}.{ext}"))
    }

    /// A replacement half journaled under its parent: `<parent>@<half>.pend`.
    fn pend(&self, parent: &str, half: &str) -> PathBuf {
        self.dir.join(format!("{parent}{JOURNAL_SEP}{half}.{JOURNAL_EXT}"))
    }

    /// The journal files in the directory.
    fn journal(&self) -> io::Result<Vec<PathBuf>> {
        Ok(fs::read_dir(&self.dir)?
            .flatten()
            .map(|e| e.path())
            .filter(|p| p.extension().is_some_and(|e| e == JOURNAL_EXT))
            .collect())
    }

    /// Write `body` to `tmp` and flush it to the device (`fsync`), so the rename that publishes
    /// it can never expose a file whose contents are still only in the page cache.
    fn write_synced(&self, tmp: &Path, body: &[u8]) -> io::Result<()> {
        let write = |path: &Path| -> io::Result<()> {
            self.step(Step::Write, path)?;
            let mut file = fs::File::create(path)?;
            io::Write::write_all(&mut file, body)?;
            self.step(Step::Sync, path)?;
            file.sync_all()
        };
        match write(tmp) {
            // iOS may purge the whole Caches subdirectory while the app runs: recreate it once
            // (another cache's `open` may have recreated it first).
            Err(err) if err.kind() == io::ErrorKind::NotFound => {
                create_dir_synced(&self.dir)?;
                write(tmp)
            }
            other => other,
        }
    }

    fn rename(&self, from: &Path, to: &Path, step: Step) -> io::Result<()> {
        self.step(step, to)?;
        fs::rename(from, to)
    }

    /// After a committed write: listings may retry a stuck journal again, and the bounds are
    /// enforced. Eviction trouble is logged and never an error, because the write itself is
    /// committed (the caller would otherwise keep a second copy in memory); the next write
    /// enforces the bound again.
    fn after_write(&self) -> Vec<String> {
        self.recovery_failed.store(false, Ordering::Relaxed);
        self.prune().unwrap_or_else(|err| {
            diagnostic(log::Level::Debug, format!("could not prune the cache: {err}"));
            Vec::new()
        })
    }

    /// Make the directory's entries (renames, deletes) durable. Failures are errors; only a
    /// platform that cannot sync a directory at all (not Unix) skips it.
    fn sync_dir(&self) -> io::Result<()> {
        self.step(Step::SyncDir, &self.dir)?;
        #[cfg(unix)]
        fs::File::open(&self.dir)?.sync_all()?;
        Ok(())
    }

    /// The batch ids in the directory, oldest first; an unreadable directory is an error. A
    /// split committed but not yet published (its publish failed) is published first, so its
    /// halves are pending — uploaded, purged and counted — like any other batch.
    fn ids(&self) -> io::Result<Vec<String>> {
        let list = || -> io::Result<(Vec<String>, bool)> {
            let (mut ids, mut journaled) = (Vec::new(), false);
            for path in fs::read_dir(&self.dir)?.flatten().map(|e| e.path()) {
                journaled |= path.extension().is_some_and(|e| e == JOURNAL_EXT);
                if let Some(id) =
                    path.file_stem().and_then(|s| s.to_str()).filter(|_| is_batch(&path))
                {
                    ids.push(id.to_owned());
                }
            }
            Ok((ids, journaled))
        };
        let (mut ids, journaled) = list()?;
        if journaled && !self.recovery_failed.load(Ordering::Relaxed) {
            match self.recover() {
                Ok(()) => ids = list()?.0,
                Err(err) => {
                    self.recovery_failed.store(true, Ordering::Relaxed);
                    diagnostic(
                        log::Level::Debug,
                        format!("journaled split not published yet: {err}; retried after a write"),
                    );
                }
            }
        }
        ids.sort_unstable();
        Ok(ids)
    }

    /// Finish or roll back a split a crash interrupted. Its halves are journaled as
    /// `<parent>@<half>.pend` before the parent is deleted (the commit point): with the parent
    /// still there the journal is discarded (the parent is whole); without it the halves are
    /// published. A rebinding — one half whose id extends its host-less parent's with the project
    /// the parent lacks — is rolled forward instead: a journal file is complete once it exists (its
    /// contents are `fsync`ed before the rename), and only it records the project. Either way
    /// every record is there exactly once. Under [`DIR_LOCK`], so no split is in progress: a
    /// journal next to its parent is a crash's or a failed rollback's.
    fn recover(&self) -> io::Result<()> {
        let _locked = lock_dir();
        #[cfg(test)]
        self.recoveries.fetch_add(1, Ordering::Relaxed);
        let journal = self.journal()?;
        if journal.is_empty() {
            return Ok(());
        }
        for pend in journal {
            let Some((parent, half)) =
                pend.file_stem().and_then(|s| s.to_str()).and_then(|s| s.split_once(JOURNAL_SEP))
            else {
                let _ = fs::remove_file(&pend);
                continue;
            };
            let rebinding = is_rebinding(parent, half);
            if self.path(parent, BATCH_EXT).exists() {
                if !rebinding {
                    fs::remove_file(&pend)?;
                    continue;
                }
                // Parent first: a crash in between leaves a committed journal, published below.
                fs::remove_file(self.path(parent, BATCH_EXT))?;
            }
            fs::rename(&pend, self.path(half, BATCH_EXT))?;
        }
        self.sync_dir()
    }

    /// Delete the oldest batches until the rest fit `max_bytes` and the file-count bound (a
    /// committed split is published by the listing first). The newest batch is never a
    /// candidate, whatever happened to the older ones, as in [`MemoryCache`], so it gets its
    /// upload attempt. Returns the ids this pass removed. An id counts once this pass unlinked
    /// its file, even when the directory sync after the unlink failed: the count follows the
    /// visible removal and does not promise a delete that survives a crash. A power loss can
    /// bring the file back, and the next open evicts it again or keeps it for upload, like any
    /// other batch. A file that was already gone (another cache on the directory, or an
    /// overlapping pass, evicted it) frees its room but is that pass's eviction, not this one's.
    fn prune(&self) -> io::Result<Vec<String>> {
        let mut removed = Vec::new();
        let kept = self.ids()?;
        let sizes: Vec<u64> = kept
            .iter()
            .map(|id| fs::metadata(self.path(id, BATCH_EXT)).map(|m| m.len()).unwrap_or(0))
            .collect();
        let mut total: u64 = sizes.iter().sum();
        let mut count = kept.len();
        for (id, len) in kept.iter().zip(sizes).take(kept.len().saturating_sub(1)) {
            if total <= self.max_bytes && count <= self.max_batches {
                break;
            }
            // Gone, by this pass or another, frees its room. Only the pass that unlinked it
            // reports it, whatever its directory sync said afterwards. A batch that could not
            // be deleted still takes its room: keep evicting.
            let (mine, _) = self.unlink(id);
            let went = mine || matches!(self.path(id, BATCH_EXT).try_exists(), Ok(false));
            if went {
                total -= len;
                count -= 1;
                if mine {
                    removed.push(id.clone());
                }
            }
        }
        Ok(removed)
    }

    /// Delete `id`'s file. The flag says whether this call unlinked it (`false`: it was
    /// already gone, by an earlier unlink whose directory sync failed, or by another cache on
    /// the directory). The result is the delete's: an error when the file could not be
    /// deleted or when the directory sync after it failed. The flag is kept apart from the
    /// result, so a failed sync never turns another pass's unlink into this one's. A gone
    /// directory (the OS purged it) has nothing left to sync.
    fn unlink(&self, id: &str) -> (bool, io::Result<()>) {
        let path = self.path(id, BATCH_EXT);
        if let Err(err) = self.step(Step::Delete, &path) {
            return (false, Err(err));
        }
        let unlinked = match fs::remove_file(&path) {
            Ok(()) => true,
            Err(err) if err.kind() == io::ErrorKind::NotFound => false,
            Err(err) => return (false, Err(err)),
        };
        let synced = match self.sync_dir() {
            Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
            other => other,
        };
        (unlinked, synced)
    }

    /// The journaled section of [`BatchCache::replace`], under [`DIR_LOCK`]. `Err` only before the
    /// commit point; a missing `old` returns NotFound. A rebinding has exactly one member: recovery could not
    /// tell a crashed multi-member rebinding from a finished one.
    fn split(&self, old: &str, new: &[(String, Vec<u8>)]) -> io::Result<()> {
        let locked = lock_dir();
        if !self.path(old, BATCH_EXT).try_exists()? {
            return Err(io::ErrorKind::NotFound.into());
        }
        if new.len() > 1 && new.iter().any(|(half, _)| is_rebinding(old, half)) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "a rebinding replaces a batch with exactly one batch",
            ));
        }
        let pends: Vec<PathBuf> = new.iter().map(|(half, _)| self.pend(old, half)).collect();
        let roll_back = |err: io::Error| {
            for ((half, _), pend) in new.iter().zip(&pends) {
                let _ = fs::remove_file(self.path(half, TMP_EXT));
                let _ = fs::remove_file(pend);
            }
            Err(err)
        };
        for ((half, body), pend) in new.iter().zip(&pends) {
            let tmp = self.path(half, TMP_EXT);
            if let Err(err) =
                self.write_synced(&tmp, body).and_then(|()| self.rename(&tmp, pend, Step::Rename))
            {
                return roll_back(err);
            }
        }
        if let Err(err) = self.sync_dir() {
            return roll_back(err);
        }
        if let Err(err) = self
            .step(Step::Delete, &self.path(old, BATCH_EXT))
            .and_then(|()| fs::remove_file(self.path(old, BATCH_EXT)))
        {
            return roll_back(err);
        }
        // Committed: the records are the halves now, whatever follows. A failure from here leaves
        // them journaled — published by the next listing (`ids`), purged by an opt-out — so the
        // split succeeded and must not be read as "the parent is still there".
        let published = self
            .sync_dir()
            .and_then(|()| {
                new.iter().zip(&pends).try_for_each(|((half, _), pend)| {
                    self.rename(pend, &self.path(half, BATCH_EXT), Step::Publish)
                })
            })
            .and_then(|()| self.sync_dir());
        // Logged after the lock: a log line may call into the host.
        drop(locked);
        if let Err(err) = published {
            diagnostic(
                log::Level::Warn,
                format!("split committed but not published yet ({err}); it will be"),
            );
        }
        Ok(())
    }
}

/// The durability boundary: a batch is committed once `push` returns `Ok` — its file was written
/// to a `.tmp`, `fsync`ed, renamed into place and the directory `fsync`ed; if any of that fails
/// the file is removed and `push` errs (the caller keeps the batch in memory instead). A crash
/// before that point loses only the batch being written (its `.tmp` is removed at the next
/// open). Deletes are `fsync`ed the same way. A split (`replace`) is journaled: see
/// `FileCache::recover`.
impl BatchCache for FileCache {
    fn push(&self, id: &str, body: &[u8]) -> io::Result<Vec<String>> {
        file_operation(|| {
            let (tmp, dest) = (self.path(id, TMP_EXT), self.path(id, BATCH_EXT));
            // Under `DIR_LOCK`: another cache opening on the directory sweeps stray `.tmp` files.
            let locked = lock_dir();
            let written = self
                .write_synced(&tmp, body)
                .and_then(|()| self.rename(&tmp, &dest, Step::Rename))
                .and_then(|()| self.sync_dir());
            if let Err(err) = written {
                // Not committed: nothing of it may stay (ENOSPC leaves a truncated `.tmp`, a failed
                // directory sync a published file of unknown durability).
                let _ = fs::remove_file(&tmp);
                let _ = fs::remove_file(&dest);
                return Err(err);
            }
            drop(locked);
            Ok(self.after_write())
        })
    }

    /// Journaled: both halves are written, `fsync`ed and renamed to `<old>@<half>.pend`, the
    /// directory is `fsync`ed, then `old` is deleted — the commit point — and the halves are
    /// published. Before the commit point any failure rolls back; an existing `old` stays whole. After it
    /// the halves are safe in the journal and published by the next listing. Nothing is evicted
    /// until the replacement is complete. A rebinding (a host-less `old` that a half extends with
    /// its project) has exactly one member. Otherwise this returns `InvalidInput` and writes
    /// nothing.
    fn replace(&self, old: &str, new: &[(String, Vec<u8>)]) -> io::Result<Vec<String>> {
        file_operation(|| {
            self.split(old, new)?;
            Ok(self.after_write())
        })
    }

    fn pending(&self) -> Vec<String> {
        file_operation(|| self.ids().unwrap_or_default())
    }

    fn read(&self, id: &str) -> io::Result<Vec<u8>> {
        fs::read(self.path(id, BATCH_EXT))
    }

    fn remove(&self, id: &str) -> io::Result<()> {
        file_operation(|| self.unlink(id).1)
    }

    fn clear(&self) -> io::Result<()> {
        file_operation(|| {
            // Try every file — batches and journaled split halves — then report the first failure,
            // an unreadable directory included. A directory the OS purged is already empty.
            let ids = match self.ids() {
                Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(()),
                other => other?,
            };
            #[cfg(test)]
            self.step(Step::ClearListed, &self.dir)?;
            let mut first = Ok(());
            for id in ids {
                if let Err(err) = self.unlink(&id).1 {
                    first = first.and(Err(err));
                }
            }
            for pend in self.journal()? {
                if let Err(err) = fs::remove_file(&pend) {
                    first = first.and(Err(err));
                }
            }
            // A `.tmp` that a failed write could not remove still holds a batch. Under `DIR_LOCK`, so
            // no push or split is writing one right now.
            {
                let _locked = lock_dir();
                let entries = match fs::read_dir(&self.dir) {
                    Err(err) if err.kind() == io::ErrorKind::NotFound => return first,
                    other => other?,
                };
                for entry in entries.flatten() {
                    let path = entry.path();
                    if path.extension().is_some_and(|e| e == TMP_EXT) {
                        if let Err(err) = fs::remove_file(&path) {
                            first = first.and(Err(err));
                        }
                    }
                }
            }
            // Nothing journaled is left to publish.
            self.recovery_failed.store(false, Ordering::Relaxed);
            first.and_then(|()| self.sync_dir())
        })
    }

    fn take_evicted(&self) -> Vec<String> {
        std::mem::take(&mut *self.evicted.lock().unwrap_or_else(|e| e.into_inner()))
    }
}

/// A cache that never refuses a batch: what the primary cannot store (disk full, directory
/// gone) goes to memory instead, so it still uploads — it just no longer survives the process.
/// Every spill is counted as `cache_write_errors`.
#[cfg_attr(
    not(test),
    expect(dead_code, reason = "the pipeline (`Telemetry`, `Exporter`) uses it, not in place yet")
)]
pub(crate) struct SpillCache {
    primary: std::sync::Arc<dyn BatchCache>,
    spill: MemoryCache,
    counters: std::sync::Arc<crate::stats::Counters>,
    /// Writes check opt-out under the process-wide spill barrier and hold it through primary
    /// commit or memory fallback. Clear takes the same barrier across generations. The SDK
    /// revokes and clears every generation, including each generation's private memory spill.
    revoked: std::sync::Arc<AtomicBool>,
    /// The primary is failing: the first spill of an episode warns, a successful write ends it.
    spilling: AtomicBool,
}

#[cfg_attr(
    not(test),
    expect(dead_code, reason = "the pipeline (`Telemetry`, `Exporter`) uses it, not in place yet")
)]
impl SpillCache {
    pub fn new(
        primary: std::sync::Arc<dyn BatchCache>,
        max_bytes: u64,
        counters: std::sync::Arc<crate::stats::Counters>,
        revoked: std::sync::Arc<AtomicBool>,
    ) -> Self {
        Self {
            primary,
            spill: MemoryCache::new(max_bytes),
            counters,
            revoked,
            spilling: AtomicBool::new(false),
        }
    }

    /// Hold the write gate; `Err` once the app opted out.
    fn open_gate(&self) -> io::Result<SpillGuard> {
        let gate = spill_gate()?;
        if self.revoked.load(Ordering::SeqCst) {
            return Err(io::Error::other("telemetry was disabled"));
        }
        Ok(gate)
    }
}

impl BatchCache for SpillCache {
    fn push(&self, id: &str, body: &[u8]) -> io::Result<Vec<String>> {
        outside_gate(|| {
            let gate = self.open_gate()?;
            let err = match self.primary.push(id, body) {
                Ok(evicted) => {
                    self.spilling.store(false, Ordering::Relaxed);
                    return Ok(evicted);
                }
                Err(err) => err,
            };
            crate::stats::Counters::add(&self.counters.cache_write_errors, 1);
            let first = !self.spilling.swap(true, Ordering::Relaxed);
            let spilled = self.spill.push(id, body);
            // Logged after the gate: a log line may call into the host.
            drop(gate);
            if first {
                log::warn!("cannot write the telemetry cache ({err}); keeping batches in memory");
            }
            spilled
        })
    }

    /// A batch spilled to memory is split in memory; one on disk is split on disk or not at all
    /// — never moved from disk to memory, where a crash would lose it.
    fn replace(&self, old: &str, new: &[(String, Vec<u8>)]) -> io::Result<Vec<String>> {
        outside_gate(|| {
            let _gate = self.open_gate()?;
            if self.spill.pending().iter().any(|id| id == old) {
                self.spill.replace(old, new)
            } else {
                self.primary.replace(old, new)
            }
        })
    }

    fn pending(&self) -> Vec<String> {
        let mut ids = self.primary.pending();
        ids.extend(self.spill.pending());
        ids.sort_unstable();
        ids
    }

    fn read(&self, id: &str) -> io::Result<Vec<u8>> {
        self.spill.read(id).or_else(|_| self.primary.read(id))
    }

    fn remove(&self, id: &str) -> io::Result<()> {
        self.spill.remove(id)?;
        self.primary.remove(id)
    }

    fn clear(&self) -> io::Result<()> {
        outside_gate(|| {
            let _gate = spill_gate()?;
            self.spill.clear()?;
            self.primary.clear()
        })
    }

    fn take_evicted(&self) -> Vec<String> {
        self.primary.take_evicted()
    }
}

thread_local! {
    /// Diagnostics held back while this thread holds the spill gate; `None` logs them at once.
    static HELD_BACK: RefCell<Option<Vec<(log::Level, String)>>> = const { RefCell::new(None) };
}

/// Log `message` now, or once the spill gate this thread holds is released (see
/// [`outside_gate`]): a host logger may call back into the cache, and must not find it locked.
fn diagnostic(level: log::Level, message: String) {
    let now = HELD_BACK.with(|held| match held.borrow_mut().as_mut() {
        Some(queue) => {
            queue.push((level, message));
            None
        }
        None => Some(message),
    });
    if let Some(message) = now {
        log::log!(level, "{message}");
    }
}

/// Run `gated`, which takes the spill gate, and log the diagnostics it raised once it returned,
/// its gate released (into an outer `outside_gate`, if any).
#[cfg_attr(
    not(test),
    expect(dead_code, reason = "`SpillCache` uses it, and the pipeline is not in place yet")
)]
fn outside_gate<T>(gated: impl FnOnce() -> T) -> T {
    struct Release(Option<Vec<(log::Level, String)>>);
    impl Drop for Release {
        fn drop(&mut self) {
            let held =
                HELD_BACK.with(|held| std::mem::replace(&mut *held.borrow_mut(), self.0.take()));
            for (level, message) in held.unwrap_or_default() {
                diagnostic(level, message);
            }
        }
    }
    let _release = Release(HELD_BACK.with(|held| held.borrow_mut().replace(Vec::new())));
    gated()
}

/// Create `dir` (one that exists already is fine) and, when this call created it, make its
/// entry durable by syncing its parent: a directory's own `fsync` covers the entries inside it,
/// not the entry that names it.
// ponytail: `create_dir`, not `create_dir_all` — the recursive variant drags in
// `Path::components` machinery (~3 KiB) for a parent the host always provides.
fn create_dir_synced(dir: &Path) -> io::Result<()> {
    match fs::create_dir(dir) {
        Ok(()) => {
            // A bare relative name (`Path::parent` is the empty path) lives in the current
            // directory: that is the parent to sync.
            let parent =
                dir.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new("."));
            #[cfg(unix)]
            fs::File::open(parent)?.sync_all()?;
            #[cfg(test)]
            tests::SYNCED_PARENT.with(|synced| *synced.borrow_mut() = Some(parent.to_path_buf()));
            Ok(())
        }
        Err(err) if err.kind() == io::ErrorKind::AlreadyExists => Ok(()),
        Err(err) => Err(err),
    }
}

/// A batch file: our extension and a creation stamp in its name. Anything else in the directory
/// (a half-written `.tmp`, a stray or mangled file) is removed by [`FileCache::open`].
fn is_batch(path: &Path) -> bool {
    path.extension().is_some_and(|e| e == BATCH_EXT)
        && path
            .file_stem()
            .and_then(|s| s.to_str())
            .and_then(|s| s.split('-').next())
            .is_some_and(|stamp| stamp.parse::<u64>().is_ok())
}

/// Whether `half` rebinds `parent`: it extends the host-less parent's id (which ends with its
/// empty host) with the project the parent lacks. A split's halves rename the sequence instead.
fn is_rebinding(parent: &str, half: &str) -> bool {
    parent.ends_with('-') && half.len() > parent.len() && half.starts_with(parent)
}

/// A directory no other test uses. Tests run in parallel threads, and the wall clock repeats
/// within its resolution (a microsecond on macOS), so the process id and a counter tell them apart:
/// two tests on one directory would share a cache and delete each other's files.
#[cfg(test)]
pub(crate) fn temp_dir(tag: &str) -> PathBuf {
    temp_dir_at(tag, crate::event::now_unix_nanos())
}

/// [`temp_dir`] at a given wall-clock time.
#[cfg(test)]
fn temp_dir_at(tag: &str, unix_nanos: u64) -> PathBuf {
    static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let dir = std::env::temp_dir()
        .join(format!("livekit-telemetry-{tag}-{}-{n}-{unix_nanos}", std::process::id()));
    let _ = fs::remove_dir_all(&dir);
    dir
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use super::*;

    thread_local! {
        /// The parent directory the last directory creation on this thread synced.
        pub(super) static SYNCED_PARENT: std::cell::RefCell<Option<PathBuf>> =
            const { std::cell::RefCell::new(None) };
        /// Signals an actual contended commit lock, without a timer/readiness guess.
        pub(super) static COMMIT_WAIT: RefCell<Option<Box<dyn Fn()>>> = const { RefCell::new(None) };
    }

    /// Tests in parallel threads ask for directories within one clock tick; each gets its own,
    /// and asking for one never deletes another test's.
    #[test]
    fn temp_dirs_are_unique_within_one_clock_tick() {
        let first = temp_dir_at("same-tick", 42);
        fs::create_dir_all(&first).expect("first directory");
        let second = temp_dir_at("same-tick", 42);
        let kept = first.exists();
        let _ = fs::remove_dir_all(&first);
        assert_ne!(first, second);
        assert!(kept, "the second request deleted the first directory");
    }

    /// Missing parents cannot be resurrected after the eviction was already counted.
    #[test]
    fn replacement_requires_a_parent_in_both_caches() {
        let dir = temp_dir("missing-parent");
        let file = FileCache::open(&dir, 1).unwrap();
        let memory = MemoryCache::new(1);
        let mut results = Vec::new();
        let mut listings = Vec::new();
        for cache in [&file as &dyn BatchCache, &memory as &dyn BatchCache] {
            cache.push(&id(1), b"parent").unwrap();
            assert_eq!(cache.push(&id(2), b"newest").unwrap(), [id(1)]);
            for children in [vec![], vec![(id(3), b"child".to_vec())]] {
                results.push(cache.replace(&id(1), &children).map_err(|e| e.kind()));
                listings.push(cache.pending());
            }
        }
        drop(file);
        fs::remove_dir_all(dir).unwrap();
        assert!(
            listings.iter().all(|ids| ids == &[id(2)]),
            "no children on missing-parent error: {listings:?}"
        );
        assert!(results.iter().all(|r| matches!(r, Err(io::ErrorKind::NotFound))), "{results:?}");
    }

    /// Hold a real write and distinguish a blocked clear from a stale listing deterministically.
    #[test]
    fn standalone_clear_waits_before_listing_another_instances_held_write() {
        use std::{sync::mpsc::sync_channel, thread, time::Duration};
        let dir = temp_dir("clear-held-write");
        let writer = Arc::new(FileCache::open(&dir, 1024).unwrap());
        let clearer = Arc::new(FileCache::open(&dir, 1024).unwrap());
        let writer_weak = Arc::downgrade(&writer);
        let clearer_weak = Arc::downgrade(&clearer);
        let (entered_tx, entered_rx) = sync_channel(1);
        let (write_tx, write_rx) = sync_channel(1);
        let write_rx = Mutex::new(write_rx);
        writer.inject(move |step, _| {
            if step == Step::Write {
                let _ = entered_tx.send(());
                write_rx
                    .lock()
                    .unwrap()
                    .recv_timeout(Duration::from_secs(5))
                    .map_err(io::Error::other)?;
            }
            Ok(())
        });
        let (observed_tx, observed_rx) = sync_channel(2);
        let listed_tx = observed_tx.clone();
        let (clear_tx, clear_rx) = sync_channel(1);
        let clear_rx = Mutex::new(clear_rx);
        clearer.inject(move |step, _| {
            if step == Step::ClearListed {
                let _ = listed_tx.send("listed");
                clear_rx
                    .lock()
                    .unwrap()
                    .recv_timeout(Duration::from_secs(5))
                    .map_err(io::Error::other)?;
            }
            Ok(())
        });
        let writing = thread::spawn(move || writer.push(&id(1), b"payload"));
        let entered = entered_rx.recv_timeout(Duration::from_secs(5));
        let clearing = thread::spawn(move || {
            COMMIT_WAIT.with(|slot| {
                *slot.borrow_mut() = Some(Box::new(move || {
                    let _ = observed_tx.send("blocked");
                }))
            });
            clearer.clear()
        });
        let observed = observed_rx.recv_timeout(Duration::from_secs(5));
        let _ = write_tx.send(());
        let wrote = writing.join();
        let _ = clear_tx.send(());
        let cleared = clearing.join();
        let files: Vec<_> = fs::read_dir(&dir).unwrap().collect();
        fs::remove_dir_all(dir).unwrap();
        assert!(writer_weak.upgrade().is_none() && clearer_weak.upgrade().is_none());
        assert!(entered.is_ok() && matches!(wrote, Ok(Ok(_))) && matches!(cleared, Ok(Ok(()))));
        assert_eq!(observed.unwrap(), "blocked", "clear must wait before taking its listing");
        assert!(files.is_empty(), "a clear cannot miss a held writer's later commit");
    }

    /// A custom primary holds a write after consent admission, before any FileCache gate.
    struct HeldPrimary {
        memory: MemoryCache,
        entered: std::sync::mpsc::SyncSender<()>,
        resume: Mutex<std::sync::mpsc::Receiver<()>>,
        cleared: std::sync::mpsc::SyncSender<&'static str>,
    }

    impl BatchCache for HeldPrimary {
        fn push(&self, id: &str, body: &[u8]) -> io::Result<Vec<String>> {
            let _ = self.entered.send(());
            self.resume
                .lock()
                .unwrap()
                .recv_timeout(std::time::Duration::from_secs(5))
                .map_err(io::Error::other)?;
            self.memory.push(id, body)
        }
        fn replace(&self, old: &str, new: &[(String, Vec<u8>)]) -> io::Result<Vec<String>> {
            self.memory.replace(old, new)
        }
        fn pending(&self) -> Vec<String> {
            self.memory.pending()
        }
        fn read(&self, id: &str) -> io::Result<Vec<u8>> {
            self.memory.read(id)
        }
        fn remove(&self, id: &str) -> io::Result<()> {
            self.memory.remove(id)
        }
        fn clear(&self) -> io::Result<()> {
            let _ = self.cleared.send("cleared");
            self.memory.clear()
        }
    }

    /// A different generation's clear waits through consent admission and custom-cache commit.
    #[test]
    fn spill_clear_waits_for_another_generations_admitted_write() {
        use std::{sync::mpsc::sync_channel, thread, time::Duration};
        let (entered_tx, entered_rx) = sync_channel(1);
        let (resume_tx, resume_rx) = sync_channel(1);
        let (observed_tx, observed_rx) = sync_channel(2);
        let primary = Arc::new(HeldPrimary {
            memory: MemoryCache::new(1024),
            entered: entered_tx,
            resume: Mutex::new(resume_rx),
            cleared: observed_tx.clone(),
        });
        let primary_weak = Arc::downgrade(&primary);
        let revoked = Arc::new(AtomicBool::new(false));
        let writer =
            Arc::new(SpillCache::new(primary.clone(), 1024, Arc::default(), revoked.clone()));
        let clearer =
            Arc::new(SpillCache::new(primary.clone(), 1024, Arc::default(), revoked.clone()));
        let writer_weak = Arc::downgrade(&writer);
        let clearer_weak = Arc::downgrade(&clearer);
        let writing = thread::spawn(move || writer.push(&id(1), b"payload"));
        let entered = entered_rx.recv_timeout(Duration::from_secs(5));
        revoked.store(true, Ordering::SeqCst);
        let clearing = thread::spawn(move || {
            COMMIT_WAIT.with(|slot| {
                *slot.borrow_mut() = Some(Box::new(move || {
                    let _ = observed_tx.send("blocked");
                }))
            });
            clearer.clear()
        });
        let observed = observed_rx.recv_timeout(Duration::from_secs(5));
        let _ = resume_tx.send(());
        let wrote = writing.join();
        let cleared = clearing.join();
        let pending = primary.pending();
        let late = SpillCache::new(primary.clone(), 1024, Arc::default(), revoked);
        let refused = late.push(&id(2), b"late").is_err();
        drop(late);
        drop(primary);
        assert!(
            primary_weak.upgrade().is_none()
                && writer_weak.upgrade().is_none()
                && clearer_weak.upgrade().is_none()
        );
        assert!(entered.is_ok() && matches!(wrote, Ok(Ok(_))) && matches!(cleared, Ok(Ok(()))));
        assert_eq!(observed.unwrap(), "blocked");
        assert!(pending.is_empty() && refused);
    }

    /// Error and unwind paths release both barriers and destroy their cache owners.
    #[test]
    fn commit_barriers_release_after_failed_or_cancelled_writes() {
        for crash in [false, true] {
            let dir = temp_dir("barrier-failure");
            let file = Arc::new(FileCache::open(&dir, 1024).unwrap());
            let weak = Arc::downgrade(&file);
            file.inject(move |step, _| {
                if step == Step::Write {
                    assert!(!crash, "cancelled synchronous write");
                    return Err(io::Error::other("failed write"));
                }
                Ok(())
            });
            let cache = SpillCache::new(file.clone(), 1024, Arc::default(), Arc::default());
            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                cache.push(&id(1), b"body")
            }));
            let spilled = cache.spill.pending();
            file.inject(|_, _| Ok(()));
            let cleared = cache.clear();
            let empty = fs::read_dir(&dir).unwrap().count() == 0 && cache.pending().is_empty();
            drop(cache);
            drop(file);
            fs::remove_dir_all(dir).unwrap();
            assert!(weak.upgrade().is_none() && cleared.is_ok() && empty);
            if crash {
                assert!(result.is_err());
            } else {
                assert!(matches!(result, Ok(Ok(_))) && spilled == [id(1)]);
            }
        }
    }

    /// A custom primary's synchronous reentry fails rather than waiting on its own barrier.
    #[test]
    fn spill_mutation_reentry_returns_would_block() {
        let cache =
            SpillCache::new(Arc::new(MemoryCache::new(1024)), 1024, Arc::default(), Arc::default());
        let held = spill_gate().unwrap();
        let push = cache.push(&id(1), b"body").map_err(|err| err.kind());
        let replace = cache.replace(&id(1), &[]).map_err(|err| err.kind());
        let clear = cache.clear().map_err(|err| err.kind());
        drop(held);
        let next = cache.push(&id(2), b"next");
        let cleared = cache.clear();
        assert_eq!(push, Err(io::ErrorKind::WouldBlock));
        assert_eq!(replace, Err(io::ErrorKind::WouldBlock));
        assert_eq!(clear, Err(io::ErrorKind::WouldBlock));
        assert!(next.is_ok() && cleared.is_ok());
    }

    /// Ids the exporter would mint: a real timestamp (fixed for the test run, so ids stay
    /// stable however slow the disk) + seq.
    fn id(n: u64) -> String {
        static BASE: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
        let base =
            *BASE.get_or_init(|| crate::event::now_unix_nanos() / 1_000_000_000 * 1_000_000_000);
        format!("{:020}-{n:06}-1", base + n)
    }

    #[test]
    fn the_newest_batch_always_stays_whatever_the_cap() {
        let dir = temp_dir("floor");
        let cache = FileCache::open(&dir, 25).expect("open");
        assert_eq!(cache.push(&id(1), &[b'a'; 30]).expect("push"), Vec::<String>::new());
        assert_eq!(cache.pending(), [id(1)], "a batch over the cap still gets its upload");
        assert_eq!(cache.push(&id(2), &[b'b'; 30]).expect("push"), [id(1)]);
        assert_eq!(cache.pending(), [id(2)]);
        let _ = fs::remove_dir_all(&dir);
        let dir = temp_dir("floor-zero");
        let cache = FileCache::open(&dir, 0).expect("open");
        cache.push(&id(1), b"x").expect("push");
        assert_eq!(cache.pending(), [id(1)], "a cap of 0 keeps the push");
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn an_eviction_whose_directory_sync_fails_still_counts() {
        use std::sync::atomic::{AtomicBool, Ordering};
        let dir = temp_dir("evict-sync");
        let cache = FileCache::open(&dir, 10).expect("open");
        cache.push(&id(1), &[b'a'; 10]).expect("push");
        let deleted = Arc::new(AtomicBool::new(false));
        cache.inject(move |step, _| {
            if step == Step::Delete {
                deleted.store(true, Ordering::SeqCst);
            }
            if step == Step::SyncDir && deleted.load(Ordering::SeqCst) {
                return Err(io::Error::other("injected"));
            }
            Ok(())
        });
        assert_eq!(cache.push(&id(2), &[b'b'; 10]).expect("push"), [id(1)]);
        assert_eq!(cache.pending(), [id(2)], "the batch just committed stays");
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn a_purged_directory_is_already_empty() {
        let dir = temp_dir("purged");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        cache.push(&id(1), b"x").expect("push");
        fs::remove_dir_all(&dir).expect("the OS purges the directory");
        assert!(cache.remove(&id(1)).is_ok());
        assert!(cache.clear().is_ok(), "an opt-out with nothing on disk succeeds");
    }

    #[test]
    fn a_rebinding_replaces_a_batch_with_exactly_one() {
        let dir = temp_dir("rebinding");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        let parent = format!("{}-l-s-", id(1));
        cache.push(&parent, b"parent").expect("push");
        let two = [
            (format!("{parent}a.example"), b"a".to_vec()),
            (format!("{parent}b.example"), b"b".to_vec()),
        ];
        let err = cache.replace(&parent, &two).expect_err("rejected");
        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
        assert_eq!(cache.pending(), std::slice::from_ref(&parent), "the parent is kept");
        let one = [(format!("{parent}a.example"), b"a".to_vec())];
        cache.replace(&parent, &one).expect("one member");
        assert_eq!(cache.pending(), [format!("{parent}a.example")]);
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn clear_lets_listings_publish_a_journal_again() {
        let dir = temp_dir("clear-recovery");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        cache.recovery_failed.store(true, Ordering::Relaxed);
        cache.clear().expect("clear");
        assert!(!cache.recovery_failed.load(Ordering::Relaxed));
        let _ = fs::remove_dir_all(&dir);
    }

    /// A second cache opening on the directory (a reconfigure) never sweeps away the `.tmp` of a
    /// push still in flight.
    #[test]
    fn a_second_cache_never_breaks_a_push_in_progress() {
        let dir = temp_dir("push-two-generations");
        let old = FileCache::open(&dir, 1 << 20).expect("open");
        let (opened, path) = (Arc::new(Mutex::new(None)), dir.clone());
        let handle = opened.clone();
        old.inject(move |step, _| {
            let mut handle = handle.lock().expect("lock");
            if step == Step::Rename && handle.is_none() {
                let path = path.clone();
                *handle = Some(std::thread::spawn(move || {
                    FileCache::open(&path, 1 << 20).expect("open");
                }));
                // Give it every chance to sweep the directory mid-push.
                std::thread::sleep(std::time::Duration::from_millis(100));
            }
            Ok(())
        });
        old.push(&id(1), b"in flight").expect("the push commits");
        opened.lock().expect("lock").take().expect("opened").join().expect("open succeeded");
        assert_eq!(old.pending(), [id(1)]);
        let _ = fs::remove_dir_all(&dir);
    }

    /// A `.tmp` that a failed write could not remove (the directory went read-only) is deleted by
    /// the next `clear` once the directory is writable again: an opt-out leaves no batch bytes.
    #[cfg(unix)]
    #[test]
    fn clear_deletes_a_tmp_a_failed_write_left_behind() {
        use std::os::unix::fs::PermissionsExt;
        let dir = temp_dir("clear-tmp");
        let cache = FileCache::open(&dir, 1024).expect("open");
        cache.inject(|step, path| {
            if step == Step::Sync {
                let parent = path.parent().expect("the cache directory");
                fs::set_permissions(parent, fs::Permissions::from_mode(0o555))?;
                return Err(io::Error::other("disk gone"));
            }
            Ok(())
        });
        let pushed = cache.push(&id(1), b"telemetry bytes");
        fs::set_permissions(&dir, fs::Permissions::from_mode(0o755)).expect("writable again");
        cache.inject(|_, _| Ok(()));
        assert!(pushed.is_err());
        assert_eq!(fs::read_dir(&dir).expect("dir").count(), 1, "the failed write left its .tmp");
        cache.clear().expect("clear");
        let left: Vec<_> = fs::read_dir(&dir).expect("dir").flatten().map(|e| e.path()).collect();
        let _ = fs::remove_dir_all(&dir);
        assert!(left.is_empty(), "{left:?}");
    }

    /// A warning raised inside a gated write reaches the host logger only once the spill gate is
    /// released: a logger that calls back into the cache (a purge, say) does not deadlock.
    #[test]
    fn diagnostics_reach_the_logger_outside_the_spill_gate() {
        crate::test_log::capture();
        let dir = temp_dir("diagnostics-outside-the-gate");
        let primary = Arc::new(FileCache::open(&dir, 1024).expect("open"));
        primary.push(&id(1), b"parent").expect("push");
        primary.inject(|step, _| match step {
            Step::Publish => Err(io::Error::other("publish failed")),
            _ => Ok(()),
        });
        let cache = Arc::new(SpillCache::new(primary, 1024, Arc::default(), Arc::default()));
        let gated = Arc::downgrade(&cache);
        crate::test_log::mark_while(move || {
            gated.upgrade().is_some_and(|_| SPILL_ACTIVE.with(Cell::get))
        });
        let halves =
            [(format!("{}a", id(1)), b"a".to_vec()), (format!("{}b", id(1)), b"b".to_vec())];
        cache.replace(&id(1), &halves).expect("split");
        let lines = crate::test_log::take();
        let _ = fs::remove_dir_all(&dir);
        assert!(lines.iter().any(|l| l.contains("split committed but not published")), "{lines:?}");
        assert!(lines.iter().all(|l| !l.starts_with("held: ")), "{lines:?}");
    }

    /// A primary that fails while its flag is set.
    struct Flaky(std::sync::atomic::AtomicBool);

    impl BatchCache for Flaky {
        fn push(&self, _: &str, _: &[u8]) -> io::Result<Vec<String>> {
            if self.0.load(Ordering::SeqCst) {
                return Err(io::Error::other("disk full"));
            }
            Ok(Vec::new())
        }
        fn replace(&self, _: &str, _: &[(String, Vec<u8>)]) -> io::Result<Vec<String>> {
            Ok(Vec::new())
        }
        fn pending(&self) -> Vec<String> {
            Vec::new()
        }
        fn read(&self, _: &str) -> io::Result<Vec<u8>> {
            Err(io::ErrorKind::NotFound.into())
        }
        fn remove(&self, _: &str) -> io::Result<()> {
            Ok(())
        }
        fn clear(&self) -> io::Result<()> {
            Ok(())
        }
    }

    #[test]
    fn each_spill_episode_warns_once() {
        crate::test_log::capture();
        let primary = Arc::new(Flaky(true.into()));
        let counters = Arc::new(crate::stats::Counters::default());
        let cache = SpillCache::new(primary.clone(), 1 << 20, counters.clone(), Arc::default());
        let warning = "cannot write the telemetry cache (disk full); keeping batches in memory";
        cache.push(&id(1), b"a").expect("spilled");
        cache.push(&id(2), b"b").expect("spilled");
        assert_eq!(crate::test_log::take(), [warning], "two spills, one warning");
        primary.0.store(false, Ordering::SeqCst);
        cache.push(&id(3), b"c").expect("written");
        primary.0.store(true, Ordering::SeqCst);
        cache.push(&id(4), b"d").expect("spilled");
        assert_eq!(crate::test_log::take(), [warning], "a new episode warns again");
        assert_eq!(counters.snapshot().cache_write_errors, 3);
    }

    #[test]
    fn memory_cache_evicts_oldest_beyond_max_bytes() {
        let cache = MemoryCache::new(25);
        for (n, body) in [b"aaaaaaaaaa", b"bbbbbbbbbb", b"cccccccccc"].into_iter().enumerate() {
            cache.push(&id(n as u64), body).expect("push");
        }
        assert_eq!(cache.pending(), [id(1), id(2)]);
        assert_eq!(cache.read(&id(1)).expect("read"), b"bbbbbbbbbb");
        cache.remove(&id(1)).expect("remove");
        assert_eq!(cache.pending(), [id(2)]);
        cache.clear().expect("clear");
        assert!(cache.pending().is_empty());
    }

    #[test]
    fn file_cache_evicts_oldest_beyond_max_bytes_and_drops_stray_tmp() {
        let dir = temp_dir("cache");
        let cache = FileCache::open(&dir, 25).expect("open");
        fs::write(dir.join("crashed.tmp"), b"half").expect("write");
        for (n, body) in [b"aaaaaaaaaa", b"bbbbbbbbbb", b"cccccccccc"].into_iter().enumerate() {
            cache.push(&id(n as u64), body).expect("push");
        }
        assert_eq!(cache.pending(), [id(1), id(2)], "10-byte batches under a 25-byte cap");
        assert_eq!(cache.read(&id(1)).expect("read"), b"bbbbbbbbbb");
        // A stray `.tmp` goes at the next open, never mid-run: it may be a replaced pipeline's
        // write in progress.
        assert!(dir.join("crashed.tmp").exists());
        let cache = FileCache::open(&dir, 25).expect("reopen");
        assert!(!dir.join("crashed.tmp").exists());
        cache.clear().expect("clear");
        assert!(cache.pending().is_empty());
        let _ = fs::remove_dir_all(&dir);
    }

    /// Days offline at a one-second cadence are tiny batches, not big ones: the byte cap alone
    /// would leave thousands of files in the directory.
    #[test]
    fn file_cache_caps_the_number_of_batches() {
        let dir = temp_dir("count");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        // `id` stamps the current second, so keep the ids rather than recomputing them.
        let ids: Vec<String> = (0..(MAX_BATCHES as u64 + 8)).map(id).collect();
        for batch in &ids {
            cache.push(batch, b"x").expect("push");
        }
        let kept = cache.pending();
        assert_eq!(kept.len(), MAX_BATCHES, "far under the byte cap, still bounded");
        assert_eq!(kept.first(), Some(&ids[8]), "the oldest went first");
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn file_cache_recreates_a_purged_directory() {
        let dir = temp_dir("purged");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        fs::remove_dir_all(&dir).expect("purge like iOS does");
        cache.push(&id(1), b"after purge").expect("push recreates the dir");
        assert_eq!(cache.pending(), [id(1)]);
        let _ = fs::remove_dir_all(&dir);
    }

    #[cfg(unix)]
    #[test]
    fn file_cache_failed_write_leaves_no_partial_file() {
        use std::os::unix::fs::PermissionsExt;
        let dir = temp_dir("readonly");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        // Stand-in for ENOSPC: any write into the directory fails.
        fs::set_permissions(&dir, fs::Permissions::from_mode(0o555)).expect("chmod");
        let result = cache.push(&id(1), b"no room");
        fs::set_permissions(&dir, fs::Permissions::from_mode(0o755)).expect("chmod back");
        assert!(result.is_err());
        assert_eq!(fs::read_dir(&dir).expect("dir").count(), 0, "no stray .tmp");
        let _ = fs::remove_dir_all(&dir);
    }

    /// A real split, crashed at every one of its storage steps and
    /// reopened under byte and file-count pressure, leaves every record exactly once — the
    /// parent whole, or both halves — and recovery runs before anything is evicted.
    #[test]
    fn a_split_crashed_at_every_step_loses_and_duplicates_nothing() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        let (parent, a, b) = (id(1), format!("{}a", id(1)), format!("{}b", id(1)));
        let halves = [(a.clone(), vec![b'a'; 60]), (b.clone(), vec![b'b'; 60])];
        // Room for the parent and not a byte more, or for the two halves and not a file more: a
        // duplicate would force an eviction.
        for (max_bytes, max_batches) in [(120, MAX_BATCHES), (1 << 20, 2)] {
            let mut step = 0;
            loop {
                let dir = temp_dir(&format!("split-{step}-{max_batches}"));
                let cache =
                    FileCache::open_with_max_batches(&dir, max_bytes, max_batches).expect("open");
                cache.push(&parent, &[b'p'; 120]).expect("push");
                let calls = Arc::new(AtomicUsize::new(0));
                let (counter, crash_at) = (calls.clone(), step);
                cache.inject(move |_, _| {
                    if counter.fetch_add(1, Ordering::SeqCst) == crash_at {
                        panic!("killed");
                    }
                    Ok(())
                });
                let crashed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    let _ = cache.replace(&parent, &halves);
                }))
                .is_err();
                drop(cache);
                let cache =
                    FileCache::open_with_max_batches(&dir, max_bytes, max_batches).expect("reopen");
                assert!(cache.take_evicted().is_empty(), "step {step}: nothing evicted");
                let pending = cache.pending();
                let whole = pending == [parent.clone()]
                    && cache.read(&parent).expect("parent").len() == 120;
                let split = pending == [a.clone(), b.clone()];
                assert!(whole || split, "step {step}: {pending:?}");
                let _ = fs::remove_dir_all(&dir);
                if !crashed {
                    assert!(split, "a replace that ran to the end split the batch");
                    break;
                }
                step += 1;
            }
            assert!(
                step >= 10,
                "every write, sync, rename, delete and publish was crashed at: {step}"
            );
        }
    }

    /// A split that fails after its commit point (a failed publish, a failed
    /// directory sync) has succeeded — its halves are safe in the journal and published by the
    /// next listing — and an opt-out deletes the journal too, leaving the directory empty.
    #[test]
    fn a_split_failing_after_its_commit_point_is_kept_and_purgeable() {
        use std::sync::atomic::{AtomicBool, Ordering};
        let (parent, a, b) = (id(1), format!("{}a", id(1)), format!("{}b", id(1)));
        let halves = [(a.clone(), b"half a".to_vec()), (b.clone(), b"half b".to_vec())];
        for failing in [Step::Publish, Step::SyncDir] {
            let dir = temp_dir(&format!("post-commit-{failing:?}"));
            let cache = FileCache::open(&dir, 1 << 20).expect("open");
            cache.push(&parent, b"parent").expect("push");
            let seen = Arc::new(AtomicBool::new(false));
            cache.inject(move |step, _| {
                if step == Step::Delete {
                    seen.store(true, Ordering::SeqCst);
                }
                if step == failing && seen.load(Ordering::SeqCst) {
                    return Err(io::Error::other("injected"));
                }
                Ok(())
            });
            cache.replace(&parent, &halves).expect("committed: not an error");
            assert!(!cache.pending().contains(&parent), "{failing:?}: the parent is gone");
            cache.inject(|_, _| Ok(())); // the storage recovers; then the app opts out
            assert!(cache.clear().is_ok());
            assert_eq!(
                fs::read_dir(&dir).expect("dir").count(),
                0,
                "{failing:?}: nothing survives the opt-out"
            );

            // Without the opt-out, the next listing publishes the journaled halves.
            let dir2 = temp_dir(&format!("post-commit-publish-{failing:?}"));
            let cache = FileCache::open(&dir2, 1 << 20).expect("open");
            cache.push(&parent, b"parent").expect("push");
            let flag = Arc::new(AtomicBool::new(true));
            let fail = flag.clone();
            let seen = Arc::new(AtomicBool::new(false));
            cache.inject(move |step, _| {
                if step == Step::Delete {
                    seen.store(true, Ordering::SeqCst);
                }
                if step == failing && seen.load(Ordering::SeqCst) && fail.load(Ordering::SeqCst) {
                    return Err(io::Error::other("injected"));
                }
                Ok(())
            });
            cache.replace(&parent, &halves).expect("committed");
            flag.store(false, Ordering::SeqCst);
            cache.push(&id(9), b"next").expect("push");
            assert_eq!(cache.pending(), [a.clone(), b.clone(), id(9)], "{failing:?}: published");
            let _ = fs::remove_dir_all(&dir);
            let _ = fs::remove_dir_all(&dir2);
        }
    }

    /// A reconfigure opens a second cache on the directory while the replaced
    /// pipeline still drains. Its open, pushes and listings must never see that pipeline's split
    /// half-done — deleting the journal of a split about to commit would lose both halves.
    #[test]
    fn a_second_cache_on_the_directory_never_breaks_a_split_in_progress() {
        let dir = temp_dir("two-generations");
        let (parent, a, b) = (id(1), format!("{}a", id(1)), format!("{}b", id(1)));
        let halves = [(a.clone(), b"half a".to_vec()), (b.clone(), b"half b".to_vec())];
        let old = FileCache::open(&dir, 1 << 20).expect("open");
        old.push(&parent, b"parent").expect("push");
        let other = Arc::new(Mutex::new(None));
        let (handle, path) = (other.clone(), dir.clone());
        let (waiting_tx, waiting_rx) = std::sync::mpsc::sync_channel(1);
        let waiting_rx = Mutex::new(waiting_rx);
        old.inject(move |step, _| {
            // Journaled, parent still there, commit next: the new generation starts now.
            let mut handle = handle.lock().expect("lock");
            if step == Step::Delete && handle.is_none() {
                let path = path.clone();
                let waiting_tx = waiting_tx.clone();
                *handle = Some(std::thread::spawn(move || {
                    COMMIT_WAIT.with(|slot| {
                        *slot.borrow_mut() = Some(Box::new(move || {
                            let _ = waiting_tx.try_send(());
                        }))
                    });
                    let new = FileCache::open(&path, 1 << 20).expect("open");
                    new.push(&id(2), b"new generation").expect("push");
                }));
                // The opener actually reached the held commit barrier mid-split.
                waiting_rx
                    .lock()
                    .unwrap()
                    .recv_timeout(std::time::Duration::from_secs(5))
                    .map_err(io::Error::other)?;
            }
            Ok(())
        });
        old.replace(&parent, &halves).expect("split");
        let new = other.lock().expect("lock").take().expect("the new generation ran");
        new.join().expect("the new generation succeeded");
        let reopened = FileCache::open(&dir, 1 << 20).expect("reopen");
        assert_eq!(reopened.pending(), [a, b, id(2)], "every record exactly once");
        let _ = fs::remove_dir_all(&dir);
    }

    /// A journal that cannot be published is not retried (and logged) by every
    /// listing — only again after the cache has written something.
    #[cfg(unix)]
    #[test]
    fn a_stuck_journal_is_retried_after_a_write_not_every_listing() {
        let dir = temp_dir("stuck-journal");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        let (parent, half) = (id(1), format!("{}a", id(1)));
        fs::write(cache.pend(&parent, &half), b"half").expect("journal");
        // The half's name is taken by a non-empty directory: the publishing rename fails.
        fs::create_dir(cache.path(&half, BATCH_EXT)).expect("blocker");
        fs::write(cache.path(&half, BATCH_EXT).join("x"), b"").expect("blocker content");
        let before = cache.recoveries.load(Ordering::Relaxed);
        for _ in 0..5 {
            cache.pending();
        }
        assert_eq!(cache.recoveries.load(Ordering::Relaxed), before + 1, "tried once");
        cache.push(&id(2), b"next").expect("push");
        assert_eq!(cache.recoveries.load(Ordering::Relaxed), before + 2, "a write retries");
        fs::remove_dir_all(cache.path(&half, BATCH_EXT)).expect("unblock");
        cache.push(&id(3), b"next").expect("push");
        assert_eq!(cache.pending(), [half, id(2), id(3)], "published once unblocked");
        let _ = fs::remove_dir_all(&dir);
    }

    /// A write killed half-way (bytes on disk, no rename) leaves no partial batch
    /// and costs no committed one.
    #[test]
    fn a_write_killed_half_way_leaves_no_partial_batch() {
        let dir = temp_dir("torn");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        cache.push(&id(1), b"committed").expect("push");
        cache.inject(|step, path| {
            if step == Step::Write {
                fs::write(path, b"half of a batc").expect("partial");
                panic!("killed mid-write");
            }
            Ok(())
        });
        let killed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _ = cache.push(&id(2), b"half of a batch that never lands");
        }));
        assert!(killed.is_err());
        drop(cache);
        let cache = FileCache::open(&dir, 1 << 20).expect("reopen");
        assert_eq!(cache.pending(), [id(1)]);
        assert_eq!(cache.read(&id(1)).expect("read"), b"committed");
        assert_eq!(fs::read_dir(&dir).expect("dir").count(), 1, "the torn .tmp is gone");
        let _ = fs::remove_dir_all(&dir);
    }

    /// A push whose file or directory cannot be made durable is not committed —
    /// it errs and leaves nothing behind — and a split that cannot be made durable before its
    /// commit point keeps the parent.
    #[test]
    fn durability_failures_are_errors_not_silent_successes() {
        for failing in [Step::Sync, Step::SyncDir] {
            let dir = temp_dir("sync");
            let cache = FileCache::open(&dir, 1 << 20).expect("open");
            cache.push(&id(1), b"parent").expect("push");
            cache.inject(move |step, _| {
                if step == failing {
                    return Err(io::Error::other("fsync failed"));
                }
                Ok(())
            });
            assert!(cache.push(&id(2), b"new").is_err(), "{failing:?}: not committed");
            assert_eq!(cache.pending(), [id(1)], "{failing:?}: nothing left of it");
            let halves =
                [(format!("{}a", id(1)), b"a".to_vec()), (format!("{}b", id(1)), b"b".to_vec())];
            assert!(cache.replace(&id(1), &halves).is_err());
            assert_eq!(cache.pending(), [id(1)], "{failing:?}: the parent stays whole");
            let _ = fs::remove_dir_all(&dir);
        }
    }

    /// An unreadable directory is an error for `clear`, not an empty success.
    #[cfg(unix)]
    #[test]
    fn clearing_an_unreadable_directory_fails() {
        use std::os::unix::fs::PermissionsExt;
        let dir = temp_dir("unreadable");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        cache.push(&id(1), b"batch").expect("push");
        fs::set_permissions(&dir, fs::Permissions::from_mode(0o300)).expect("chmod");
        assert!(cache.clear().is_err(), "cannot list, cannot claim it is empty");
        fs::set_permissions(&dir, fs::Permissions::from_mode(0o755)).expect("chmod");
        assert_eq!(cache.pending(), [id(1)]);
        let _ = fs::remove_dir_all(&dir);
    }

    /// The retention floor holds even when the older batches cannot be deleted: the newest is
    /// never a candidate.
    #[test]
    fn the_newest_batch_stays_when_older_deletions_fail() {
        let dir = temp_dir("floor-undeletable");
        let cache = FileCache::open(&dir, 25).expect("open");
        cache.push(&id(1), &[b'a'; 10]).expect("push");
        cache.push(&id(2), &[b'b'; 10]).expect("push");
        let newest = cache.path(&id(3), BATCH_EXT);
        cache.inject(move |step, path| {
            if step == Step::Delete && path != newest {
                return Err(io::Error::other("undeletable"));
            }
            Ok(())
        });
        let evicted = cache.push(&id(3), &[b'c'; 10]).expect("push");
        assert_eq!(evicted, Vec::<String>::new(), "nothing older could go, the newest stays");
        assert_eq!(cache.pending(), [id(1), id(2), id(3)]);
        let _ = fs::remove_dir_all(&dir);
    }

    /// A batch another pruning pass already deleted (a second cache on the directory, or an
    /// overlapping push) is not this pass's eviction: the exporter would count it twice.
    #[test]
    fn an_eviction_another_pass_already_made_is_not_counted_twice() {
        let dir = temp_dir("evict-twice");
        let cache = FileCache::open(&dir, 10).expect("open");
        cache.push(&id(1), &[b'a'; 10]).expect("push");
        let gone = cache.path(&id(1), BATCH_EXT);
        cache.inject(move |step, path| {
            if step == Step::Delete && path == gone {
                // The other pass wins the unlink.
                let _ = fs::remove_file(path);
            }
            Ok(())
        });
        let evicted = cache.push(&id(2), &[b'b'; 10]).expect("push");
        assert_eq!(evicted, Vec::<String>::new(), "the other pass counts it");
        assert_eq!(cache.pending(), [id(2)]);
        let _ = fs::remove_dir_all(&dir);
    }

    /// The compound case: another pass unlinked the batch, and then this pass's directory sync
    /// fails. The eviction is still the other pass's, not this one's, whatever the sync said.
    /// Control: this pass's own unlink followed by a failed sync still counts
    /// (`an_eviction_whose_directory_sync_fails_still_counts`).
    #[test]
    fn an_eviction_another_pass_made_is_not_counted_even_when_the_sync_fails() {
        use std::sync::atomic::{AtomicBool, Ordering};
        let dir = temp_dir("evict-twice-sync");
        let cache = FileCache::open(&dir, 10).expect("open");
        cache.push(&id(1), &[b'a'; 10]).expect("push");
        let gone = cache.path(&id(1), BATCH_EXT);
        let taken = Arc::new(AtomicBool::new(false));
        cache.inject({
            let taken = taken.clone();
            move |step, path| {
                if step == Step::Delete && path == gone {
                    // The other pass wins the unlink.
                    let _ = fs::remove_file(path);
                    taken.store(true, Ordering::SeqCst);
                }
                if step == Step::SyncDir && taken.load(Ordering::SeqCst) {
                    return Err(io::Error::other("injected"));
                }
                Ok(())
            }
        });
        let evicted = cache.push(&id(2), &[b'b'; 10]).expect("push");
        assert_eq!(evicted, Vec::<String>::new(), "the other pass counts it");
        assert_eq!(cache.pending(), [id(2)]);
        let _ = fs::remove_dir_all(&dir);
    }

    /// Creating the cache directory syncs its parent: the directory named by an absolute path,
    /// and the current directory for a bare relative name. An existing directory is not
    /// created, so nothing is synced for it.
    #[test]
    fn creating_the_directory_syncs_its_parent() {
        let dir = temp_dir("parent-sync");
        SYNCED_PARENT.with(|synced| synced.take());
        FileCache::open(&dir, 1 << 20).expect("open");
        let parent = dir.parent().map(Path::to_path_buf);
        assert_eq!(SYNCED_PARENT.with(|synced| synced.take()), parent, "an absolute path");
        FileCache::open(&dir, 1 << 20).expect("open again");
        assert_eq!(SYNCED_PARENT.with(|synced| synced.take()), None, "it existed");
        let _ = fs::remove_dir_all(&dir);
        let relative =
            PathBuf::from(format!("lk-telemetry-parent-sync-{}", crate::event::now_unix_nanos()));
        FileCache::open(&relative, 1 << 20).expect("open");
        let _ = fs::remove_dir_all(&relative);
        assert_eq!(
            SYNCED_PARENT.with(|synced| synced.take()),
            Some(PathBuf::from(".")),
            "a bare relative name lives in the current directory"
        );
    }

    /// Purge recovery: the first write finds the directory gone, and by the time it recreates
    /// it another generation's `open` already did. The recreated directory is used.
    #[test]
    fn purge_recovery_tolerates_a_directory_another_cache_recreated() {
        use std::sync::atomic::{AtomicBool, Ordering};
        let dir = temp_dir("purge-race");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        let first = AtomicBool::new(true);
        cache.inject(move |step, _| {
            if step == Step::Write && first.swap(false, Ordering::SeqCst) {
                return Err(io::ErrorKind::NotFound.into());
            }
            Ok(())
        });
        cache.push(&id(1), b"x").expect("the recreated directory is used");
        assert_eq!(cache.pending(), [id(1)]);
        let _ = fs::remove_dir_all(&dir);
    }

    /// A lowered file-count bound is in force while the cache opens, so opening-time pruning
    /// (and the crash test's two-file arm) really runs under count pressure.
    #[test]
    fn a_lowered_file_count_bound_applies_when_the_cache_opens() {
        let dir = temp_dir("open-bound");
        let cache = FileCache::open(&dir, 1 << 20).expect("open");
        for n in 1..=3 {
            cache.push(&id(n), b"x").expect("push");
        }
        drop(cache);
        let cache = FileCache::open_with_max_batches(&dir, 1 << 20, 2).expect("reopen");
        assert_eq!(cache.take_evicted(), [id(1)], "opening pruned to the requested bound");
        assert_eq!(cache.pending(), [id(2), id(3)]);
        let _ = fs::remove_dir_all(&dir);
    }
}