kache-store 1.2.2

use anyhow::{Context, Result};
use std::fs;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};

static ATOMIC_TMP_NONCE: AtomicU64 = AtomicU64::new(0);

/// fsync a directory so a rename/create into it survives a crash. Without it,
/// a file's contents can be durable while its directory entry is not. No-op on non-Unix:
/// Windows has no directory-handle fsync and `File::open` on a directory fails.
#[cfg(unix)]
pub fn fsync_dir(dir: &Path) -> std::io::Result<()> {
    fs::File::open(dir)?.sync_all()
}
#[cfg(not(unix))]
pub fn fsync_dir(_dir: &Path) -> std::io::Result<()> {
    Ok(())
}

/// fsync a file so its bytes are durable on disk.
/// On Windows, FlushFileBuffers requires write access.
pub fn fsync_file(path: &Path) -> std::io::Result<()> {
    #[cfg(windows)]
    let file = fs::OpenOptions::new().write(true).open(path)?;
    #[cfg(not(windows))]
    let file = fs::File::open(path)?;
    file.sync_all()
}

/// Whether a failed rename is a transient Windows state worth retrying.
/// ERROR_ACCESS_DENIED (5) and ERROR_SHARING_VIOLATION (32) surface when a
/// concurrent put/remove leaves the destination delete-pending or open; both
/// clear on their own. No such transient rename error exists off Windows.
pub(crate) fn is_transient_rename_error(e: &std::io::Error) -> bool {
    #[cfg(windows)]
    {
        matches!(e.raw_os_error(), Some(5) | Some(32))
    }
    #[cfg(not(windows))]
    {
        let _ = e;
        false
    }
}

/// Backoff before the next rename retry: 1, 2, 4, … ms capped at 64 ms
/// (≈0.3 s total across the retry budget), bounding how long a wedged Windows
/// destination can delay a write.
fn rename_backoff_ms(attempt: u32) -> u64 {
    (1u64 << attempt.min(6)).min(64)
}

/// How many times a transient Windows state is waited out before giving up.
pub(crate) const TRANSIENT_ATTEMPTS: u32 = 10;

/// Whether this error should be slept out and tried again. Split from the
/// loop so a `true` match-guard mutant cannot hang: the `for` bound below
/// still stops, and tests of this helper reject `true` / `||` directly.
fn should_retry_transient(transient: bool, attempt: u32) -> bool {
    transient && attempt + 1 < TRANSIENT_ATTEMPTS
}

/// Run a filesystem operation, waiting out a transient Windows sharing or
/// delete-pending state.
///
/// On Unix an unlinked file keeps working through open handles and its name is
/// free the moment it is removed, so a concurrent put/remove pair never
/// collides on the name. Windows keeps the directory entry until the last
/// handle closes and fails anything touching that name meanwhile with
/// ERROR_ACCESS_DENIED or ERROR_SHARING_VIOLATION. Those clear on their own,
/// so they are worth waiting out rather than reporting as a fault.
///
/// Production callers pass [`is_transient_rename_error`] as `is_transient`.
/// Off Windows that predicate is always false, so `op` runs exactly once. It
/// is a parameter so the loop is reachable off Windows: with it inlined, the
/// retry arm is dead code on the Linux runner that scores mutants, and every
/// mutation of the bound and the counter survives against a branch nothing
/// can enter.
pub(crate) fn retry_transient<T>(
    mut op: impl FnMut() -> std::io::Result<T>,
    is_transient: impl Fn(&std::io::Error) -> bool,
) -> std::io::Result<T> {
    let mut last_err = None;
    for attempt in 0..TRANSIENT_ATTEMPTS {
        match op() {
            Ok(value) => return Ok(value),
            Err(e) if should_retry_transient(is_transient(&e), attempt) => {
                std::thread::sleep(std::time::Duration::from_millis(rename_backoff_ms(attempt)));
                last_err = Some(e);
            }
            Err(e) => return Err(e),
        }
    }
    Err(last_err.expect("retry_transient records an error before the budget is spent"))
}

/// Remove a file, clearing the read-only attribute first to ensure deletion
/// succeeds even on Windows when the file was marked read-only.
///
/// On Unix, when the path is one of several hardlinks (`nlink > 1`), the RO bit
/// is left alone: clearing it would make every name on the shared inode
/// writable (including a published content-addressed blob the temp still
/// shares). Unlink alone is enough — the remaining names keep their mode.
fn remove_file_robust(path: &Path) -> std::io::Result<()> {
    if let Ok(meta) = fs::metadata(path) {
        let mut perms = meta.permissions();
        if perms.readonly() {
            #[cfg(unix)]
            {
                use std::os::unix::fs::MetadataExt;
                if meta.nlink() > 1 {
                    return fs::remove_file(path);
                }
            }
            perms.set_readonly(false);
            let _ = fs::set_permissions(path, perms);
        }
    }
    fs::remove_file(path)
}

