Crash safety: durable writes and the op head as commit point 4dcb8c0916b7

//! Durable file updates (spec §13.6): every write that must survive a

//! crash goes through here.

//!

//! A file is replaced by writing a uniquely named temporary file in the

//! same directory, syncing it, renaming it over the target, and syncing

//! the directory so the rename itself is on disk. A crash at any point

//! leaves either the old file or the new one, never a torn one; at worst

//! a stray temporary file, which nothing reads.

//!

//! In tests, `crash_after(n)` makes the n-th durable step on the current

//! thread fail as if the process had died there, so a test can stop an

//! operation at every step and check what is left behind.

use rand::RngCore;

use std::fs;

use std::io::{self, Write};

use std::path::Path;

/// Replace `path` with `data`, atomically and durably.

pub fn write_atomic(path: &Path, data: &[u8]) -> io::Result<()> {

    let dir = path.parent().unwrap_or(Path::new("."));

    let name = path

        .file_name()

        .map(|n| n.to_string_lossy().into_owned())

        .unwrap_or_default();

    let mut nonce = [0u8; 6];

    rand::rngs::OsRng.fill_bytes(&mut nonce);

    let tmp = dir.join(format!(".{name}.tmp-{}", hex::encode(nonce)));

    step("create temporary file")?;

    let written = (|| {

        let mut f = fs::File::create(&tmp)?;

        f.write_all(data)?;

        f.sync_all()?;

        step("rename into place")?;

        fs::rename(&tmp, path)

    })();

    if let Err(e) = written {

        if !is_injected(&e) {

            let _ = fs::remove_file(&tmp);

        }

        return Err(e);

    }

    sync_dir(dir)

}

/// Rename `from` to `to` and make the rename durable.

pub fn rename(from: &Path, to: &Path) -> io::Result<()> {

    step("rename")?;

    fs::rename(from, to)?;

    sync_dir(to.parent().unwrap_or(Path::new(".")))

}

/// Flush a directory's entries (new, renamed or removed files) to disk.

/// A no-op where directories cannot be opened for syncing.

pub fn sync_dir(dir: &Path) -> io::Result<()> {

    #[cfg(unix)]

    {

        fs::File::open(dir)?.sync_all()

    }

    #[cfg(not(unix))]

    {

        let _ = dir;

        Ok(())

    }

}

/// A point where a crash may happen. Outside tests it does nothing.

pub fn step(what: &str) -> io::Result<()> {

    #[cfg(test)]

    {

        failpoint::hit(what)

    }

    #[cfg(not(test))]

    {

        let _ = what;

        Ok(())

    }

}

const INJECTED: &str = "injected crash";

fn is_injected(e: &io::Error) -> bool {

    e.to_string().starts_with(INJECTED)

}

#[cfg(test)]

pub(crate) mod failpoint {

    use std::cell::Cell;

    use std::io;

    thread_local! {

        static COUNTDOWN: Cell<Option<usize>> = const { Cell::new(None) };

        static FIRED: Cell<bool> = const { Cell::new(false) };

    }

    /// Fail the `n`-th durable step from now (1-based) on this thread,

    /// and every step after it, as a dead process would.

    pub fn crash_after(n: usize) {

        COUNTDOWN.with(|c| c.set(Some(n)));

        FIRED.with(|f| f.set(false));

    }

    /// Stop injecting; returns whether a crash was injected.

    pub fn disarm() -> bool {

        COUNTDOWN.with(|c| c.set(None));

        FIRED.with(|f| f.replace(false))

    }

    pub(super) fn hit(what: &str) -> io::Result<()> {

        if FIRED.with(|f| f.get()) {

            return Err(io::Error::other(format!(

                "{} (after) at {what}",

                super::INJECTED

            )));

        }

        COUNTDOWN.with(|c| match c.get() {

            Some(1) => {

                c.set(None);

                FIRED.with(|f| f.set(true));

                Err(io::Error::other(format!("{} at {what}", super::INJECTED)))

            }

            Some(n) => {

                c.set(Some(n - 1));

                Ok(())

            }

            None => Ok(()),

        })

    }

}

pub mod durable;

    Ok(crate::durable::write_atomic(path, data)?)

        // The head is the commit point: everything it names must be on disk first.

        self.store.flush()?;

/// Copy a directory tree (for restoring a pristine repository).

fn copy_tree(from: &Path, to: &Path) {

    fs::create_dir_all(to).unwrap();

    for e in fs::read_dir(from).unwrap() {

        let e = e.unwrap();

        let target = to.join(e.file_name());

        if e.file_type().unwrap().is_dir() {

            copy_tree(&e.path(), &target);

        } else {

            fs::copy(e.path(), &target).unwrap();

        }

    }

}

/// Spec §13.6: stop `commit` and `push` at every durable step, as a crash

/// would, and check that both repositories still open, pass `fsck`, and

/// accept the same operation again.

#[test]

