wsb-daemon 0.258.6441

//! Applying what a peer sent, transactionally.
//!
//! A transfer lands in three steps, and the order is the whole point:
//!
//! 1. **Stage.** Every incoming file is written to the machine's staging
//!    directory first, and verified against the hash it was sent with. A file
//!    is never streamed over the working copy of itself — an interrupted
//!    stream would leave a half-file that looks like an edit, and the next
//!    scan would carry that half-file back to the peer as work.
//! 2. **Check.** The git epoch is re-validated against the repository as it is
//!    NOW. A checkout that started while the bytes were in flight invalidates
//!    the whole transfer, and none of it is applied.
//! 3. **Apply.** Each staged file is renamed into place. A rename is atomic on
//!    every filesystem this runs on, so a reader sees the old file or the new
//!    one and never a partial one. A path that no longer holds what the plan
//!    found there was edited after the plan read it, and is left alone.
//!
//! Only what actually landed is reported back, so the sender advances its
//! baseline over exactly that and retries the rest.

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use wsb::peer::compatibility::{Epoch, RepositoryFacts};
use wsb::peer::converge::{Entry, EntryKind};
use wsb::peer::model::RepositoryId;
use wsb::peer::protocol::Write;
use wsb::peer::scan;

/// Where one transfer's files are staged.
///
/// Under the machine directory, never inside the workspace: a partial file
/// inside a working tree would be seen by a build, an editor and this daemon's
/// own watcher.
pub struct Staging {
    dir: PathBuf,
}

impl Staging {
    pub fn create(root: &Path, token: &str) -> Result<Self, String> {
        let dir = root.join(token);
        // Remove first: a directory left by a daemon that crashed mid-transfer
        // holds files nobody acknowledged, and reusing them would apply a
        // transfer whose sender may have moved on.
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir)
            .map_err(|e| format!("cannot create the staging directory {}: {e}", dir.display()))?;
        Ok(Self { dir })
    }

    /// Write one incoming file to staging, verifying it as it lands.
    pub fn stage(&self, index: usize, write: &Write) -> Result<PathBuf, String> {
        // Verified here, before anything else looks at it: the content must be
        // what its entry says, or the baseline would record a hash nothing on
        // disk has.
        let bytes = write.decoded()?;
        let staged = self.dir.join(format!("{index}.part"));
        std::fs::write(&staged, &bytes)
            .map_err(|e| format!("cannot stage {}: {e}", write.path))?;
        Ok(staged)
    }

}

impl Drop for Staging {
    /// Staging is per transfer. Whatever happened, its directory goes.
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.dir);
    }
}

/// Apply a verified transfer to a working tree.
///
/// `found` is what the plan read in this working tree at each path the
/// transfer touches; a path it found nothing at is absent.
///
/// Returns the paths that actually landed. A path that could not be applied is
/// simply absent from the result: the sender's baseline does not move for it,
/// so the next round tries again rather than losing the change. So is a path
/// whose file changed after the plan read it: the next round compares that
/// edit too, and two edits of one file are a conflict, not a replacement.
#[allow(clippy::too_many_arguments)]
pub fn apply(
    workdir: &Path,
    repository: &RepositoryId,
    git_dir: &Path,
    epoch: &Epoch,
    writes: &[Write],
    deletions: &[String],
    found: &BTreeMap<String, Entry>,
    staging_root: &Path,
) -> Result<Vec<String>, ApplyRefusal> {
    // Every path is resolved before anything is written, so a transfer
    // carrying one forbidden name does not half-apply before it is noticed.
    let mut targets: Vec<(usize, &Write, PathBuf)> = Vec::new();
    for (index, write) in writes.iter().enumerate() {
        let target = scan::local_path(workdir, &write.path)
            .map_err(|e| ApplyRefusal::Forbidden(e))?;
        targets.push((index, write, target));
    }
    let mut removals: Vec<(String, PathBuf)> = Vec::new();
    for path in deletions {
        let target = scan::local_path(workdir, path).map_err(ApplyRefusal::Forbidden)?;
        removals.push((path.clone(), target));
    }

    let token = format!(
        "{}-{}",
        repository.as_str().replace('/', "_"),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or_default()
    );
    let staging = Staging::create(staging_root, &token).map_err(ApplyRefusal::Internal)?;
    let mut staged = Vec::new();
    for (index, write, target) in &targets {
        let path = staging.stage(*index, write).map_err(ApplyRefusal::Malformed)?;
        staged.push((write, target, path));
    }

    // Re-checked with every byte already on disk and nothing yet applied. This
    // is the moment a checkout that started during the transfer is caught.
    let now = RepositoryFacts::read(repository.clone(), workdir, git_dir)
        .map_err(ApplyRefusal::Internal)?;
    if !epoch.still_current(&now) {
        return Err(ApplyRefusal::EpochMoved);
    }

    let mut applied = Vec::new();
    for (write, target, source) in staged {
        if !as_planned(repository, &write.path, target, found) {
            continue;
        }
        if let Some(parent) = target.parent() {
            if std::fs::create_dir_all(parent).is_err() {
                continue;
            }
        }
        let landed = match write.entry.kind {
            EntryKind::File => place_file(&source, target, write.entry.executable),
            EntryKind::Symlink => place_symlink(&source, target),
        };
        if landed.is_ok() {
            applied.push(write.path.clone());
        }
    }

    for (wire, target) in removals {
        match std::fs::symlink_metadata(&target) {
            // Already gone is the outcome asked for.
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => applied.push(wire),
            Err(_) => {}
            Ok(meta) => {
                let removed = if meta.is_dir() {
                    // A directory where a file was is not this transfer's to
                    // remove: the peer deleted a FILE, and a tree here with the
                    // same name is something else entirely.
                    Err(())
                } else if !as_planned(repository, &wire, &target, found) {
                    // The peer deleted the version both machines had agreed
                    // on, and this one is newer.
                    Err(())
                } else {
                    std::fs::remove_file(&target).map_err(|_| ())
                };
                if removed.is_ok() {
                    prune_empty_parents(workdir, &target);
                    // Said out loud, every time. A file that leaves this
                    // machine because another machine deleted it is the one
                    // thing continuity does that a person cannot undo and
                    // cannot see: it used to happen in silence, so "my new
                    // file disappeared" had no evidence anywhere on either
                    // machine. Writes are many and recoverable; removals are
                    // rare and are not.
                    tracing::info!(
                        repository = %repository,
                        path = %wire,
                        "removed a file because a paired machine deleted it"
                    );
                    applied.push(wire);
                }
            }
        }
    }

    Ok(applied)
}