/// Perform an atomic file replacement. This creates a temporary file beside the
/// destination path, executes `write_fn` to populate it, flushes it (`fsync_file`),
/// copies permissions from the original destination (if it exists) to preserve file
/// modes, and atomic renames it with transient error retries on Windows.
///
/// Returns `Ok(true)` if the rename succeeded.
/// If `allow_concurrent_winner` is `true`, a concurrent writer winning the race to create
/// the file is treated as a benign lost race and returns `Ok(false)`. If `allow_concurrent_winner`
/// is `false`, it retries or errors out on conflicts.
pub fn atomic_write_and_replace<F>(
    dest_path: &Path,
    allow_concurrent_winner: bool,
    write_fn: F,
) -> Result<bool>
where
    F: FnOnce(&Path) -> Result<()>,
{
    atomic_write_and_replace_with(dest_path, allow_concurrent_winner, write_fn, |_| Ok(()))
}

/// Like [`atomic_write_and_replace`], but runs `after_fsync` on the temp path
/// after the durable flush and before the rename. Used by store hardlink ingest
/// to enforce the read-only guard on a shared inode only once fsync has
/// finished (Windows `FlushFileBuffers` needs a writable handle, #196).
pub fn atomic_write_and_replace_with<F, A>(
    dest_path: &Path,
    allow_concurrent_winner: bool,
    write_fn: F,
    after_fsync: A,
) -> Result<bool>
where
    F: FnOnce(&Path) -> Result<()>,
    A: FnOnce(&Path) -> Result<()>,
{
    atomic_write_and_replace_with_dir_sync(
        dest_path,
        allow_concurrent_winner,
        write_fn,
        after_fsync,
        |parent| {
            let _ = fsync_dir(parent);
            Ok(())
        },
    )
}

/// [`atomic_write_and_replace_with`] whose flush the caller may defer. With
/// `durable == false` neither the temp file nor its directory is fsynced;
/// the rename is still atomic, so a reader sees the whole file or none. The
/// store uses this for blobs it marks pending and flushes off the build path
/// (deferred durability); `after_fsync` then runs after the write instead.
pub fn atomic_write_and_replace_deferrable<F, A>(
    dest_path: &Path,
    allow_concurrent_winner: bool,
    write_fn: F,
    after_fsync: A,
    durable: bool,
) -> Result<bool>
where
    F: FnOnce(&Path) -> Result<()>,
    A: FnOnce(&Path) -> Result<()>,
{
    if durable {
        return atomic_write_and_replace_with(
            dest_path,
            allow_concurrent_winner,
            write_fn,
            after_fsync,
        );
    }
    atomic_write_and_replace_core(
        dest_path,
        allow_concurrent_winner,
        write_fn,
        after_fsync,
        false,
        |_| Ok(()),
    )
}

/// [`atomic_replace`] whose flush the caller may defer; see
/// [`atomic_write_and_replace_deferrable`].
pub fn atomic_replace_deferrable(dest_path: &Path, bytes: &[u8], durable: bool) -> Result<()> {
    atomic_write_and_replace_deferrable(
        dest_path,
        false,
        |temp_path| {
            fs::write(temp_path, bytes)?;
            Ok(())
        },
        |_| Ok(()),
        durable,
    )?;
    Ok(())
}

fn atomic_write_and_replace_with_dir_sync<F, A, D>(
    dest_path: &Path,
    allow_concurrent_winner: bool,
    write_fn: F,
    after_fsync: A,
    sync_dir: D,
) -> Result<bool>
where
    F: FnOnce(&Path) -> Result<()>,
    A: FnOnce(&Path) -> Result<()>,
    D: Fn(&Path) -> Result<()>,
{
    atomic_write_and_replace_core(
        dest_path,
        allow_concurrent_winner,
        write_fn,
        after_fsync,
        true,
        sync_dir,
    )
}

