agent_first_data/

atomic_file.rs

//! One atomic file installation, shared by every writer in this crate.
//!
//! Installing a file so that a crash can never expose a half-written one is a
//! fixed sequence — private same-directory temporary file, write, permissions,
//! fsync the file, atomically install it over the target, fsync the directory
//! entry — and each step has a way to be got subtly wrong. Writing that
//! sequence a second time somewhere else is how one copy ends up durable and
//! the other only looks it: a `rename` that is never followed by a parent
//! directory fsync returns success while the new directory entry may still be
//! lost to a power cut.
//!
//! So the sequence lives here once, and callers bring only what is genuinely
//! theirs: the pre-write guard (what may be overwritten is a policy question,
//! and different writers answer it differently) and the error type. Failures
//! name the step that failed through [`AtomicError`], whose `Display` is the
//! tail of a caller's own message — `write: ` + `install \`p\`: …`.

use std::fs::{self, File, OpenOptions};
use std::io::Write as _;
use std::path::{Path, PathBuf};

/// Atomically replace a regular file, preserving its permissions. On Unix a
/// new file starts owner-only (`0600`); Windows has no mode bits, so a new file
/// takes the directory's inherited ACL. Symbolic links and special files are refused; parent
/// directories must already exist. A directory-sync error reports an uncertain
/// durable commit through [`AtomicError::commit_uncertain`].
///
/// Hard links are not checked. The new file is renamed over the path, so a
/// target with other hard links is split from them: this path gets the new
/// bytes and every other link keeps the old file. A caller that needs a
/// hard-linked target refused writes through
/// [`DocumentFile`](crate::document::DocumentFile) instead, whose
/// [`ensure_mutable`](crate::document::DocumentFile::ensure_mutable) guard (run
/// by `save` and by a replacing `create_atomic`) refuses one on Unix.
pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), AtomicError> {
    let permissions = match fs::symlink_metadata(path) {
        Ok(metadata) if metadata.file_type().is_file() => Some(metadata.permissions()),
        Ok(_) => {
            return Err(AtomicError::at(
                AtomicStep::Guard,
                path,
                std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    "atomic target must be a regular file, not a symlink, directory, or special file",
                ),
            ));
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
        Err(error) => return Err(AtomicError::at(AtomicStep::Guard, path, error)),
    };
    install(
        path,
        AtomicInstall::replacing(bytes).with_permissions(permissions),
    )
}

/// How the finished temporary file becomes the target.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InstallMode {
    /// `rename` over the target. Replaces whatever is at the path — including a
    /// symbolic link, which is replaced rather than followed.
    Replace,
    /// `hard_link`, which refuses to install over an existing target. The
    /// caller distinguishes that refusal by [`AtomicError::target_exists`].
    NewOnly,
}

/// What to install, and how.
pub(crate) struct AtomicInstall<'a> {
    /// The complete new contents.
    pub bytes: &'a [u8],
    /// How to install the finished file.
    pub mode: InstallMode,
    /// Permissions to re-apply, normally the replaced file's own.
    pub permissions: Option<fs::Permissions>,
    /// Explicit unix permission bits, applied after `permissions`.
    pub unix_mode: Option<u32>,
}

impl<'a> AtomicInstall<'a> {
    /// Replace whatever is at the path, leaving permissions to the platform.
    pub(crate) fn replacing(bytes: &'a [u8]) -> Self {
        Self {
            bytes,
            mode: InstallMode::Replace,
            permissions: None,
            unix_mode: None,
        }
    }

    /// Re-apply these permissions to the new file.
    pub(crate) fn with_permissions(mut self, permissions: Option<fs::Permissions>) -> Self {
        self.permissions = permissions;
        self
    }

    /// Apply these unix permission bits to the new file.
    pub(crate) fn with_unix_mode(mut self, unix_mode: Option<u32>) -> Self {
        self.unix_mode = unix_mode;
        self
    }