/// Whether a path still holds what the plan found there, read the moment
/// before it is replaced or removed.
///
/// A scan takes seconds, and a person goes on working while it runs: a file
/// saved after this machine was scanned is an edit the plan never saw, and
/// replacing it with the peer's version destroyed it without a conflict ever
/// being recorded. A path that cannot be read is not as planned either.
fn as_planned(
    repository: &RepositoryId,
    path: &str,
    target: &Path,
    found: &BTreeMap<String, Entry>,
) -> bool {
    let now = match Entry::read(target) {
        Ok(now) => now,
        Err(error) => {
            tracing::debug!(%repository, %path, %error, "left alone: it cannot be read");
            return false;
        }
    };
    let planned = match (now.as_ref(), found.get(path)) {
        (None, None) => true,
        (Some(now), Some(found)) => now.same_as(found),
        _ => false,
    };
    if !planned {
        tracing::debug!(%repository, %path, "left alone: it changed after the plan read it");
    }
    planned
}

/// Why a whole transfer was refused. Each is a state the sender handles.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ApplyRefusal {
    /// The repository moved between the plan and the apply.
    EpochMoved,
    /// A path that may never be written, or one that escapes the repository.
    Forbidden(String),
    /// Content that does not match what it was sent as.
    Malformed(String),
    Internal(String),
}

/// Move a staged file into place, atomically, with its mode.
fn place_file(source: &Path, target: &Path, executable: bool) -> Result<(), std::io::Error> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let mode = if executable { 0o755 } else { 0o644 };
        std::fs::set_permissions(source, std::fs::Permissions::from_mode(mode))?;
    }
    #[cfg(not(unix))]
    let _ = executable;

    // A symlink at the target would be followed by a rename onto it, so it is
    // removed first: the peer sent a FILE for this path.
    if let Ok(meta) = std::fs::symlink_metadata(target) {
        if meta.is_symlink() {
            std::fs::remove_file(target)?;
        }
    }
    std::fs::rename(source, target)
}

/// Replace whatever is at the target with a symlink to what the peer sent.
fn place_symlink(source: &Path, target: &Path) -> Result<(), std::io::Error> {
    let text = std::fs::read_to_string(source)?;
    match std::fs::symlink_metadata(target) {
        Ok(meta) if meta.is_dir() => {
            // A directory where a link should be is not something to remove
            // recursively on a peer's say-so.
            return Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                "a directory is at the path a symlink was sent for",
            ));
        }
        Ok(_) => std::fs::remove_file(target)?,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => return Err(e),
    }
    #[cfg(unix)]
    {
        std::os::unix::fs::symlink(text.trim_end_matches('\n'), target)
    }
    #[cfg(windows)]
    {
        // Windows needs to know whether the target is a directory, and
        // creating one requires a privilege an ordinary account does not have.
        // Refused rather than guessed: a file where a link belongs is a
        // difference that would then synchronise back the other way for ever.
        let _ = text;
        Err(std::io::Error::new(
            std::io::ErrorKind::Unsupported,
            "symlinks cannot be created on this machine",
        ))
    }
}

/// Remove directories a deletion emptied, up to (but never including) the
/// repository root.
fn prune_empty_parents(workdir: &Path, from: &Path) {
    let mut current = from.parent().map(Path::to_path_buf);
    while let Some(dir) = current {
        if dir == workdir || !dir.starts_with(workdir) {
            return;
        }
        let empty = match std::fs::read_dir(&dir) {
            Ok(mut entries) => entries.next().is_none(),
            Err(_) => return,
        };
        if !empty || std::fs::remove_dir(&dir).is_err() {
            return;
        }
        current = dir.parent().map(Path::to_path_buf);
    }
}