fn atomic_write_and_replace_core<F, A, D>(
    dest_path: &Path,
    allow_concurrent_winner: bool,
    write_fn: F,
    after_fsync: A,
    sync_file: bool,
    sync_dir: D,
) -> Result<bool>
where
    F: FnOnce(&Path) -> Result<()>,
    A: FnOnce(&Path) -> Result<()>,
    D: Fn(&Path) -> Result<()>,
{
    let nonce = ATOMIC_TMP_NONCE.fetch_add(1, Ordering::Relaxed);
    let pid = std::process::id();
    let file_name = dest_path
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("file");
    let temp_path = dest_path.with_file_name(format!(".{file_name}.{pid}.{nonce}.tmp"));

    let res = (|| -> Result<()> {
        write_fn(&temp_path)?;
        if sync_file {
            fsync_file(&temp_path).context("flushing temp file")?;
        }
        after_fsync(&temp_path)?;
        Ok(())
    })();

    if let Err(e) = res {
        let _ = remove_file_robust(&temp_path);
        return Err(e);
    }

    // Preserve the destination's permissions so atomic replacement does not widen them.
    // Setting permissions is best-effort (e.g. read-only mounts or filesystem limitations).
    if let Ok(meta) = fs::metadata(dest_path) {
        let _ = fs::set_permissions(&temp_path, meta.permissions());
    }

    const MAX_ATTEMPTS: u32 = 10;
    let mut last_err = None;
    for attempt in 0..MAX_ATTEMPTS {
        match fs::rename(&temp_path, dest_path) {
            Ok(()) => {
                if let Some(parent) = dest_path.parent() {
                    sync_dir(parent)?;
                }
                return Ok(true);
            }
            Err(e) => {
                if allow_concurrent_winner && dest_path.is_file() {
                    let _ = remove_file_robust(&temp_path);
                    if let Some(parent) = dest_path.parent() {
                        sync_dir(parent)?;
                    }
                    return Ok(false);
                }
                if dest_path.is_dir() || !is_transient_rename_error(&e) {
                    let _ = remove_file_robust(&temp_path);
                    return Err(e).context("atomic rename");
                }
                last_err = Some(e);
                std::thread::sleep(std::time::Duration::from_millis(rename_backoff_ms(attempt)));
            }
        }
    }

    let _ = remove_file_robust(&temp_path);
    if allow_concurrent_winner && dest_path.is_file() {
        if let Some(parent) = dest_path.parent() {
            sync_dir(parent)?;
        }
        return Ok(false);
    }
    let err = last_err.unwrap_or_else(|| std::io::Error::other("atomic rename failed"));
    Err(anyhow::Error::new(err).context("atomic rename failed after retries"))
}

pub fn atomic_replace(dest_path: &Path, bytes: &[u8]) -> Result<()> {
    atomic_write_and_replace(dest_path, false, |temp_path| {
        fs::write(temp_path, bytes)?;
        Ok(())
    })?;
    Ok(())
}

