fs-transaction 0.2.0

//! The write-ahead journal — what makes a whole [`ChangeSet`](crate::ChangeSet)
//! crash-atomic, not just each file in it.
//!
//! [`crate::change`] already lands every file write atomically (via
//! [`Storage::write_atomic`]) and unwinds the set in memory on any *error*. The
//! one failure that leaves behind — a `kill -9` or a power cut *between* two of a
//! set's writes — is what this closes. The mechanism is the classic write-ahead
//! log, specialized to the one shape a change set takes: a sequence of
//! whole-file writes, copies, renames, removes, mode flips, and links, each
//! self-contained.
//!
//! ## The protocol
//!
//! Before touching a single file, [`ChangeSet::apply`](crate::ChangeSet::apply)
//! writes this journal — the complete list of intended ops — and flushes it. That
//! flush is the **commit point**. Because the journal is itself written through
//! [`Storage::write_atomic`], it appears whole or not at all, so a crash leaves
//! the disk in exactly one of two states:
//!
//! - **No journal** (the crash beat the commit point). No file write had
//!   started yet either, so the tree is untouched — nothing to recover.
//! - **A whole journal** (the crash came after the commit point). Some, all, or
//!   none of the file writes may have landed. [`recover`] replays the journal
//!   forward — idempotently, so already-applied ops are no-ops — bringing the
//!   tree to the fully-applied state, then deletes the journal.
//!
//! So an interrupted change set always resolves to a *consistent* tree:
//! either fully before it (the commit point was never reached) or fully after it
//! (recovery rolled it forward). The one honesty worth stating plainly:
//!
//! > Which of the two an interruption yields depends on whether the process
//! > kept control. An **error** returned mid-apply is unwound in memory — the
//! > tree ends up fully *before*. A **crash** loses that chance, so recovery
//! > rolls the journaled set fully *forward* instead. Both endpoints are
//! > consistent; they are simply different consistent states, and this crate
//! > does not pretend a lost-power change didn't happen when its intent was
//! > already durably on disk.
//!
//! ## Format
//!
//! A compact, length-prefixed binary encoding with a magic header and a trailing
//! checksum. The journal is ephemeral machine state, not something the user
//! owns, so it is not meant to be read by hand — and binary keeps opaque
//! payloads (an image staged for a write) exact without escaping.
//! The checksum is belt-and-suspenders: `write_atomic` already makes the journal
//! all-or-nothing, so a torn *write* is impossible, but bit-rot on the way back
//! is not, and a journal that cannot be trusted must be refused loudly rather
//! than replayed into corruption.
//!
//! ## Payload by reference
//!
//! One op — [`FileOp::CopyFrom`] — journals a *source path* in place of the bytes
//! it will write. Without it, a change set putting a whole captured tree back
//! would duplicate that entire tree into the journal at the commit point,
//! making a restore two full-tree writes and bounding it by the total size of
//! the tree rather than the number of files in it.
//!
//! Journaling a reference stays deterministic to replay, but only because the
//! referent is *required* to be immutable. A content-addressed blob
//! satisfies that by construction — its path is the digest of its own contents —
//! so replay either finds exactly the bytes the set intended, or finds nothing
//! and fails loudly. That requirement is a real obligation on whoever stages the
//! op: pointed at a mutable file, it would let recovery write bytes the
//! original change never intended, which is the one thing a write-ahead log
//! exists to prevent.

use std::borrow::Cow;
use std::path::{Component, Path, PathBuf};

use crate::change::FileOp;
use crate::error::{Error, Result};
use crate::fs::Storage;

/// Where a change set's write-ahead journal lives — and, because they must
/// agree about it, both halves of the protocol that depends on the answer.
///
/// The name is a single transient dotfile, by default in the root: it exists
/// only between a change set's commit point and its completion, so in steady
/// state the tree carries no journal at all, and no dotfolder is spawned to
/// hold one. It survives a crash solely so [`Journal::recover`] can find it,
/// and is removed the moment recovery (or a clean apply) finishes.
///
/// ## Why this is a type and not a parameter
///
/// [`apply`](Journal::apply) and [`recover`](Journal::recover) have to name the
/// same file. If they disagree, nothing fails loudly — recovery simply looks
/// where no journal is and reports [`Recovered::Nothing`], leaving an
/// interrupted change half-applied forever. A caller that holds one `Journal`
/// and uses it for both cannot make that mistake, which is why the two
/// operations live on the value rather than taking the name separately.
///
/// [`ChangeSet::apply`](crate::ChangeSet::apply) and the free
/// [`recover`] are shorthands for `Journal::default()`; reach for a named one
/// when the default would collide with something the tree already means, or
/// when an existing deployment already writes a journal under its own name.
///
/// ```
/// # use fs_transaction::journal::Journal;
/// let journal = Journal::named(".myapp-journal")?;
/// assert_eq!(journal.name(), ".myapp-journal");
/// // Not a single path component — refused rather than escaping the root.
/// assert!(Journal::named("../elsewhere").is_err());
/// # Ok::<(), fs_transaction::Error>(())
/// ```
///
/// ## A journal outside the tree
///
/// By default the journal lands in the root it applies to, which is right for
/// a tree only this machine writes. It is wrong for a tree something *syncs* —
/// an iCloud or Dropbox folder — because the journal is this process's crash
/// state, and a sync service cannot tell it from content: the file travels to
/// machines that never crashed, where a recovery would replay *another
/// machine's* intent against a tree that may have moved on, and an apply
/// would refuse a "stale" journal no local change left behind.
/// [`kept_in`](Journal::kept_in) is the fix: the journal lives in a directory
/// the caller owns and nothing syncs (an application-support or cache
/// directory), and the tree itself never holds a journal at all, transiently
/// or otherwise.
///
/// Two obligations come with a homed journal, both the caller's. The home
/// must be **absolute** — a relative one would resolve against whatever the
/// process's current directory happens to be, and a journal written from one
/// directory and sought from another is exactly the stranding this type
/// exists to prevent; `kept_in` refuses anything else. And the pairing of
/// home and root is not recorded anywhere: the journal does not know which
/// tree it belongs to, so recovering it against a different root replays
/// intent against the wrong tree. One home directory, one root, one name —
/// a caller with several roots keeps several names.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Journal {
    name: Cow<'static, str>,
    /// Where the journal file lives, when not in the root itself. Always an
    /// absolute directory — [`Journal::kept_in`] refuses anything else.
    home: Option<PathBuf>,
}