fn crash_at_every_step_of_commit_and_push_leaves_sound_repositories() {

    use crate::durable::failpoint;

    hermetic();

    let dir = tempfile::tempdir().unwrap();

    let pristine = dir.path().join("pristine");

    let origin_path = pristine.join("origin");

    Repo::init(&origin_path).unwrap();

    let mut a = Repo::clone(&origin_path.to_string_lossy(), &pristine.join("a"))

        .unwrap()

        .with_home(dir.path().join("home"));

    a.create_identity("Ada", "[email protected]", IdentityScope::Repo)

        .unwrap();

    write(&pristine.join("a"), "f.txt", b"one\n");

    a.commit("first", &[]).unwrap();

    assert!(a.push("origin", false).unwrap()[0].ok);

    drop(a);

    type Op = fn(&Repo) -> Result<()>;

    let edit_and_commit: Op = |r| {

        write(r.root(), "f.txt", b"one\ntwo\n");

        write(r.root(), "big.bin", &vec![7u8; 300_000]);

        r.commit("second", &[]).map(|_| ())

    };

    let push: Op = |r| {

        let res = r.push("origin", false)?;

        if res.iter().all(|x| x.ok) {

            Ok(())

        } else {

            Err(RepoError::Msg(format!("{res:?}")))

        }

    };

    for (name, prepare, op) in [

        ("commit", None, edit_and_commit),

        ("push", Some(edit_and_commit), push),

    ] {

        let mut steps = 0;

        for n in 1.. {

            let run = dir.path().join(format!("{name}-{n}"));

            copy_tree(&pristine, &run);

            // Clones record their remote by path; point it at this copy.

            let open = |p: &Path| {

                let mut r = Repo::open(p).unwrap().with_home(dir.path().join("home"));

                r.config

                    .set("remote.origin.url", &run.join("origin").to_string_lossy())

                    .unwrap();

                r.save_config().unwrap();

                r

            };

            let a = open(&run.join("a"));

            if let Some(prep) = prepare {

                prep(&a).unwrap();

            }

            failpoint::crash_after(n);

            let outcome = op(&a);

            let crashed = failpoint::disarm();

            drop(a);

            if !crashed {

                outcome.unwrap_or_else(|e| panic!("{name} failed without a crash: {e}"));

                steps = n - 1;

                break;

            }

            assert!(

                outcome.is_err(),

                "{name}: a crash at step {n} was swallowed"

            );

            for repo in ["a", "origin"] {

                let r = Repo::open(&run.join(repo)).unwrap_or_else(|e| {

                    panic!("{name} crash at step {n}: {repo} no longer opens: {e}")

                });

                let rep = r.fsck().unwrap();

                assert_eq!(

                    rep.errors(),

                    0,

                    "{name} crash at step {n}: {repo} fails fsck: {:?}",

                    rep.problems

                );

            }

            // The interrupted operation can simply be run again.

            let a = open(&run.join("a"));

            // A crash after the commit point leaves the commit done.

            let landed = a

                .bookmark_heads()

                .unwrap()

                .iter()

                .any(|h| a.store().get_change(h).unwrap().message == "second");

            let again = match (name, landed) {

                ("commit", true) => Ok(()),

                ("commit", false) => a.commit("second", &[]).map(|_| ()),

                _ => op(&a),

            };

            again.unwrap_or_else(|e| panic!("{name} crash at step {n}: retry failed: {e}"));

            assert_eq!(a.fsck().unwrap().errors(), 0);

            fs::remove_dir_all(&run).unwrap();

        }

        assert!(steps >= 3, "{name} has only {steps} durable steps?");

    }

}

        // The pack first: an index never names a pack that is not there.

        crate::durable::rename(tmp_pack, &pack_path)?;

        crate::durable::rename(&tmp_idx, &idx_path)?;

        self.write_pending(pending)

    }

    /// Make every object written so far durable, even inside an open

    /// batch, which stays open. Call before a reference that names them

    /// is moved (spec §13.6): the operation head, a bookmark pointer.

    pub fn flush(&self) -> Result<()> {

        let Some(pending) = self.pending.borrow_mut().take() else {

            return Ok(());

        };

        self.write_pending(pending)?;

        *self.pending.borrow_mut() = Some(Pending {

            buffered: vec![],

            buffered_bytes: 0,

            lookup: Default::default(),

            pack: None,

            report: PackReport::default(),

        });

        Ok(())

    }

    fn write_pending(&self, pending: Pending) -> Result<()> {

        let fanout = path.parent().unwrap();

        if !fanout.is_dir() {

            fs::create_dir_all(fanout)?;

            crate::durable::sync_dir(&self.dir.join("loose"))?;

        }

        crate::durable::step("write loose object")?;

        crate::durable::rename(&tmp, &path)?;

        crate::durable::write_atomic(&path, text.as_bytes())?;

        crate::durable::rename(&tmp, &path)?;

### 13.6 Durability

Every update that must survive a crash replaces a file atomically: write a uniquely named temporary file in the same directory, `fsync` it, rename it over the target, and `fsync` the directory. Loose objects and packs are written the same way, a pack before its index. The operation head (`.oxy/ops/head`) is the commit point of every command: before it moves, every object the new operation names MUST be durable, including objects still held by an open write batch (§13.4), which is flushed first. A crash therefore leaves the repository at the operation before or after the interrupted command, never between; stray temporary files may remain and are ignored. The reference implementation's tests stop `commit` and `push` at each durable step and require both repositories to open, pass `oxy fsck` (§21.1) and accept the same command again.

*Changelog.* Draft 0.8h (2026-09-28): `oxy fsck` (§21.1); durability rules and the operation head as commit point (§13.6). Draft 0.8g (2026-09-27): delegates of a rule's identities may sign reference attestations (§14.5); local-path pushes bind unbound repositories like HTTP ones. Draft 0.8f (2026-09-27): challenge signatures (§14.3) for hosts. Draft 0.8e (2026-09-27): anonymous changes (zero author) are authenticated by a seal from any key, so imported and pre-identity history can be pushed under a policy (§14.4). Draft 0.8d (2026-09-27): unbound hosted repositories adopt the first pusher's id (§16.1). Draft 0.8c (2026-09-27): bare repositories (`oxy init --bare`, §12.2: no workspace, no working copy; hosting) and the reference server as an embeddable library (§15.1). Draft 0.8b (2026-09-27): the machine interface for editors (§24): `root`, `cat`, `version`, `hunks`, `discard`, and `--json` on the reporting commands.