haematite 0.7.0

//! The pending directory-entry mutation journal (FENCE-CHAIN r8 R1a).
//!
//! This is `cfg(test)` machinery: it exists so the crash battery can simulate the
//! one failure a real filesystem will not produce on demand — a directory entry
//! that was renamed/created but whose covering fence never reached the platter, so
//! power loss rolls the entry back while the durable claim above it survives.
//!
//! ## Thread scope
//!
//! The journal and its FAILPOINT/LOSSY/OBSERVER hooks are THREAD-LOCAL: fence
//! sites are instrumented in every test build, so a process-global journal would
//! let one test's armed hooks contaminate another test running concurrently. Each
//! test thread carries its own journal, so tests are isolated with no suite-wide
//! serialisation. Every fence a pin needs teeth on (the `data_dir` fence, the
//! store-root fence via a direct `sync_dirty_dirs`, the metadata/refs entry
//! fences) runs on the test thread that armed the hooks.
//!
//! ## What it records
//!
//! Every instrumented directory-entry mutation registers `(parent, child,
//! generation, kind)`. `generation` is a per-`(parent, child)` counter so stacked
//! replacements order deterministically; `kind` is [`EntryKind`]. A
//! [`EntryKind::ReplaceInstall`] captures a restorable preimage — the replaced
//! file's bytes, copied at registration.
//!
//! ## The intent/commit protocol
//!
//! [`reserve_create_file`] / [`reserve_create_dir`] / [`reserve_install`] capture
//! the preimage and RESERVE a generation without yet publishing a pending entry.
//! The caller then executes the rename/mkdir and calls [`Reservation::commit`] on
//! success or [`Reservation::cancel`] on failure, so a failed mutation never leaves
//! a false pending entry.
//!
//! ## Clearing and the CUT
//!
//! A real fence ([`super::sync::sync_dir_entry`]) snapshots the pending generations
//! for its parent at its start and, on success, clears exactly those — a mutation
//! registered mid-sync stays pending. The LOSSY hook suppresses the real fsync so
//! nothing clears. [`cut`] then rewinds every still-pending entry per `(parent,
//! child)` in DESCENDING generation order: [`EntryKind::CreateFile`] /
//! [`EntryKind::CreateDir`] remove the entry (idempotent if already absent),
//! [`EntryKind::ReplaceInstall`] restores the preimage — so stacked lossy
//! replacements `A → B → C` rewind to `A`, not `B`.
//!
//! Fixed code clears everything before READY, so the CUT touches nothing and cold
//! recovery is exact; a patched-away fence leaves entries pending, so the CUT
//! rewinds them and the pin fails.

use std::cell::RefCell;
use std::collections::BTreeMap;
use std::ffi::OsString;
use std::io;
use std::path::{Path, PathBuf};

use super::sync::normalised_parent;

/// What a pending directory-entry mutation reverts to under a CUT.
#[derive(Clone, Debug)]
enum EntryKind {
    /// A newly created regular file; the CUT removes it.
    CreateFile,
    /// A newly created directory; the CUT removes it (recursively — a rewound
    /// store root drops the prefix subdirs and node files created beneath it).
    CreateDir,
    /// A clobbering replace of an existing file; the CUT restores these bytes.
    ReplaceInstall(Vec<u8>),
}

/// One registered-and-committed directory-entry mutation awaiting a fence.
#[derive(Clone, Debug)]
struct PendingEntry {
    child: OsString,
    generation: u64,
    kind: EntryKind,
}

/// One test thread's journal.
#[derive(Default)]
struct Journal {
    /// Committed pending entries, keyed by normalised parent.
    pending: BTreeMap<PathBuf, Vec<PendingEntry>>,
    /// The next generation to hand out per `(normalised parent, child)`.
    generations: BTreeMap<(PathBuf, OsString), u64>,
    /// OBSERVER: real directory-entry syncs counted by normalised parent.
    sync_counts: BTreeMap<PathBuf, usize>,
    /// LOSSY: suppress real fsyncs so nothing clears.
    lossy: bool,
    /// FAILPOINT: the next [`super::sync::sync_dir_entry`] returns an error.
    fail_next: bool,
}