    /// Install with `hard_link`, refusing an existing target.
    pub(crate) fn new_only(mut self) -> Self {
        self.mode = InstallMode::NewOnly;
        self
    }
}

/// The step of an atomic installation that failed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AtomicStep {
    /// Inspecting the target without following links.
    Guard,
    /// The target has no parent directory to write the temporary file into.
    NoParent,
    /// The target's own file name is not valid UTF-8.
    NonUtf8Name,
    /// Creating the private temporary file.
    Temp,
    /// Every candidate temporary name was taken.
    TempExhausted,
    /// Writing the new bytes.
    Write,
    /// Re-applying the replaced file's permissions.
    PreservePermissions,
    /// Applying explicitly requested permission bits. Unix only, because the
    /// bits themselves are: the step does not exist on a platform that has no
    /// mode to set, and a variant no code can construct is a step a reader
    /// would look for and never find.
    #[cfg(unix)]
    SetPermissions,
    /// Flushing the temporary file to storage.
    Sync,
    /// Installing the temporary file over the target.
    Install,
    /// Recording the new directory entry.
    SyncParent,
}

/// A failed atomic installation: which step, which path, and the OS error.
#[derive(Debug)]
pub struct AtomicError {
    /// The step that failed.
    pub step: AtomicStep,
    /// The path that step was about — the target, its temporary file, or the
    /// parent directory, depending on the step.
    pub path: PathBuf,
    /// The underlying OS error, absent only where the step had none.
    pub source: Option<std::io::Error>,
}

impl AtomicError {
    fn at(step: AtomicStep, path: &Path, source: std::io::Error) -> Self {
        Self {
            step,
            path: path.to_path_buf(),
            source: Some(source),
        }
    }

    /// Whether this is [`InstallMode::NewOnly`] refusing an existing target,
    /// which is a caller-visible outcome rather than an I/O fault. Crate-only:
    /// the public [`write_atomic`] always replaces, so it could never be true
    /// there.
    pub(crate) fn target_exists(&self) -> bool {
        self.step == AtomicStep::Install
            && self
                .source
                .as_ref()
                .is_some_and(|error| error.kind() == std::io::ErrorKind::AlreadyExists)
    }

    /// Whether the target may already carry the new bytes.
    ///
    /// Only the final directory fsync can fail after the installation itself
    /// succeeded: the file is complete and in place, but its survival across a
    /// crash is unconfirmed. Every other step fails before the install and
    /// leaves the original untouched.
    pub fn commit_uncertain(&self) -> bool {
        self.step == AtomicStep::SyncParent
    }
}

impl std::fmt::Display for AtomicError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let path = self.path.display();
        let phrase = match self.step {
            AtomicStep::Guard => "inspect",
            AtomicStep::NoParent => {
                return write!(formatter, "has no parent directory for `{path}`");
            }
            AtomicStep::NonUtf8Name => {
                return write!(formatter, "path is not valid UTF-8: `{path}`");
            }
            AtomicStep::TempExhausted => {
                return write!(formatter, "could not allocate temporary file in `{path}`");
            }
            AtomicStep::Temp => "temporary file in",
            AtomicStep::Write => "write",
            AtomicStep::PreservePermissions => "preserve permissions",
            #[cfg(unix)]
            AtomicStep::SetPermissions => "set permissions on",
            AtomicStep::Sync => "fsync",
            AtomicStep::Install => "install",
            AtomicStep::SyncParent => "fsync parent directory",
        };
        write!(formatter, "{phrase} `{path}`")?;
        match &self.source {
            Some(source) => write!(formatter, ": {source}"),
            None => Ok(()),
        }
    }
}

/// The smallest `NAME_MAX` across the platforms this runs on.
const MAX_NAME_BYTES: usize = 255;