pub fn cleanup_temp_files(dir: &Path, prefix: &str, min_age: std::time::Duration) -> Result<()> {
    if let Ok(entries) = fs::read_dir(dir) {
        let now = std::time::SystemTime::now();
        let expected_prefix = format!(".{prefix}.");
        for entry in entries.flatten() {
            let path = entry.path();
            if let Some(name) = path.file_name().and_then(|n| n.to_str())
                && name.starts_with(&expected_prefix)
                && name.ends_with(".tmp")
                && let Ok(meta) = entry.metadata()
                && let Ok(modified) = meta.modified()
                && let Ok(age) = now.duration_since(modified)
                && age > min_age
            {
                let _ = remove_file_robust(&path);
            }
        }
    }
    Ok(())
}

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

    #[test]
    fn rename_backoff_ms_caps_at_64() {
        assert_eq!(rename_backoff_ms(0), 1);
        assert_eq!(rename_backoff_ms(1), 2);
        assert_eq!(rename_backoff_ms(5), 32);
        assert_eq!(rename_backoff_ms(6), 64);
        assert_eq!(rename_backoff_ms(9), 64);
    }

    #[test]
    fn is_transient_rename_error_only_flags_windows_race_codes() {
        use std::io::Error;
        let access_denied = Error::from_raw_os_error(5);
        let sharing_violation = Error::from_raw_os_error(32);
        let other = Error::from_raw_os_error(2);
        #[cfg(windows)]
        {
            assert!(is_transient_rename_error(&access_denied));
            assert!(is_transient_rename_error(&sharing_violation));
            assert!(!is_transient_rename_error(&other));
        }
        #[cfg(not(windows))]
        {
            assert!(!is_transient_rename_error(&access_denied));
            assert!(!is_transient_rename_error(&sharing_violation));
            assert!(!is_transient_rename_error(&other));
        }
    }

    /// A success and a non-transient failure both settle on the first call.
    ///
    /// Off Windows this is the whole behaviour, since no error classifies as
    /// transient there: the retry must never turn a real fault into a delay.
    #[test]
    fn retry_transient_does_not_retry_a_settled_outcome() {
        let mut calls = 0;
        let value = retry_transient(
            || {
                calls += 1;
                Ok::<_, std::io::Error>(7)
            },
            is_transient_rename_error,
        )
        .unwrap();
        assert_eq!((value, calls), (7, 1));

        let mut calls = 0;
        let err = retry_transient(
            || {
                calls += 1;
                Err::<(), _>(std::io::Error::from_raw_os_error(2))
            },
            is_transient_rename_error,
        )
        .unwrap_err();
        assert_eq!(err.raw_os_error(), Some(2));
        assert_eq!(calls, 1, "a non-transient error must not be retried");
    }

    /// A delete-pending destination clears on its own, so the retry has to
    /// outlast it and then return the eventual success.
    ///
    /// Drives [`retry_transient`] with an always-transient classifier: the
    /// real one is a compile-time `false` off Windows, which would
    /// leave this asserting nothing on most runners.
    #[test]
    fn retry_transient_waits_out_a_transient_failure() {
        let mut calls = 0;
        let value = retry_transient(
            || {
                calls += 1;
                if calls < 3 {
                    return Err(std::io::Error::from_raw_os_error(5));
                }
                Ok(calls)
            },
            |_| true,
        )
        .unwrap();
        assert_eq!((value, calls), (3, 3));
    }

    /// A state that never clears is a real failure, and the budget is what
    /// keeps it from hanging: the last attempt's error surfaces, after exactly
    /// the budgeted number of tries. The exact count is the assertion that
    /// pins the bound — off by one either way and this fails.
    #[test]
    fn retry_transient_gives_up_after_exactly_its_budget() {
        let mut calls: u32 = 0;
        let err = retry_transient(
            || {
                calls += 1;
                Err::<(), _>(std::io::Error::from_raw_os_error(32))
            },
            |_| true,
        )
        .unwrap_err();
        assert_eq!(err.raw_os_error(), Some(32));
        assert_eq!(calls, TRANSIENT_ATTEMPTS);
    }

    #[test]
    fn should_retry_transient_requires_a_transient_error_inside_the_budget() {
        assert!(should_retry_transient(true, 0));
        assert!(should_retry_transient(true, TRANSIENT_ATTEMPTS - 2));
        assert!(!should_retry_transient(true, TRANSIENT_ATTEMPTS - 1));
        assert!(!should_retry_transient(false, 0));
        assert!(!should_retry_transient(false, TRANSIENT_ATTEMPTS - 2));
    }

    /// A classifier that rejects the error settles on the first call even when
    /// the error carries a Windows race code. This is the whole behaviour off
    /// Windows, where nothing classifies as transient.
    #[test]
    fn retry_transient_respects_a_rejecting_classifier() {
        let mut calls = 0;
        let err = retry_transient(
            || {
                calls += 1;
                Err::<(), _>(std::io::Error::from_raw_os_error(5))
            },
            |_| false,
        )
        .unwrap_err();
        assert_eq!(err.raw_os_error(), Some(5));
        assert_eq!(calls, 1);
    }

    /// With the real classifier it retries a Windows race code there and
    /// settles at once everywhere else.
    #[test]
    fn retry_transient_with_the_real_classifier_retries_only_on_windows() {
        let mut calls: u32 = 0;
        let err = retry_transient(
            || {
                calls += 1;
                Err::<(), _>(std::io::Error::from_raw_os_error(5))
            },
            is_transient_rename_error,
        )
        .unwrap_err();
        assert_eq!(err.raw_os_error(), Some(5));
        #[cfg(windows)]
        assert_eq!(calls, TRANSIENT_ATTEMPTS);
        #[cfg(not(windows))]
        assert_eq!(calls, 1);
    }

    #[test]
    fn test_atomic_replace_writes_content_and_syncs() {
        let dir = tempfile::tempdir().unwrap();
        let dest = dir.path().join("dest.txt");

        atomic_replace(&dest, b"hello world").unwrap();
        assert_eq!(fs::read(&dest).unwrap(), b"hello world");
    }

    #[test]
    fn durable_replace_syncs_parent_after_publication() {
        let dir = tempfile::tempdir().unwrap();
        let dest = dir.path().join("intent.json");
        let observed = std::cell::Cell::new(false);

        atomic_write_and_replace_with_dir_sync(
            &dest,
            false,
            |temp| {
                fs::write(temp, b"intent")?;
                Ok(())
            },
            |_| Ok(()),
            |parent| {
                assert_eq!(parent, dir.path());
                assert_eq!(fs::read(&dest).unwrap(), b"intent");
                observed.set(true);
                Ok(())
            },
        )
        .unwrap();

        assert!(observed.get(), "directory sync must follow the rename");
    }

    #[test]
    fn durable_replace_propagates_directory_sync_failure() {
        let dir = tempfile::tempdir().unwrap();
        let dest = dir.path().join("intent.json");

        let error = atomic_write_and_replace_with_dir_sync(
            &dest,
            false,
            |temp| {
                fs::write(temp, b"intent")?;
                Ok(())
            },
            |_| Ok(()),
            |_| {
                Err(anyhow::Error::new(std::io::Error::new(
                    std::io::ErrorKind::PermissionDenied,
                    "injected directory fsync failure",
                )))
            },
        )
        .unwrap_err();

        assert!(
            format!("{error:#}").contains("injected directory fsync failure"),
            "unexpected error: {error:#}"
        );
        assert_eq!(
            fs::read(&dest).unwrap(),
            b"intent",
            "rename happens before the injected durability failure"
        );
    }

    #[test]
    #[cfg(unix)]
    fn test_atomic_replace_preserves_permissions() {
        use std::os::unix::fs::PermissionsExt;
        let dir = tempfile::tempdir().unwrap();
        let dest = dir.path().join("dest.txt");

        // Write initial file and set permissions to 0o600
        fs::write(&dest, b"initial").unwrap();
        fs::set_permissions(&dest, fs::Permissions::from_mode(0o600)).unwrap();

        // Do atomic replacement
        atomic_replace(&dest, b"new content").unwrap();

        // Verify content and that mode 0o600 was preserved
        assert_eq!(fs::read(&dest).unwrap(), b"new content");
        let meta = fs::metadata(&dest).unwrap();
        assert_eq!(meta.permissions().mode() & 0o777, 0o600);
    }

    #[test]
    fn test_cleanup_temp_files_only_removes_matching_stale_temps() {
        let dir = tempfile::tempdir().unwrap();

        // Pattern: .<prefix>.<pid>.<nonce>.tmp
        let temp_live = dir.path().join(".testfile.1234.0.tmp");
        let temp_stale = dir.path().join(".testfile.5678.1.tmp");
        let other_file = dir.path().join(".otherfile.1234.0.tmp");
        let unrelated = dir.path().join("unrelated.txt");

        fs::write(&temp_live, b"live temp").unwrap();
        fs::write(&temp_stale, b"stale temp").unwrap();
        fs::write(&other_file, b"other temp").unwrap();
        fs::write(&unrelated, b"unrelated").unwrap();

        // Adjust mtimes so only temp_stale and other_file are older than 5 minutes
        let stale_time = std::time::SystemTime::now() - std::time::Duration::from_secs(400);
        filetime::set_file_mtime(
            &temp_stale,
            filetime::FileTime::from_system_time(stale_time),
        )
        .unwrap();
        filetime::set_file_mtime(
            &other_file,
            filetime::FileTime::from_system_time(stale_time),
        )
        .unwrap();

        // Run cleanup with prefix "testfile" and min_age 5 minutes (300 secs)
        cleanup_temp_files(dir.path(), "testfile", std::time::Duration::from_secs(300)).unwrap();

        // stale temp must be deleted
        assert!(!temp_stale.exists());
        // live temp must still exist (not old enough)
        assert!(temp_live.exists());
        // other temp must still exist (prefix is different)
        assert!(other_file.exists());
        // unrelated file must still exist
        assert!(unrelated.exists());
    }

    // Regression guard for #196: `fsync_file` must durably flush a freshly
    // written (writable) file. The bug was a read-only handle — fine for Unix
    // `fsync(2)`, but Windows `FlushFileBuffers` rejects it with
    // ERROR_ACCESS_DENIED, so every blob store failed ("flushing blob to disk").
    // Passes on Unix before and after; the real failing→passing signal is on
    // Windows, where the old read-only-handle form errored here.
    #[test]
    fn fsync_file_flushes_a_writable_blob() {
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.bin");
        fs::write(&blob, b"some bytes").unwrap();
        fsync_file(&blob).expect("fsync of a writable file must succeed on all platforms");
    }

    #[test]
    fn fsync_dir_succeeds_on_a_real_directory() {
        // On Unix this exercises the open+sync_all path; elsewhere it's a no-op.
        let dir = tempfile::tempdir().unwrap();
        fsync_dir(dir.path()).expect("fsync of a directory must not error");
    }
}