impl Journal {
    /// The name used when none is given: a dotted, crate-namespaced file
    /// unlikely to collide with anything the tree itself means.
    pub const DEFAULT_NAME: &'static str = ".fstx-journal";

    /// A journal under `name`, which must be a single path component — not
    /// empty, not `.` or `..`, and containing no separator.
    ///
    /// The check is what keeps the name from being an escape hatch: it is
    /// joined onto a caller-supplied root, and a name like `../../elsewhere`
    /// would write outside the very tree
    /// [`ChangeSet::apply`](crate::ChangeSet::apply) clamps every staged op
    /// into.
    pub fn named(name: impl Into<Cow<'static, str>>) -> Result<Self> {
        let name = name.into();
        let mut components = Path::new(name.as_ref()).components();
        let single =
            matches!(components.next(), Some(Component::Normal(_))) && components.next().is_none();
        if !single {
            return Err(Error::InvalidJournalName(name.into_owned()));
        }
        Ok(Self { name, home: None })
    }

    /// This journal, kept in `home` instead of in the root it applies to.
    ///
    /// For trees something syncs — see the type docs for why the journal must
    /// not live where a sync service can carry it to another machine, and for
    /// the two obligations (an absolute home, and a stable home–root pairing)
    /// that come with taking this.
    ///
    /// `home` is a directory; the journal keeps its [`name`](Journal::name)
    /// inside it. A relative `home` is refused
    /// ([`Error::InvalidJournalHome`]): it would resolve against the process's
    /// current directory, which apply and a later recovery have no reason to
    /// share.
    ///
    /// ```
    /// # use fs_transaction::journal::Journal;
    /// let journal = Journal::named(".myapp-journal")?
    ///     .kept_in("/var/lib/myapp/journals")?;
    /// assert!(Journal::default().kept_in("not/absolute").is_err());
    /// # Ok::<(), fs_transaction::Error>(())
    /// ```
    pub fn kept_in(self, home: impl Into<PathBuf>) -> Result<Self> {
        let home = home.into();
        if !home.is_absolute() {
            return Err(Error::InvalidJournalHome(home));
        }
        Ok(Self {
            home: Some(home),
            ..self
        })
    }

    /// The journal's file name.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// The directory this journal is [kept in](Journal::kept_in), if it is
    /// not the root itself. `apply` uses this to make the home before writing
    /// into it — a root exists by the time anything applies against it, but a
    /// cache directory may not.
    pub(crate) fn home(&self) -> Option<&Path> {
        self.home.as_deref()
    }

    /// Where this journal lives when applying to `root`: its
    /// [home](Journal::kept_in) if it has one, otherwise the root itself.
    pub fn path_in(&self, root: &Path) -> PathBuf {
        match &self.home {
            Some(home) => home.join(self.name.as_ref()),
            None => root.join(self.name.as_ref()),
        }
    }

    /// Whether `path` names this journal, or the staging sibling
    /// [`Storage::write_atomic`] publishes it through.
    ///
    /// A containment test on the file name rather than an equality one, so
    /// that both the journal and its transient `write_atomic` temporary are
    /// recognized. A [homed](Journal::kept_in) journal additionally requires
    /// the path to sit in its home: the homed design's whole point is that no
    /// file in the root is this journal, so a same-named file there — synced
    /// in, or simply a coincidence — must not be claimed. Used by a
    /// fault-injecting backend to leave the journal's own writes alone and
    /// fail only the file writes it means to.
    pub fn owns_path(&self, path: &Path) -> bool {
        let name_matches = path
            .file_name()
            .and_then(|n| n.to_str())
            .is_some_and(|n| n.contains(self.name.as_ref()));
        match &self.home {
            Some(home) => name_matches && path.parent() == Some(home.as_path()),
            None => name_matches,
        }
    }
}

impl Default for Journal {
    fn default() -> Self {
        Self {
            name: Cow::Borrowed(Self::DEFAULT_NAME),
            home: None,
        }
    }
}

/// The magic prefix stamped on every journal, embedding a one-byte format
/// version (`1`). A file that does not start with this is not a journal this
/// crate wrote — or is one from an incompatible future version — and is refused
/// rather than guessed at.
const MAGIC: &[u8; 8] = b"FSTXJRN1";

/// The magic this crate stamped before it was lifted out of `prov`, where it
/// was named after a project older still. Accepted on read and never written.
///
/// The format is byte-identical, so this costs one comparison — and refusing it
/// would strand the one tree that can be carrying such a journal: a workspace
/// interrupted mid-apply by the crash that is the whole reason a journal
/// outlives its change. There is nothing to roll that forward but this.
const LEGACY_MAGIC: &[u8; 8] = b"COLOJRN1";

/// Serialize a change set's ops into journal bytes: `MAGIC`, the op count, each
/// op, then a checksum over everything preceding it.
pub fn encode(ops: &[FileOp]) -> Result<Vec<u8>> {
    let mut buf = Vec::with_capacity(64);
    buf.extend_from_slice(MAGIC);
    buf.extend_from_slice(&(ops.len() as u64).to_le_bytes());
    for op in ops {
        match op {
            FileOp::Write { path, bytes } => {
                buf.push(0);
                put_path(&mut buf, path)?;
                put_bytes(&mut buf, bytes);
            }
            FileOp::Rename { from, to } => {
                buf.push(1);
                put_path(&mut buf, from)?;
                put_path(&mut buf, to)?;
            }
            FileOp::Remove { path } => {
                buf.push(2);
                put_path(&mut buf, path)?;
            }
            // Two paths, no payload — the point of the op. See [`FileOp::CopyFrom`]
            // for why journaling a *reference* is still deterministic to replay.
            FileOp::CopyFrom { path, source } => {
                buf.push(3);
                put_path(&mut buf, path)?;
                put_path(&mut buf, source)?;
            }
            FileOp::SetExecutable { path, executable } => {
                buf.push(4);
                put_path(&mut buf, path)?;
                buf.push(u8::from(*executable));
            }
            // The target is encoded on `put_path`'s terms — UTF-8 or refused at
            // the commit point — even though it is a link's text rather than a
            // file of the tree: a journal must replay identically wherever it
            // is read, and a mangled target is an invented one.
            FileOp::SetLink { path, target } => {
                buf.push(5);
                put_path(&mut buf, path)?;
                put_path(&mut buf, target)?;
            }
        }
    }
    let checksum = fnv1a(&buf);
    buf.extend_from_slice(&checksum.to_le_bytes());
    Ok(buf)
}