/// Compose a temporary file name that stays within one path component's limit.
///
/// The marker, pid, and attempt add about 35 bytes. A target whose own name is
/// close to `NAME_MAX` would push the composed name past it, and the failure
/// would surface at install time — after the caller had already done its work,
/// on a file it opened successfully. Truncating the stem keeps the write
/// possible; uniqueness still comes from the pid and attempt, with `create_new`
/// retrying the rare collision.
fn temp_file_name(file_name: &str, pid: u32, attempt: u32) -> String {
    let suffix = format!(".afdata.{pid}.{attempt}.tmp");
    // One byte for the leading dot that hides the temporary file.
    let budget = MAX_NAME_BYTES.saturating_sub(suffix.len() + 1);
    let stem = &file_name[..file_name.floor_char_boundary(budget)];
    format!(".{stem}{suffix}")
}

/// Split a target into the directory the temporary file must share with it and
/// the target's own name. Sharing the directory is what makes the final install
/// a rename within one filesystem, which is the atomic operation this relies on.
fn parent_and_name(path: &Path) -> Result<(&Path, String), AtomicError> {
    let parent = match path.parent() {
        Some(parent) if parent.as_os_str().is_empty() => Path::new("."),
        Some(parent) => parent,
        None => {
            return Err(AtomicError {
                step: AtomicStep::NoParent,
                path: path.to_path_buf(),
                source: None,
            });
        }
    };
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .ok_or_else(|| AtomicError {
            step: AtomicStep::NonUtf8Name,
            path: path.to_path_buf(),
            source: None,
        })?
        .to_string();
    Ok((parent, file_name))
}

fn allocate_private_temp(parent: &Path, file_name: &str) -> Result<(PathBuf, File), AtomicError> {
    let pid = std::process::id();
    for attempt in 0..32_u32 {
        let candidate = parent.join(temp_file_name(file_name, pid, attempt));
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt as _;
            // Nothing may read the new contents through the temporary name
            // before the caller's own permissions are applied below.
            options.mode(0o600);
        }
        match options.open(&candidate) {
            Ok(file) => return Ok((candidate, file)),
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(error) => return Err(AtomicError::at(AtomicStep::Temp, parent, error)),
        }
    }
    Err(AtomicError {
        step: AtomicStep::TempExhausted,
        path: parent.to_path_buf(),
        source: None,
    })
}

#[cfg(unix)]
fn sync_parent(parent: &Path) -> Result<(), AtomicError> {
    File::open(parent)
        .and_then(|directory| directory.sync_all())
        .map_err(|error| AtomicError::at(AtomicStep::SyncParent, parent, error))
}

#[cfg(not(unix))]
fn sync_parent(_parent: &Path) -> Result<(), AtomicError> {
    // Rust does not expose a portable directory handle on non-unix targets.
    // The file itself is still synced before its atomic installation.
    Ok(())
}

fn write_temp(
    mut temp_file: File,
    temp_path: &Path,
    target: &Path,
    request: &AtomicInstall<'_>,
    write: impl FnOnce(&mut File, &[u8]) -> std::io::Result<()>,
) -> Result<(), AtomicError> {
    write(&mut temp_file, request.bytes)
        .map_err(|error| AtomicError::at(AtomicStep::Write, target, error))?;
    if let Some(permissions) = request.permissions.clone() {
        temp_file
            .set_permissions(permissions)
            .map_err(|error| AtomicError::at(AtomicStep::PreservePermissions, target, error))?;
    }
    #[cfg(unix)]
    if let Some(unix_mode) = request.unix_mode {
        use std::os::unix::fs::PermissionsExt as _;
        temp_file
            .set_permissions(fs::Permissions::from_mode(unix_mode))
            .map_err(|error| AtomicError::at(AtomicStep::SetPermissions, target, error))?;
    }
    temp_file
        .sync_all()
        .map_err(|error| AtomicError::at(AtomicStep::Sync, temp_path, error))
}

/// Install `request` at `path` atomically.
///
/// The caller has already decided that this target may be written; this only
/// performs the write. Every failure before the install leaves the original
/// untouched and removes the temporary file. A failure at the final parent
/// directory fsync — the one [`AtomicError::commit_uncertain`] reports — means
/// the new file may already be in place without confirmed durability.
pub(crate) fn install(path: &Path, request: AtomicInstall<'_>) -> Result<(), AtomicError> {
    install_with_writer(path, request, |file, bytes| file.write_all(bytes))
}

