gunnar_server/

durable.rs

//! One place that writes a small control file durably.
//!
//! Both `replication.json` and `webhooks.json` are tiny JSON documents in the
//! data root that an operator sets over the control plane and that MUST survive
//! a power cut: a box that acknowledged the change and then lost power has to
//! come back with it. The write is the same on both, and getting it subtly
//! wrong is what left a 0-byte `replication.json` that panicked PID 1 on the
//! next boot. So there is one writer, and it is here (one implementation).
//!
//! # Why the fsync dance
//!
//! `std::fs::write` + `rename` is NOT durable on xfs: the rename's directory
//! entry is journaled while the data blocks are not, so an interrupted write
//! leaves the file present but empty. So the bytes are fsync'd BEFORE the
//! rename, and the directory is fsync'd AFTER it — a crash then leaves either
//! the whole old file or the whole new one, never a torn tail.

use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use anyhow::{Context, Result};

/// Exponential backoff: `base · 2^(attempt-1)`, capped at `max`; zero for
/// `attempt == 0`.
///
/// One writer for the retry ladder both the replicator and the webhook worker
/// walk (one implementation) — a "still trying" schedule that a black-holing endpoint cannot
/// turn into a busy loop or an overflow.
pub fn backoff(attempt: u32, base: Duration, max: Duration) -> Duration {
    if attempt == 0 {
        return Duration::ZERO;
    }
    // Clamp the shift: 2^30 · a millisecond base already exceeds any sane cap,
    // and clamping keeps the multiply from wrapping before `min` sees it.
    let factor = 1u32.checked_shl((attempt - 1).min(30)).unwrap_or(u32::MAX);
    base.saturating_mul(factor).min(max)
}

/// Atomically replace `<root>/<filename>` with `bytes`, durably.
///
/// Writes a `.tmp` sibling, fsyncs it, renames it over the target, then fsyncs
/// the directory so the rename itself survives a cut and not just the bytes it
/// points at.
pub fn write_atomic(root: &Path, filename: &str, bytes: &[u8]) -> Result<()> {
    // The twin hears of it FIRST (`gunnar_core::txlog::sink`): a document
    // under a primary's data root is a `File` frame before it is a file.
    let _framed = gunnar_core::txlog::sink::replace_file(&root.join(filename), bytes, false)
        .with_context(|| format!("framing {filename} in the transaction log"))?;
    write_atomic_unframed(root, filename, bytes)
}

/// [`write_atomic`] without the frame: the per-box metadata MIRROR (`slow`'s
/// `meta`), which is a second copy on this box's other spindle and never a
/// transaction — the fallback slow root sits UNDER the data root, where a
/// framed mirror write would put every document on the twin twice.
pub fn write_atomic_unframed(root: &Path, filename: &str, bytes: &[u8]) -> Result<()> {
    static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
    let path = root.join(filename);
    // A temp of this save's own (stinky row 8): two saves of one document used
    // to share `<file>.tmp` and truncate, or rename away, each other's bytes.
    // It keeps the `.tmp` suffix — `txlog::is_cache_path` knows it as scratch.
    let seq = TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
    let temp = root.join(format!("{filename}.{}-{seq}.tmp", std::process::id()));
    let written = (|| -> Result<()> {
        let mut f =
            std::fs::File::create(&temp).with_context(|| format!("writing {}", temp.display()))?;
        std::io::Write::write_all(&mut f, bytes)
            .with_context(|| format!("writing {}", temp.display()))?;
        f.sync_all()
            .with_context(|| format!("flushing {} to disk", temp.display()))?;
        std::fs::rename(&temp, &path).with_context(|| format!("replacing {}", path.display()))
    })();
    if let Err(e) = written {
        let _ = std::fs::remove_file(&temp);
        return Err(e);
    }
    // The rename is durable only once the directory is: a save whose
    // directory fsync failed is not reported as saved.
    sync_dir(root).with_context(|| format!("flushing the save of {filename} in {}", root.display()))
}

/// **fsync a directory**, so a rename into it or an unlink from it survives a
/// power cut — its failure is the caller's failure, never swallowed.
pub fn sync_dir(dir: &Path) -> Result<()> {
    #[cfg(test)]
    if let Some(e) = dir_sync_seam::injected() {
        return Err(e).with_context(|| format!("flushing {}", dir.display()));
    }
    let d = std::fs::File::open(dir)
        .with_context(|| format!("opening {} to flush it", dir.display()))?;
    d.sync_all()
        .with_context(|| format!("flushing {}", dir.display()))
}

/// **Remove a per-box MARKER durably**: unlink it, then fsync the directory, so
/// the removal survives a power cut and not only the next `stat`.
///
/// MEASURED under KVM 2026-09-21 (lane T3): `twin::restore_finish` unlinked
/// `txlog-restoring` with a bare `remove_file`, and the operator's "restart the
/// box" came as a stop ten seconds later. The unlink was still in the guest's
/// page cache; the restored box reopened WITH the marker, as a box mid-restore —
/// follower-only, every control-plane write refused ("this box is a twin (or
/// restoring); it originates no transactions"), its audit trail written to a
/// per-box path, no re-seal. A marker whose PRESENCE changes what a box is must
/// have its ABSENCE made as durable as its presence was (`write_atomic_unframed`
/// already fsyncs the directory after creating one). A missing file is `Ok`.
pub fn remove_durable(root: &Path, filename: &str) -> Result<()> {
    match std::fs::remove_file(root.join(filename)) {
        Ok(()) => {}
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(e) => {
            return Err(e).with_context(|| format!("removing {}", root.join(filename).display()))
        }
    }
    sync_dir(root)
        .with_context(|| format!("flushing the removal of {filename} from {}", root.display()))
}