/// Parse journal bytes back into ops, verifying the magic and the checksum. A
/// mismatch is an [`Error::Corrupt`] — a journal that cannot be trusted is
/// refused, never partially replayed.
pub fn decode(bytes: &[u8]) -> Result<Vec<FileOp>> {
    let corrupt = |what: &str| Error::Corrupt(what.to_string());

    let stamp = bytes.get(..MAGIC.len());
    if bytes.len() < MAGIC.len() + 8 + 8
        || !matches!(stamp, Some(m) if m == MAGIC || m == LEGACY_MAGIC)
    {
        return Err(corrupt("not a journal (bad header)"));
    }
    let body_end = bytes.len() - 8;
    let stored = u64::from_le_bytes(bytes[body_end..].try_into().unwrap());
    if fnv1a(&bytes[..body_end]) != stored {
        return Err(corrupt("checksum mismatch"));
    }

    let mut cur = Cursor {
        bytes: &bytes[..body_end],
        at: MAGIC.len(),
    };
    let count = cur.take_u64()?;
    // Each op costs at least its one-byte tag, so a count the body cannot
    // possibly hold is a lie about the record, not a large journal — and it
    // must be refused *before* it sizes an allocation, or a crafted header
    // aborts the process instead of erroring.
    if count > (cur.bytes.len() - cur.at) as u64 {
        return Err(corrupt("op count exceeds the journal body"));
    }
    let mut ops = Vec::with_capacity(count as usize);
    for _ in 0..count {
        let op = match cur.take_u8()? {
            0 => FileOp::Write {
                path: cur.take_path()?,
                bytes: cur.take_bytes()?.to_vec(),
            },
            1 => FileOp::Rename {
                from: cur.take_path()?,
                to: cur.take_path()?,
            },
            2 => FileOp::Remove {
                path: cur.take_path()?,
            },
            3 => FileOp::CopyFrom {
                path: cur.take_path()?,
                source: cur.take_path()?,
            },
            4 => FileOp::SetExecutable {
                path: cur.take_path()?,
                // Strictly 0 or 1: any other byte means this is not the
                // record it claims to be, and a journal that cannot be
                // trusted is refused, never guessed at.
                executable: match cur.take_u8()? {
                    0 => false,
                    1 => true,
                    other => {
                        return Err(corrupt(&format!("invalid executable flag {other}")));
                    }
                },
            },
            5 => FileOp::SetLink {
                path: cur.take_path()?,
                target: cur.take_path()?,
            },
            other => return Err(corrupt(&format!("unknown op tag {other}"))),
        };
        ops.push(op);
    }
    if cur.at != cur.bytes.len() {
        return Err(corrupt("trailing bytes after the last op"));
    }
    Ok(ops)
}

/// The outcome of a [`recover`] pass.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Recovered {
    /// No journal was present — steady state, the common case.
    Nothing,
    /// A journal was found and its `ops` ops were rolled forward, then it was
    /// removed. The tree was interrupted mid-change and is now consistent.
    Applied(usize),
}

impl Journal {
    /// Finish any change set a crash left journaled at `root`, rolling the tree
    /// forward to the fully-applied state, then remove the journal.
    ///
    /// The recovery entry point: run it before anything reads the tree, so an
    /// interrupted change heals first. A no-op when no journal is present, so
    /// it is cheap to call unconditionally. Replay is idempotent — a write
    /// already landed is simply rewritten, a rename already done is recognized
    /// and skipped — so recovering the *same* journal twice (a crash *during*
    /// recovery) is safe.
    ///
    /// Must name the same journal the interrupted [`apply`](Journal::apply)
    /// wrote. A different one finds nothing and reports
    /// [`Recovered::Nothing`], which is why both live on this value.
    pub async fn recover<FS: Storage>(&self, fs: &FS, root: &Path) -> Result<Recovered> {
        let journal = self.path_in(root);
        let bytes = match fs.read(&journal).await {
            Ok(bytes) => bytes,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Recovered::Nothing),
            Err(e) => return Err(e.into()),
        };
        let ops = decode(&bytes)?;
        // Clamped to the root on `apply`'s own terms, and with more reason: a
        // journal is bytes found on disk, not a set this process staged, and
        // the checksum authenticates nothing. An apply refuses an escaping
        // path before writing; a replay that did not would be the same escape
        // through the back door — refused instead, with the journal left in
        // place like any other journal that cannot be trusted.
        crate::change::guard_ops(&ops)?;
        let mut touched = std::collections::BTreeSet::new();
        for op in &ops {
            replay(fs, root, op, &mut touched).await?;
        }
        // Recovery makes the same promise a clean apply does: once the
        // journal is given up, the state it certified survives a power cut.
        // The replayed renames, removals, bits, and fresh directory chains
        // are flushed — barriers capped by one durable sync — before the
        // journal goes; the deletion itself is not flushed, because a
        // resurrected journal replays idempotently over ops already durable.
        crate::fs::flush_all_durable(fs, touched, root).await?;
        fs.remove_file(&journal).await?;
        Ok(Recovered::Applied(ops.len()))
    }
}

/// Recover the [default](Journal::DEFAULT_NAME) journal at `root` — shorthand
/// for [`Journal::default().recover(..)`](Journal::recover).
pub async fn recover<FS: Storage>(fs: &FS, root: &Path) -> Result<Recovered> {
    Journal::default().recover(fs, root).await
}