// The writer seam keeps disk-full, short-write, and interrupted-process
// regressions at the primitive that owns publication, without exposing a
// callback or partial-file API to callers.
fn install_with_writer(
    path: &Path,
    request: AtomicInstall<'_>,
    write: impl FnOnce(&mut File, &[u8]) -> std::io::Result<()>,
) -> Result<(), AtomicError> {
    let (parent, file_name) = parent_and_name(path)?;
    let (temp_path, temp_file) = allocate_private_temp(parent, &file_name)?;
    let result = (|| -> Result<(), AtomicError> {
        write_temp(temp_file, &temp_path, path, &request, write)?;
        match request.mode {
            InstallMode::Replace => {
                fs::rename(&temp_path, path)
                    .map_err(|error| AtomicError::at(AtomicStep::Install, path, error))?;
            }
            InstallMode::NewOnly => {
                fs::hard_link(&temp_path, path)
                    .map_err(|error| AtomicError::at(AtomicStep::Install, path, error))?;
                // The file is installed from here on. Failing to unlink the
                // temporary link leaves a stray file, but reporting an error
                // would tell the caller the commit did not happen — and a retry
                // would then refuse a target that this call in fact created.
                let _ = fs::remove_file(&temp_path);
            }
        }
        sync_parent(parent)?;
        Ok(())
    })();
    if let Err(error) = &result {
        // A parent-directory fsync failure comes after the install, where the
        // temporary name no longer refers to anything to clean up.
        if !error.commit_uncertain() {
            let _ = fs::remove_file(&temp_path);
        }
    }
    result
}