/// [`write_atomic`], and then the SAME bytes again into the metadata mirror
/// (`crate::slow`'s `dir.meta.path` — a second copy on the other spindle).
///
/// # Why the mirror cannot fail the save
///
/// The primary write has already landed and been fsync'd by the time the
/// mirror is attempted. Returning the mirror's error would report a save that
/// DID happen as a failure, and the caller's only sane response — retry — would
/// rewrite a file that is already correct. Worse, on the appliance the caller
/// is a control-plane verb an operator typed on a box with no shell: a full or
/// read-only slow disk would take the roster away from them entirely.
///
/// So a mirror failure is one WARN naming the path, and the save is `Ok`. What
/// stops that from becoming a silent lie is the other half of this design:
/// `SlowRoot::claim_of` never reports the mirror as satisfied unless the layout
/// config's `dir.meta.off_stripe_satisfied` says so, and so nobody plans a
/// recovery around a mirror that is not there.
pub fn write_atomic_mirrored(
    root: &Path,
    slow: &crate::slow::SlowRoot,
    filename: &str,
    bytes: &[u8],
) -> Result<()> {
    write_atomic(root, filename, bytes)?;
    let mirrored = slow
        .meta_dir()
        .and_then(|dir: std::path::PathBuf| write_atomic_unframed(&dir, filename, bytes));
    if let Err(err) = mirrored {
        tracing::warn!(
            target: crate::slow::TARGET_SLOW,
            file = filename,
            mirror_dir = %slow.path_of(crate::slow::META_DIR).display(),
            error = %format!("{err:#}"),
            "the metadata mirror on the slow disk could not be written; the file itself is \
             saved on /data. This box now has ONE copy of it."
        );
    }
    Ok(())
}

/// The test seam over the directory fsync after a rename: an injected failure.
#[cfg(test)]
pub(crate) mod dir_sync_seam {
    use std::cell::Cell;

    thread_local! {
        static FAIL: Cell<bool> = const { Cell::new(false) };
    }

    /// Make every later directory fsync on this thread fail.
    pub(crate) fn fail_from_now() {
        FAIL.with(|f| f.set(true));
    }

    /// The injected failure, if armed.
    pub(crate) fn injected() -> Option<std::io::Error> {
        FAIL.with(|f| f.get())
            .then(|| std::io::Error::other("injected directory fsync failure"))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// ★ **Two writers of one file do not share a temp name** (stinky row 8).
    /// Every writer used `<file>.tmp`, so two saves of one document truncated
    /// each other's bytes mid-write, or one renamed the other's half-written
    /// file into place, or found its temp already renamed away.
    #[test]
    fn concurrent_writers_of_one_file_never_trip_over_each_others_temp() {
        let dir = tempfile::tempdir().unwrap();
        let root = std::sync::Arc::new(dir.path().to_path_buf());
        let writers: Vec<_> = (0..8u8)
            .map(|w| {
                let root = std::sync::Arc::clone(&root);
                std::thread::spawn(move || {
                    let body = vec![b'a' + w; 64 * 1024];
                    (0..40)
                        .map(|_| {
                            write_atomic_unframed(&root, "doc.json", &body)
                                .map_err(|e| format!("{e:#}"))
                        })
                        .collect::<Vec<_>>()
                })
            })
            .collect();
        let failures: Vec<String> = writers
            .into_iter()
            .flat_map(|w| w.join().unwrap())
            .filter_map(|r| r.err())
            .collect();
        assert!(
            failures.is_empty(),
            "{} of 320 saves failed, first: {:?}",
            failures.len(),
            failures.first()
        );
        let landed = std::fs::read(dir.path().join("doc.json")).unwrap();
        assert_eq!(landed.len(), 64 * 1024, "a save landed torn");
        assert!(
            landed.iter().all(|b| *b == landed[0]),
            "a save landed interleaved"
        );
    }

    /// …and a temp left behind by a crashed save does not wedge the next one.
    #[test]
    fn a_leftover_temp_does_not_wedge_the_next_save() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::create_dir(dir.path().join("doc.json.tmp")).unwrap();
        write_atomic_unframed(dir.path(), "doc.json", b"{}").unwrap();
        assert_eq!(std::fs::read(dir.path().join("doc.json")).unwrap(), b"{}");
    }

    /// ★ **A save whose rename was never made durable says so.** The directory
    /// fsync was best-effort, so a box acknowledged a document that a power cut
    /// could still take back.
    #[test]
    fn a_failed_directory_fsync_fails_the_save() {
        let dir = tempfile::tempdir().unwrap();
        dir_sync_seam::fail_from_now();
        assert!(
            write_atomic_unframed(dir.path(), "doc.json", b"{}").is_err(),
            "a save whose directory fsync failed reported success"
        );
    }
}