/// Re-apply one journaled op, tolerant of it having already landed before the
/// crash — this is what makes rolling a journal forward idempotent.
///
/// `touched` collects the same flush debt [`crate::change`]'s exec does — the
/// entries, bits, and fresh chains no per-op call flushes — for
/// [`Journal::recover`] to settle before the journal is given up.
async fn replay<FS: Storage>(
    fs: &FS,
    root: &Path,
    op: &FileOp,
    touched: &mut std::collections::BTreeSet<PathBuf>,
) -> Result<()> {
    match op {
        // Whole-file writes are idempotent by nature: writing the intended bytes
        // again reaches the same state whether or not the crash beat this op.
        // `replace`, on apply's own terms: the durability is the recovery's
        // one batched flush, not the file's.
        FileOp::Write { path, bytes } => {
            let full = root.join(path);
            ensure_parent(fs, &full, touched).await?;
            fs.replace(&full, bytes).await?;
            crate::change::settle_write_debt(fs, &full, touched).await?;
        }
        // Idempotent for the same reason a `Write` is — with the bytes fetched
        // from the source rather than carried in the journal. That is sound
        // exactly as far as the source is immutable ([`FileOp::CopyFrom`]): a
        // content-addressed blob either holds the intended bytes or is gone, and
        // gone is an error rather than a silent divergence, because replay must
        // never invent a state the original set did not intend.
        FileOp::CopyFrom { path, source } => {
            let (full, source_full) = (root.join(path), root.join(source));
            let bytes = fs.read(&source_full).await.map_err(|e| {
                Error::Recovery(format!(
                    "cannot copy {} from {} — {e}",
                    full.display(),
                    source_full.display()
                ))
            })?;
            ensure_parent(fs, &full, touched).await?;
            fs.replace(&full, &bytes).await?;
            crate::change::settle_write_debt(fs, &full, touched).await?;
        }
        // A remove of a file already gone is the state we wanted, not a failure.
        FileOp::Remove { path } => {
            let full = root.join(path);
            match fs.remove_file(&full).await {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(e.into()),
            }
            if let Some(dir) = crate::fs::parent_dir(&full) {
                touched.insert(dir.to_path_buf());
            }
        }
        // Setting a bit that is already set (or already cleared) reaches the
        // same state — idempotent by nature, like a whole-file write. On a
        // backend that models no bit the call no-ops, which is what the op
        // means there. The link guard is apply's, for apply's reason: mode
        // writes follow links, and a journal is bytes this process did not
        // author.
        FileOp::SetExecutable { path, executable } => {
            let full = root.join(path);
            crate::change::guard_not_link(fs, &full).await?;
            fs.set_executable(&full, *executable).await?;
            // The inode barriered while the name still resolves — a later op
            // in this same journal may rename or remove it — and the parent
            // batched, on exec's own terms.
            fs.sync(&full, crate::fs::Durability::Ordered).await?;
            if let Some(dir) = crate::fs::parent_dir(&full) {
                touched.insert(dir.to_path_buf());
            }
        }
        // `set_link` replaces whatever is at the path, so replaying it lands
        // the same link whether the crash beat the op, interrupted it midway
        // (a remove-then-remake backend caught between the two), or came
        // after it was done.
        FileOp::SetLink { path, target } => {
            let full = root.join(path);
            ensure_parent(fs, &full, touched).await?;
            fs.set_link(&full, target).await?;
            if let Some(dir) = crate::fs::parent_dir(&full) {
                touched.insert(dir.to_path_buf());
            }
        }
        // The one op that is not naturally idempotent: after it lands, the source
        // is gone and the destination present, so a blind re-rename would fail.
        // Recover by state — move it if the source is still there, accept it as
        // done if only the destination is, and refuse only if *neither* exists,
        // which no honest interruption of this set can produce.
        FileOp::Rename { from, to } => {
            let (from_full, to_full) = (root.join(from), root.join(to));
            if fs.try_exists(&from_full).await? {
                ensure_parent(fs, &to_full, touched).await?;
                fs.rename(&from_full, &to_full).await?;
            } else if fs.try_exists(&to_full).await? {
                // Already renamed before the crash — nothing to redo.
            } else {
                return Err(Error::Recovery(format!(
                    "neither {} nor {} exists — cannot complete the rename",
                    from_full.display(),
                    to_full.display()
                )));
            }
            // Both entries owe a flush whichever branch ran: even an
            // already-done rename was done by a crashed process that never
            // flushed it.
            for side in [&from_full, &to_full] {
                if let Some(dir) = crate::fs::parent_dir(side) {
                    touched.insert(dir.to_path_buf());
                }
            }
        }
    }
    Ok(())
}

async fn ensure_parent<FS: Storage>(
    fs: &FS,
    full: &Path,
    touched: &mut std::collections::BTreeSet<PathBuf>,
) -> Result<()> {
    if let Some(dir) = crate::fs::parent_dir(full) {
        for made in crate::fs::create_dir_all_traced(fs, dir).await? {
            touched.insert(made);
        }
    }
    Ok(())
}

// ---- encoding helpers ----

fn put_bytes(buf: &mut Vec<u8>, bytes: &[u8]) {
    buf.extend_from_slice(&(bytes.len() as u64).to_le_bytes());
    buf.extend_from_slice(bytes);
}

/// Encode a root-relative path as UTF-8, so a journal written on one platform
/// replays identically on another. A path that is not UTF-8 is refused at the
/// commit point rather than mangled into one that is.
fn put_path(buf: &mut Vec<u8>, path: &Path) -> Result<()> {
    let s = path
        .to_str()
        .ok_or_else(|| Error::NonUtf8Path(path.to_path_buf()))?;
    put_bytes(buf, s.as_bytes());
    Ok(())
}

/// A forward-only reader over the journal body, bounds-checking every take so a
/// truncated or malformed record surfaces as an error rather than a panic.
struct Cursor<'a> {
    bytes: &'a [u8],
    at: usize,
}

impl Cursor<'_> {
    fn short() -> Error {
        Error::Corrupt("unexpected end of data".into())
    }

    fn take(&mut self, n: usize) -> Result<&[u8]> {
        let end = self.at.checked_add(n).ok_or_else(Self::short)?;
        let slice = self.bytes.get(self.at..end).ok_or_else(Self::short)?;
        self.at = end;
        Ok(slice)
    }

    fn take_u8(&mut self) -> Result<u8> {
        Ok(self.take(1)?[0])
    }

    fn take_u64(&mut self) -> Result<u64> {
        Ok(u64::from_le_bytes(self.take(8)?.try_into().unwrap()))
    }

    fn take_bytes(&mut self) -> Result<&[u8]> {
        let len = self.take_u64()? as usize;
        self.take(len)
    }

    fn take_path(&mut self) -> Result<PathBuf> {
        let bytes = self.take_bytes()?;
        let s = std::str::from_utf8(bytes).map_err(|_| Error::Corrupt("non-UTF-8 path".into()))?;
        Ok(PathBuf::from(s))
    }
}