impl std::error::Error for AtomicError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        self.source.as_ref().map(|error| error as _)
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::panic, clippy::expect_used)]
    use super::*;

    #[test]
    fn temp_name_stays_within_one_path_component() {
        let long = "x".repeat(400);
        let name = temp_file_name(&long, 1234, 0);
        assert!(name.len() <= MAX_NAME_BYTES, "{} bytes", name.len());
        assert!(name.starts_with('.'));
        assert!(name.ends_with(".tmp"));
    }

    #[test]
    fn temp_name_truncates_on_a_character_boundary() {
        let long = "é".repeat(300);
        let name = temp_file_name(&long, 1234, 0);
        assert!(name.is_char_boundary(name.len()));
        assert!(name.len() <= MAX_NAME_BYTES);
    }

    #[test]
    fn bare_relative_paths_use_the_current_directory() {
        let (parent, file_name) = parent_and_name(Path::new("config.json")).unwrap();

        assert_eq!(parent, Path::new("."));
        assert_eq!(file_name, "config.json");
    }

    #[test]
    fn replace_installs_new_bytes_and_leaves_no_temporary_file() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("config.json");
        fs::write(&path, b"old").unwrap();

        install(&path, AtomicInstall::replacing(b"new")).unwrap();

        assert_eq!(fs::read(&path).unwrap(), b"new");
        let strays: Vec<_> = fs::read_dir(dir.path())
            .unwrap()
            .filter_map(Result::ok)
            .filter(|entry| entry.file_name().to_string_lossy().ends_with(".tmp"))
            .collect();
        assert!(strays.is_empty(), "temporary files left behind");
    }

    #[test]
    fn new_only_refuses_an_existing_target_without_touching_it() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("config.json");
        fs::write(&path, b"original").unwrap();

        let error = install(&path, AtomicInstall::replacing(b"new").new_only()).unwrap_err();

        assert!(error.target_exists());
        assert_eq!(fs::read(&path).unwrap(), b"original");
    }

    #[cfg(unix)]
    #[test]
    fn replace_swaps_a_symlink_rather_than_following_it() {
        use std::os::unix::fs::symlink;

        let dir = tempfile::tempdir().unwrap();
        let outside = dir.path().join("outside.txt");
        fs::write(&outside, b"untouched").unwrap();
        let link = dir.path().join("link.txt");
        symlink(&outside, &link).unwrap();

        install(&link, AtomicInstall::replacing(b"new")).unwrap();

        assert_eq!(fs::read(&outside).unwrap(), b"untouched");
        assert!(
            !fs::symlink_metadata(&link)
                .unwrap()
                .file_type()
                .is_symlink()
        );
        assert_eq!(fs::read(&link).unwrap(), b"new");
    }

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

        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("readable.md");

        install(
            &path,
            AtomicInstall::replacing(b"body").with_unix_mode(Some(0o644)),
        )
        .unwrap();

        let mode = fs::metadata(&path).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o644);
    }

    #[test]
    fn a_missing_parent_directory_fails_before_any_write() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("absent").join("config.json");

        let error = install(&path, AtomicInstall::replacing(b"new")).unwrap_err();

        assert_eq!(error.step, AtomicStep::Temp);
        assert!(!error.commit_uncertain());
        assert!(!path.exists());
    }
    #[test]
    fn failed_writes_preserve_previous_bytes_and_remove_temporary_file() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.json");
        fs::write(&path, b"old complete value").unwrap();
        for error in [
            std::io::Error::from_raw_os_error(28),
            std::io::Error::from(std::io::ErrorKind::Interrupted),
            std::io::Error::from(std::io::ErrorKind::WriteZero),
        ] {
            let error = install_with_writer(&path, AtomicInstall::replacing(b"new"), |file, _| {
                file.write_all(b"partial")?;
                Err(error)
            })
            .unwrap_err();
            assert_eq!(error.step, AtomicStep::Write);
            assert!(!error.commit_uncertain());
            assert_eq!(fs::read(&path).unwrap(), b"old complete value");
            assert_eq!(fs::read_dir(dir.path()).unwrap().count(), 1);
        }
    }

    #[test]
    fn write_all_retries_short_writes_and_interruptions_before_publish() {
        struct ShortWriter<'a> {
            file: &'a mut File,
            interrupted: bool,
        }
        impl std::io::Write for ShortWriter<'_> {
            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
                if !self.interrupted {
                    self.interrupted = true;
                    return Err(std::io::ErrorKind::Interrupted.into());
                }
                self.file.write(&bytes[..bytes.len().min(2)])
            }
            fn flush(&mut self) -> std::io::Result<()> {
                self.file.flush()
            }
        }
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.json");
        let bytes = b"new complete value";
        install_with_writer(&path, AtomicInstall::replacing(bytes), |file, bytes| {
            ShortWriter {
                file,
                interrupted: false,
            }
            .write_all(bytes)
        })
        .unwrap();
        assert_eq!(fs::read(&path).unwrap(), bytes);
    }

    #[test]
    fn interrupted_process_leaves_uncommitted_temporary_file_not_target() {
        const CHILD_PATH: &str = "AFDATA_TEST_INTERRUPTED_ATOMIC_PATH";
        if let Some(path) = std::env::var_os(CHILD_PATH) {
            let _ = install_with_writer(
                Path::new(&path),
                AtomicInstall::replacing(b"new"),
                |file, _| {
                    file.write_all(b"partial")?;
                    file.sync_all()?;
                    std::process::exit(86);
                },
            );
            panic!("child did not reach the interrupted write");
        }
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.json");
        write_atomic(&path, b"old complete value").unwrap();
        let status = std::process::Command::new(std::env::current_exe().unwrap())
            .args(["--exact", "atomic_file::tests::interrupted_process_leaves_uncommitted_temporary_file_not_target"])
            .env(CHILD_PATH, &path).status().unwrap();
        assert_eq!(status.code(), Some(86));
        assert_eq!(fs::read(&path).unwrap(), b"old complete value");
        assert_eq!(fs::read_dir(dir.path()).unwrap().count(), 2);
        write_atomic(&path, b"new complete value").unwrap();
        assert_eq!(fs::read(&path).unwrap(), b"new complete value");
    }
}