thread_local! {
    static JOURNAL: RefCell<Journal> = RefCell::new(Journal::default());
}

fn with_journal<R>(function: impl FnOnce(&mut Journal) -> R) -> R {
    JOURNAL.with(|cell| function(&mut cell.borrow_mut()))
}

/// A reserved-but-not-yet-published mutation. Dropping one without
/// [`Self::commit`] or [`Self::cancel`] is a test bug and panics, so a missed
/// intent/commit pair cannot silently skip journalling.
#[must_use = "a fence reservation must be committed or cancelled"]
pub struct Reservation {
    parent: PathBuf,
    child: OsString,
    generation: u64,
    kind: EntryKind,
    resolved: bool,
}

impl Reservation {
    /// The mutation landed: publish the pending entry.
    pub fn commit(mut self) {
        let entry = PendingEntry {
            child: std::mem::take(&mut self.child),
            generation: self.generation,
            kind: std::mem::replace(&mut self.kind, EntryKind::CreateFile),
        };
        let parent = self.parent.clone();
        with_journal(|journal| journal.pending.entry(parent).or_default().push(entry));
        self.resolved = true;
    }

    /// The mutation failed: publish nothing.
    pub fn cancel(mut self) {
        self.resolved = true;
    }
}

impl Drop for Reservation {
    fn drop(&mut self) {
        assert!(
            self.resolved,
            "fence reservation for {}/{} dropped without commit or cancel",
            self.parent.display(),
            Path::new(&self.child).display(),
        );
    }
}

fn reserve(parent: &Path, child: OsString, kind: EntryKind) -> Reservation {
    let parent = normalised_parent(parent);
    let generation = with_journal(|journal| {
        let slot = journal
            .generations
            .entry((parent.clone(), child.clone()))
            .or_insert(0);
        *slot = slot.saturating_add(1);
        *slot
    });
    Reservation {
        parent,
        child,
        generation,
        kind,
        resolved: false,
    }
}

fn child_of(target: &Path) -> OsString {
    target
        .file_name()
        .map_or_else(|| OsString::from(target), OsString::from)
}

/// Reserve the creation of a new regular file at `target`.
pub fn reserve_create_file(target: &Path) -> Reservation {
    let parent = target.parent().unwrap_or_else(|| Path::new("."));
    reserve(parent, child_of(target), EntryKind::CreateFile)
}

/// Reserve the creation of a new directory at `target`.
pub fn reserve_create_dir(target: &Path) -> Reservation {
    let parent = target.parent().unwrap_or_else(|| Path::new("."));
    reserve(parent, child_of(target), EntryKind::CreateDir)
}

/// Reserve a clobbering install at `target`, auto-detecting whether it replaces
/// an existing file (capture the preimage) or creates a new one.
pub fn reserve_install(target: &Path) -> Reservation {
    let parent = target.parent().unwrap_or_else(|| Path::new("."));
    // An absent target is a create; a readable one is a replace whose current
    // bytes become the restorable preimage. A non-`NotFound` read error is a
    // test-environment fault on a path that is in practice always either absent
    // or readable; record an empty preimage rather than panicking inside a
    // durability primitive.
    let kind = match std::fs::read(target) {
        Ok(bytes) => EntryKind::ReplaceInstall(bytes),
        Err(error) if error.kind() == io::ErrorKind::NotFound => EntryKind::CreateFile,
        Err(_other) => EntryKind::ReplaceInstall(Vec::new()),
    };
    reserve(parent, child_of(target), kind)
}

// ---- hooks used by `super::sync::sync_dir_entry` -------------------------------

/// OBSERVER: record one real directory-entry sync of `parent`.
pub(super) fn record_observed_sync(parent: &Path) {
    with_journal(|journal| *journal.sync_counts.entry(parent.to_path_buf()).or_insert(0) += 1);
}

/// FAILPOINT: consume the one-shot arm, returning whether this sync must fail.
pub(super) fn take_failpoint() -> bool {
    with_journal(|journal| std::mem::replace(&mut journal.fail_next, false))
}

/// LOSSY: whether real fsyncs are currently suppressed.
pub(super) fn is_lossy() -> bool {
    with_journal(|journal| journal.lossy)
}