/// FNV-1a, 64-bit — a small, deterministic, dependency-free checksum. It guards
/// against bit-rot in a journal read back after a crash; it is not, and need not
/// be, cryptographic.
fn fnv1a(data: &[u8]) -> u64 {
    let mut hash = 0xcbf2_9ce4_8422_2325;
    for &byte in data {
        hash ^= u64::from(byte);
        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
    }
    hash
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::exec::block_on;
    use crate::fs::StdFs;

    fn tmp(name: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!("fstx-journal-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    fn read(root: &Path, rel: &str) -> Option<String> {
        std::fs::read_to_string(root.join(rel)).ok()
    }

    // ---- encoding ----

    #[test]
    fn a_change_set_round_trips_through_the_journal() {
        let ops = vec![
            FileOp::Write {
                path: "child.md".into(),
                bytes: b"hello".to_vec(),
            },
            FileOp::Rename {
                from: "a.md".into(),
                to: "sub/a.md".into(),
            },
            FileOp::Remove {
                path: "gone.md".into(),
            },
            FileOp::SetExecutable {
                path: "run.sh".into(),
                executable: true,
            },
            FileOp::SetLink {
                path: "link.md".into(),
                target: "../elsewhere.md".into(),
            },
        ];
        let bytes = encode(&ops).unwrap();
        assert_eq!(decode(&bytes).unwrap(), ops);
    }

    #[test]
    fn an_invalid_executable_flag_is_refused_not_guessed() {
        // The flag is strictly 0 or 1: any other byte means the record is not
        // what it claims, and a journal that cannot be trusted is refused.
        let ops = vec![FileOp::SetExecutable {
            path: "run.sh".into(),
            executable: true,
        }];
        let mut bytes = encode(&ops).unwrap();
        // The flag is the byte just before the trailing 8-byte checksum.
        let flag_at = bytes.len() - 8 - 1;
        assert_eq!(bytes[flag_at], 1);
        bytes[flag_at] = 7;
        // Re-stamp the checksum so only the flag is at fault.
        let body_end = bytes.len() - 8;
        let sum = fnv1a(&bytes[..body_end]);
        bytes[body_end..].copy_from_slice(&sum.to_le_bytes());
        let err = decode(&bytes).unwrap_err();
        assert!(err.to_string().contains("executable flag"), "{err}");
    }

    #[test]
    fn a_copy_journals_a_reference_not_the_payload() {
        // The point of the op, stated as an assertion: the journal for a copy is
        // bounded by the path lengths, not by the size of what it will write.
        // Without this, restoring a captured tree writes that whole tree
        // into the journal before touching a single file.
        let payload: Vec<u8> = vec![7; 512 * 1024];
        let by_value = encode(&[FileOp::Write {
            path: "notes/photo.jpg".into(),
            bytes: payload.clone(),
        }])
        .unwrap();
        let by_reference = encode(&[FileOp::CopyFrom {
            path: "notes/photo.jpg".into(),
            source: "history/blobs/9f/86d081".into(),
        }])
        .unwrap();
        assert!(by_value.len() > payload.len(), "a Write carries its bytes");
        assert!(
            by_reference.len() < 128,
            "a CopyFrom carries two paths: {} bytes",
            by_reference.len()
        );
        assert_eq!(decode(&by_reference).unwrap().len(), 1);
    }

    #[test]
    fn binary_payloads_survive_the_journal_verbatim() {
        // An attached photo staged for a write is opaque bytes, not text — the
        // journal must carry it exactly, with no escaping or UTF-8 assumption.
        let payload: Vec<u8> = (0u8..=255).cycle().take(1000).collect();
        let ops = vec![FileOp::Write {
            path: "photo.png".into(),
            bytes: payload.clone(),
        }];
        let decoded = decode(&encode(&ops).unwrap()).unwrap();
        assert_eq!(decoded, ops);
    }

    #[test]
    fn a_tampered_journal_is_refused_not_replayed() {
        // The checksum's whole job: a journal whose bytes changed under it must be
        // rejected loudly, never silently replayed into a corrupt tree.
        let ops = vec![FileOp::Write {
            path: "child.md".into(),
            bytes: b"hello".to_vec(),
        }];
        let mut bytes = encode(&ops).unwrap();
        let mid = bytes.len() / 2;
        bytes[mid] ^= 0xff;
        let err = decode(&bytes).unwrap_err();
        assert!(err.to_string().contains("corrupt"), "{err}");
    }

    #[test]
    fn a_non_journal_file_is_rejected() {
        assert!(decode(b"not a journal at all").is_err());
        assert!(decode(b"").is_err());
    }

    // ---- recovery: simulated crashes ----
    //
    // A unit test cannot pull the power, so it constructs the exact on-disk state
    // a crash at a given instant would leave — a whole journal plus some prefix of
    // its ops applied — and asserts recovery reaches the fully-applied state.

    #[test]
    fn recovery_completes_a_change_set_that_had_not_started() {
        // Crash right after the commit point: journal on disk, no op applied yet.
        let root = tmp("recover-none-applied");
        std::fs::write(root.join("parent.md"), "old parent").unwrap();
        let ops = vec![
            FileOp::Write {
                path: "child.md".into(),
                bytes: b"child".to_vec(),
            },
            FileOp::Write {
                path: "parent.md".into(),
                bytes: b"new parent".to_vec(),
            },
        ];
        std::fs::write(Journal::default().path_in(&root), encode(&ops).unwrap()).unwrap();

        let outcome = block_on(recover(&StdFs, &root)).unwrap();

        assert_eq!(outcome, Recovered::Applied(2));
        assert_eq!(read(&root, "child.md").as_deref(), Some("child"));
        assert_eq!(read(&root, "parent.md").as_deref(), Some("new parent"));
        assert!(
            !Journal::default().path_in(&root).exists(),
            "journal must be cleared after recovery"
        );
    }

    #[test]
    fn recovery_completes_a_partially_applied_change_set() {
        // Crash mid-apply: the first write landed, the second did not.
        let root = tmp("recover-partial");
        std::fs::write(root.join("parent.md"), "old parent").unwrap();
        let ops = vec![
            FileOp::Write {
                path: "child.md".into(),
                bytes: b"child".to_vec(),
            },
            FileOp::Write {
                path: "parent.md".into(),
                bytes: b"new parent".to_vec(),
            },
        ];
        std::fs::write(Journal::default().path_in(&root), encode(&ops).unwrap()).unwrap();
        // Simulate the first op having landed before the crash.
        std::fs::write(root.join("child.md"), "child").unwrap();

        block_on(recover(&StdFs, &root)).unwrap();

        assert_eq!(read(&root, "child.md").as_deref(), Some("child"));
        assert_eq!(read(&root, "parent.md").as_deref(), Some("new parent"));
        assert!(!Journal::default().path_in(&root).exists());
    }

    #[test]
    fn recovery_rolls_a_rename_forward_from_either_side_of_the_crash() {
        // A rename is the one non-idempotent op. Recovery must complete it whether
        // the crash struck before it (source still present) or after (only the
        // destination present).
        for already_moved in [false, true] {
            let root = tmp(&format!("recover-rename-{already_moved}"));
            let ops = vec![FileOp::Rename {
                from: "a.md".into(),
                to: "sub/a.md".into(),
            }];
            std::fs::write(Journal::default().path_in(&root), encode(&ops).unwrap()).unwrap();
            if already_moved {
                std::fs::create_dir_all(root.join("sub")).unwrap();
                std::fs::write(root.join("sub/a.md"), "moved").unwrap();
            } else {
                std::fs::write(root.join("a.md"), "moved").unwrap();
            }

            block_on(recover(&StdFs, &root)).unwrap();

            assert_eq!(read(&root, "sub/a.md").as_deref(), Some("moved"));
            assert!(!root.join("a.md").exists());
            assert!(!Journal::default().path_in(&root).exists());
        }
    }

    #[test]
    fn recovery_rolls_a_copy_forward_from_its_immutable_source() {
        // The restore shape: a crash after the commit point, with the payload
        // still sitting in a content-addressed blob. Replay reads it back and
        // lands the file, whether or not the copy ran before the crash.
        for already_copied in [false, true] {
            let root = tmp(&format!("recover-copy-{already_copied}"));
            std::fs::create_dir_all(root.join("history/blobs/9f")).unwrap();
            std::fs::write(root.join("history/blobs/9f/86d081"), "captured bytes").unwrap();
            std::fs::write(root.join("notes.md"), "damaged bytes").unwrap();
            let ops = vec![FileOp::CopyFrom {
                path: "notes.md".into(),
                source: "history/blobs/9f/86d081".into(),
            }];
            std::fs::write(Journal::default().path_in(&root), encode(&ops).unwrap()).unwrap();
            if already_copied {
                std::fs::write(root.join("notes.md"), "captured bytes").unwrap();
            }

            block_on(recover(&StdFs, &root)).unwrap();

            assert_eq!(read(&root, "notes.md").as_deref(), Some("captured bytes"));
            assert!(!Journal::default().path_in(&root).exists());
            // The source is read, never consumed: the blob is shared by every
            // event that names it and must survive the restore.
            assert!(root.join("history/blobs/9f/86d081").exists());
        }
    }

    #[test]
    fn a_copy_whose_source_is_gone_fails_replay_rather_than_inventing_a_state() {
        // The cost of journaling a reference: if the referent is missing at replay
        // time there is nothing to fall back on. That must be loud — writing
        // nothing, or writing something else, would be recovery reaching a state
        // the original change set never intended.
        let root = tmp("recover-copy-missing");
        std::fs::write(root.join("notes.md"), "damaged bytes").unwrap();
        let ops = vec![FileOp::CopyFrom {
            path: "notes.md".into(),
            source: "history/blobs/9f/86d081".into(),
        }];
        std::fs::write(Journal::default().path_in(&root), encode(&ops).unwrap()).unwrap();

        let err = block_on(recover(&StdFs, &root)).unwrap_err();
        assert!(err.to_string().contains("cannot copy"), "{err}");
        // The journal stays, so the next recovery can finish once the blob arrives.
        assert!(Journal::default().path_in(&root).exists());
        assert_eq!(read(&root, "notes.md").as_deref(), Some("damaged bytes"));
    }

    #[cfg(unix)]
    #[test]
    fn recovery_rolls_modes_and_links_forward_idempotently() {
        use std::os::unix::fs::PermissionsExt as _;

        // Crash after the commit point with the link already made and the bit
        // not yet flipped: replay must redo both without tripping over the
        // half that had landed.
        let root = tmp("recover-modes-links");
        std::fs::write(root.join("run.sh"), "#!/bin/sh").unwrap();
        std::os::unix::fs::symlink("target.md", root.join("link.md")).unwrap();
        let ops = vec![
            FileOp::SetLink {
                path: "link.md".into(),
                target: "target.md".into(),
            },
            FileOp::SetExecutable {
                path: "run.sh".into(),
                executable: true,
            },
        ];
        std::fs::write(Journal::default().path_in(&root), encode(&ops).unwrap()).unwrap();

        let outcome = block_on(recover(&StdFs, &root)).unwrap();

        assert_eq!(outcome, Recovered::Applied(2));
        assert_eq!(
            std::fs::read_link(root.join("link.md")).unwrap(),
            PathBuf::from("target.md")
        );
        let mode = std::fs::metadata(root.join("run.sh"))
            .unwrap()
            .permissions()
            .mode();
        assert_ne!(mode & 0o111, 0, "the bit must be set after recovery");
    }

    #[test]
    fn recovery_refuses_a_journal_whose_paths_escape_the_root() {
        // A journal is bytes found on disk, not a set this process staged —
        // synced from another machine, or planted — and the checksum
        // authenticates nothing. Replay must clamp to the root exactly as an
        // apply would, or the escape guard has a back door.
        let root = tmp("recover-escape");
        let outside = root.join("../fstx-escaped-by-recovery.md");
        let _ = std::fs::remove_file(&outside);
        let ops = vec![FileOp::Write {
            path: "../fstx-escaped-by-recovery.md".into(),
            bytes: b"escaped".to_vec(),
        }];
        std::fs::write(Journal::default().path_in(&root), encode(&ops).unwrap()).unwrap();

        let err = block_on(recover(&StdFs, &root)).unwrap_err();

        assert!(matches!(err, crate::Error::Escape(_)), "{err:?}");
        assert!(!outside.exists(), "nothing may land outside the root");
        assert!(
            Journal::default().path_in(&root).exists(),
            "a refused journal is left in place, like any that cannot be trusted"
        );
    }

    #[test]
    fn an_impossible_op_count_is_refused_not_allocated() {
        // A crafted header claiming u64::MAX ops must surface as Corrupt —
        // sizing a Vec by it aborts the process on capacity overflow, which
        // is a denial of service handed to whoever can write the journal.
        let mut bytes = Vec::new();
        bytes.extend_from_slice(MAGIC);
        bytes.extend_from_slice(&u64::MAX.to_le_bytes());
        let checksum = fnv1a(&bytes);
        bytes.extend_from_slice(&checksum.to_le_bytes());

        let err = decode(&bytes).unwrap_err();
        assert!(err.to_string().contains("op count"), "{err}");
    }

    #[test]
    fn recovery_flushes_what_it_replayed_before_giving_up_the_journal() {
        // Recovery keeps the same promise a clean apply does: once the
        // journal is gone, the state it certified survives a power cut. The
        // replayed rename's entries must be flushed before the deletion.
        let root = tmp("recover-flush");
        std::fs::write(root.join("a.md"), "a").unwrap();
        let ops = vec![FileOp::Rename {
            from: "a.md".into(),
            to: "b.md".into(),
        }];
        std::fs::write(Journal::default().path_in(&root), encode(&ops).unwrap()).unwrap();

        let fs = crate::fs_faults::RecordingFs::local();
        let outcome = block_on(recover(&fs, &root)).unwrap();
        assert_eq!(outcome, Recovered::Applied(1));

        use crate::fs_faults::FsEvent;
        assert_eq!(
            fs.events(),
            vec![
                FsEvent::Rename(root.join("a.md"), root.join("b.md")),
                FsEvent::Sync(root.clone(), crate::fs::Durability::Durable),
                FsEvent::Remove(Journal::default().path_in(&root)),
            ]
        );
    }

    #[test]
    fn recovery_is_a_noop_when_there_is_no_journal() {
        let root = tmp("recover-noop");
        std::fs::write(root.join("doc.md"), "untouched").unwrap();
        assert_eq!(
            block_on(recover(&StdFs, &root)).unwrap(),
            Recovered::Nothing
        );
        assert_eq!(read(&root, "doc.md").as_deref(), Some("untouched"));
    }

    #[test]
    fn recovering_the_same_journal_twice_is_safe() {
        // A crash *during* recovery must be survivable: replaying an already-
        // recovered (or re-created) journal reaches the same state, never an error.
        let root = tmp("recover-twice");
        std::fs::write(root.join("parent.md"), "old").unwrap();
        let ops = vec![FileOp::Write {
            path: "parent.md".into(),
            bytes: b"new".to_vec(),
        }];
        let journal = encode(&ops).unwrap();

        std::fs::write(Journal::default().path_in(&root), &journal).unwrap();
        block_on(recover(&StdFs, &root)).unwrap();
        // Recovery removed the journal; imagine the crash left it and re-run.
        std::fs::write(Journal::default().path_in(&root), &journal).unwrap();
        block_on(recover(&StdFs, &root)).unwrap();

        assert_eq!(read(&root, "parent.md").as_deref(), Some("new"));
        assert!(!Journal::default().path_in(&root).exists());
    }

    // ---- a configurable journal name ----

    #[test]
    fn a_journal_name_must_be_a_single_component() {
        // The name is joined onto a caller-supplied root, so anything that could
        // climb out of it has to be refused where it is built, not where it is
        // used.
        for bad in ["", ".", "..", "a/b", "../elsewhere", "/absolute"] {
            assert!(
                Journal::named(bad).is_err(),
                "{bad:?} should be refused as a journal name"
            );
        }
        assert_eq!(
            Journal::named(".myapp-journal").unwrap().name(),
            ".myapp-journal"
        );
    }

    #[test]
    fn apply_and_recover_meet_at_the_named_journal() {
        // The point of the type: a set applied under one name is recovered under
        // that same name, and the default is not it.
        let root = tmp("named-journal");
        let journal = Journal::named(".myapp-journal").unwrap();
        std::fs::write(root.join("parent.md"), "old parent").unwrap();

        let ops = vec![
            FileOp::Write {
                path: "child.md".into(),
                bytes: b"child".to_vec(),
            },
            FileOp::Write {
                path: "parent.md".into(),
                bytes: b"new parent".to_vec(),
            },
        ];
        // The state a crash just after the commit point leaves behind.
        std::fs::write(journal.path_in(&root), encode(&ops).unwrap()).unwrap();

        // The default journal names a file that is not there, so it finds
        // nothing — the silent-mismatch failure this API exists to make hard.
        assert_eq!(
            block_on(Journal::default().recover(&StdFs, &root)).unwrap(),
            Recovered::Nothing
        );
        assert_eq!(read(&root, "parent.md").as_deref(), Some("old parent"));

        assert_eq!(
            block_on(journal.recover(&StdFs, &root)).unwrap(),
            Recovered::Applied(2)
        );
        assert_eq!(read(&root, "parent.md").as_deref(), Some("new parent"));
        assert!(!journal.path_in(&root).exists());
    }

    #[test]
    fn a_named_journal_round_trips_a_whole_apply() {
        let root = tmp("named-apply");
        let journal = Journal::named(".myapp-journal").unwrap();
        let mut cs = crate::ChangeSet::new();
        cs.write("a.md", "one");
        cs.write("b.md", "two");
        block_on(journal.apply(&cs, &StdFs, &root)).unwrap();

        assert_eq!(read(&root, "a.md").as_deref(), Some("one"));
        assert_eq!(read(&root, "b.md").as_deref(), Some("two"));
        // Cleared on a clean apply, and the default was never written.
        assert!(!journal.path_in(&root).exists());
        assert!(!Journal::default().path_in(&root).exists());
    }

    // ---- a journal kept outside the tree ----

    #[test]
    fn a_homed_journal_owns_no_path_in_the_root() {
        // The homed design's whole point: no file in the root is this
        // journal, so a same-named file there — synced in, or coincidence —
        // must not be claimed.
        let home = tmp("owns-home");
        let root = tmp("owns-root");
        let journal = Journal::default().kept_in(&home).unwrap();
        assert!(journal.owns_path(&journal.path_in(&root)));
        assert!(!journal.owns_path(&root.join(Journal::DEFAULT_NAME)));
    }

    #[test]
    fn a_home_must_be_absolute() {
        // A relative home resolves against the process's current directory,
        // which apply and recovery have no reason to share — refused at
        // construction, not discovered as a stranded change.
        let err = Journal::default().kept_in("relative/dir").unwrap_err();
        assert!(
            matches!(err, crate::Error::InvalidJournalHome(_)),
            "{err:?}"
        );
    }

    #[test]
    fn a_homed_journal_keeps_the_root_journal_free() {
        // The synced-folder deployment: the tree must never hold the journal,
        // transiently or otherwise, because a sync service cannot tell crash
        // state from content. The recording backend sees every write of the
        // apply, so "no write ever landed in the root under the journal's
        // name" is checked as stated, not just at the end.
        let root = tmp("homed-apply");
        let home = tmp("homed-apply-home");
        let journal = Journal::default().kept_in(&home).unwrap();

        let fs = crate::fs_faults::RecordingFs::local();
        let mut cs = crate::ChangeSet::new();
        cs.write("a.md", "a");
        cs.write("b.md", "b");
        block_on(journal.apply(&cs, &fs, &root)).unwrap();

        assert_eq!(read(&root, "a.md").as_deref(), Some("a"));
        let stray = fs.events().iter().any(|e| {
            matches!(e, crate::fs_faults::FsEvent::Write(p)
                if p.starts_with(&root)
                    && p.file_name()
                        .and_then(|n| n.to_str())
                        .is_some_and(|n| n.contains(Journal::DEFAULT_NAME)))
        });
        assert!(!stray, "events: {:?}", fs.events());
        assert!(
            !journal.path_in(&root).exists(),
            "the homed journal is cleared after a clean apply"
        );
    }

    #[test]
    fn recovery_finds_a_homed_journal_and_applies_it_to_the_root() {
        // The two halves meeting away from the tree: the journal lives in the
        // home, the ops land in the root.
        let root = tmp("homed-recover");
        let home = tmp("homed-recover-home");
        let journal = Journal::default().kept_in(&home).unwrap();
        let ops = vec![FileOp::Write {
            path: "restored.md".into(),
            bytes: b"restored".to_vec(),
        }];
        std::fs::write(journal.path_in(&root), encode(&ops).unwrap()).unwrap();

        let outcome = block_on(journal.recover(&StdFs, &root)).unwrap();

        assert_eq!(outcome, Recovered::Applied(1));
        assert_eq!(read(&root, "restored.md").as_deref(), Some("restored"));
        assert!(!journal.path_in(&root).exists());
    }

    #[test]
    fn a_stale_homed_journal_still_refuses_the_next_apply() {
        let root = tmp("homed-stale");
        let home = tmp("homed-stale-home");
        let journal = Journal::default().kept_in(&home).unwrap();
        std::fs::write(journal.path_in(&root), b"whatever a crash left").unwrap();

        let mut cs = crate::ChangeSet::new();
        cs.write("a.md", "a");
        cs.write("b.md", "b");
        let err = block_on(journal.apply(&cs, &StdFs, &root)).unwrap_err();
        assert!(matches!(err, crate::Error::StaleJournal(_)), "{err:?}");
        assert_eq!(read(&root, "a.md"), None);
    }

    #[test]
    fn an_apply_makes_a_home_that_does_not_exist_yet() {
        // A cache directory on a fresh machine: the home is the journal's own
        // infrastructure, so the apply makes it rather than failing on it.
        let root = tmp("homed-fresh");
        let home = tmp("homed-fresh-home").join("nested/never-made");
        let journal = Journal::default().kept_in(&home).unwrap();

        let mut cs = crate::ChangeSet::new();
        cs.write("a.md", "a");
        cs.write("b.md", "b");
        block_on(journal.apply(&cs, &StdFs, &root)).unwrap();
        assert_eq!(read(&root, "b.md").as_deref(), Some("b"));
    }

    #[test]
    fn a_freshly_made_home_is_flushed_before_the_intent_is_trusted_to_it() {
        // The commit point is only as durable as the chain of names holding
        // it: a journal file flushed into a directory whose own entry was
        // never flushed is one a power cut deletes wholesale — a
        // half-applied set with no record to roll forward. Every directory
        // the home's making mints must be flushed durable before the journal
        // is written.
        let base = tmp("homed-flush");
        let home = base.join("nested/journals");
        let journal = Journal::default().kept_in(&home).unwrap();
        let root = tmp("homed-flush-root");

        let fs = crate::fs_faults::RecordingFs::local();
        let mut cs = crate::ChangeSet::new();
        cs.write("a.md", "a");
        cs.write("b.md", "b");
        block_on(journal.apply(&cs, &fs, &root)).unwrap();

        let events = fs.events();
        let journal_written = events
            .iter()
            .position(|e| matches!(e, crate::fs_faults::FsEvent::Write(p) if journal.owns_path(p)))
            .expect("the journal must be written");
        for dir in [base, home.parent().unwrap().to_path_buf(), home] {
            let flushed = events.iter().position(|e| {
                matches!(e, crate::fs_faults::FsEvent::Sync(p, crate::fs::Durability::Durable)
                    if *p == dir)
            });
            match flushed {
                Some(at) => assert!(
                    at < journal_written,
                    "{} flushed only after the journal was written",
                    dir.display()
                ),
                None => panic!(
                    "{} never flushed durable; events: {events:?}",
                    dir.display()
                ),
            }
        }
    }

    #[test]
    fn the_pre_extraction_magic_still_replays() {
        // A journal written by `prov` before this crate was lifted out of it
        // carries the older stamp. The format is identical, and the only tree
        // that can be holding one is a tree interrupted mid-apply — refusing it
        // would strand exactly the change a journal exists to finish.
        let ops = vec![FileOp::Write {
            path: "parent.md".into(),
            bytes: b"new".to_vec(),
        }];
        let mut bytes = encode(&ops).unwrap();
        assert_eq!(&bytes[..MAGIC.len()], MAGIC);
        bytes[..LEGACY_MAGIC.len()].copy_from_slice(LEGACY_MAGIC);
        // The checksum covers the magic, so re-stamping invalidates it; a real
        // legacy journal carries the checksum for its own bytes.
        let body_end = bytes.len() - 8;
        let checksum = fnv1a(&bytes[..body_end]);
        bytes[body_end..].copy_from_slice(&checksum.to_le_bytes());

        assert_eq!(decode(&bytes).unwrap(), ops);
    }

    #[test]
    fn a_journal_is_only_ever_written_with_the_current_magic() {
        let bytes = encode(&[FileOp::Remove {
            path: "gone.md".into(),
        }])
        .unwrap();
        assert_eq!(&bytes[..MAGIC.len()], MAGIC);
        assert_ne!(&bytes[..LEGACY_MAGIC.len()], LEGACY_MAGIC);
    }
}