/// Snapshot the pending generations for `parent` at a fence's start.
pub(super) fn snapshot_pending_generations(parent: &Path) -> Vec<u64> {
    with_journal(|journal| {
        journal
            .pending
            .get(parent)
            .map(|entries| entries.iter().map(|entry| entry.generation).collect())
            .unwrap_or_default()
    })
}

/// Clear exactly the captured generations for `parent` after a successful fence;
/// a mutation registered mid-sync (a generation absent from `snapshot`) stays.
pub(super) fn clear_pending_generations(parent: &Path, snapshot: &[u64]) {
    with_journal(|journal| {
        if let Some(entries) = journal.pending.get_mut(parent) {
            entries.retain(|entry| !snapshot.contains(&entry.generation));
            if entries.is_empty() {
                journal.pending.remove(parent);
            }
        }
    });
}

// ---- controls used by the crash battery ---------------------------------------

/// Arm the one-shot FAILPOINT: the next [`super::sync::sync_dir_entry`] fails.
pub fn arm_failpoint() {
    with_journal(|journal| journal.fail_next = true);
}

/// Enable or disable LOSSY fsync suppression.
pub fn set_lossy(lossy: bool) {
    with_journal(|journal| journal.lossy = lossy);
}

/// Number of real directory-entry syncs observed against `parent` since the last
/// [`reset_observer`].
pub fn sync_count(parent: &Path) -> usize {
    let parent = normalised_parent(parent);
    with_journal(|journal| journal.sync_counts.get(&parent).copied().unwrap_or(0))
}

/// Total real directory-entry syncs observed since the last [`reset_observer`].
pub fn total_sync_count() -> usize {
    with_journal(|journal| journal.sync_counts.values().copied().sum())
}

/// Reset only the OBSERVER counters (leave pending state and hooks intact).
pub fn reset_observer() {
    with_journal(|journal| journal.sync_counts.clear());
}

/// Number of still-pending (unfenced) directory-entry mutations — the journal's
/// fenced/unfenced polarity as a runnable assertion.
pub fn pending_len() -> usize {
    with_journal(|journal| journal.pending.values().map(Vec::len).sum())
}

/// Clear all journal state for THIS thread: pending entries, generations, observer
/// counts, hooks.
pub fn reset() {
    with_journal(|journal| *journal = Journal::default());
}

/// Reset this thread's journal to a known-empty world, on both entry and drop, so
/// a fence test starts and ends clean regardless of thread reuse.
pub struct TestGuard;

impl Drop for TestGuard {
    fn drop(&mut self) {
        reset();
    }
}

/// Begin a fence-chain test: reset this thread's journal and return a guard that
/// resets it again on drop. No process-wide lock is needed — the journal is
/// thread-local, so concurrent tests never share it.
pub fn test_guard() -> TestGuard {
    reset();
    TestGuard
}

/// Rewind every still-pending directory-entry mutation, simulating the power-loss
/// rollback of entries whose fence never reached the platter. Per `(parent,
/// child)`, entries apply in DESCENDING generation order so a stack of lossy
/// replacements rewinds to the earliest durable state.
pub fn cut() -> io::Result<()> {
    let pending = with_journal(|journal| std::mem::take(&mut journal.pending));
    for (parent, mut entries) in pending {
        // Highest generation first: the most recent replace restores its
        // predecessor's bytes, then the next restores the state before it, and a
        // create at the bottom of the stack removes the entry entirely.
        entries.sort_by(|left, right| right.generation.cmp(&left.generation));
        for entry in entries {
            let target = parent.join(&entry.child);
            match entry.kind {
                EntryKind::CreateFile => remove_file_idempotent(&target)?,
                EntryKind::CreateDir => remove_dir_idempotent(&target)?,
                EntryKind::ReplaceInstall(bytes) => std::fs::write(&target, &bytes)?,
            }
        }
    }
    Ok(())
}

fn remove_file_idempotent(target: &Path) -> io::Result<()> {
    match std::fs::remove_file(target) {
        Ok(()) => Ok(()),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error),
    }
}

fn remove_dir_idempotent(target: &Path) -> io::Result<()> {
    match std::fs::remove_dir_all(target) {
        Ok(()) => Ok(()),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error),
    }
}