exmpeg

//! Atomic output writes for operations that mux to disk.
//!
//! Every output-producing op (remux, extract_frame, extract_audio,
//! concat, transcode) runs against a sibling
//! `<stem>.partial.<nonce>.<ext>` path and renames onto the final
//! destination only after the muxer trailer has been written
//! successfully. On error the partial file is removed.
//!
//! The partial path lives in the same directory as the destination so
//! the final rename is a single inode link rename (atomic on POSIX)
//! and so we do not silently spill output across filesystems.
//!
//! The `.partial.<nonce>` infix lands before the file extension, not
//! after: libavformat picks the muxer from the extension, and
//! `out.mp4.partial` has no known extension. `out.mp4` becomes
//! `out.partial.<nonce>.mp4`.
//!
//! The `<nonce>` (a per-process random base + the OS process id + a
//! process-wide counter + a nanosecond timestamp) makes the partial path
//! unique per call, including across two nodes on shared storage that
//! happen to share a pid. Two concurrent
//! writes to the same destination - duplicate jobs, a retry racing a
//! slow first attempt, two nodes on shared storage - therefore never
//! share a partial file, so one call can never unlink or rename another
//! call's in-progress output. The guarantee is last-complete-rename-wins:
//! every state an observer can see at the destination is a complete file.
//!
//! Because each partial path is unique it never pre-exists, so there is
//! no stale-partial cleanup on entry. A hard crash (SIGKILL, power loss)
//! mid-write can therefore leave a `<stem>.partial.*` sibling behind; it
//! is never renamed onto the destination and can be swept by the caller.

use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

use crate::errors::NativeError;

/// Run `body` against a unique `<final>.partial.<nonce>` path, then
/// rename the resulting file onto `final_path`. On error the partial
/// file is removed (best effort) so a failed call never leaves a
/// half-written output on disk.
pub(crate) fn run<T, F>(final_path: &str, body: F) -> Result<T, NativeError>
where
    F: FnOnce(&Path) -> Result<T, NativeError>,
{
    // The output string is handed to libavformat, which selects its AVIO
    // write protocol from the scheme: an output of `ftp://`, `http://`,
    // `rtmp://`, `tcp://`, or a nested wrapper like `tee:http://...`
    // would make the muxer open a network write (write-side SSRF) before
    // the local rename. The atomic-rename design assumes a local path, so
    // reject any protocol scheme at this single chokepoint that every
    // output-producing op funnels through.
    if has_url_scheme(final_path) {
        return Err(
            NativeError::new("invalid_request", "output must be a local path, not a URL")
                .with_detail("output", final_path.to_owned()),
        );
    }

    let final_path = PathBuf::from(final_path);
    let partial = partial_path_for(&final_path, &partial_nonce());

    // Remove the partial on any early exit - an `Err` from `body`, a
    // failed rename, OR a panic unwinding out of `body` (which
    // `run_with_panic_protection` catches one frame up, after this scope
    // has already dropped). The guard is disarmed only once the rename
    // onto the destination has succeeded.
    let mut guard = PartialGuard::new(&partial);

    let value = body(&partial)?;

    std::fs::rename(&partial, &final_path).map_err(|err| {
        NativeError::new(
            "io_error",
            "could not rename partial output onto destination",
        )
        .with_detail("from", partial.display().to_string())
        .with_detail("to", final_path.display().to_string())
        .with_detail("cause", err.to_string())
    })?;

    guard.disarm();
    Ok(value)
}

/// Removes the `.partial` file when dropped unless disarmed. This covers
/// the panic path the explicit error branches cannot: a panic inside
/// `body` unwinds past `run` before any value is produced, and
/// `catch_unwind` sits a frame up in `lib.rs`, so without this guard the
/// partial would leak on disk.
struct PartialGuard<'a> {
    partial: &'a Path,
    armed: bool,
}

impl<'a> PartialGuard<'a> {
    fn new(partial: &'a Path) -> Self {
        Self {
            partial,
            armed: true,
        }
    }

    fn disarm(&mut self) {
        self.armed = false;
    }
}

impl Drop for PartialGuard<'_> {
    fn drop(&mut self) {
        if self.armed {
            // Best effort: a NotFound (the muxer never created the file)
            // is fine, and we must not mask the original failure.
            let _ = std::fs::remove_file(self.partial);
        }
    }
}

/// A per-call token that makes the partial path unique. Combines a
/// per-process random base (separates two OS processes even if they
/// share a pid on shared storage and issue their nth write in the same
/// `SystemTime` tick - the case a pid + counter + timestamp alone could
/// still collide on), the OS process id, a process-wide monotonic
/// counter (separates calls within one process), and a nanosecond
/// timestamp.
fn partial_nonce() -> String {
    use std::collections::hash_map::RandomState;
    use std::hash::{BuildHasher, Hasher};
    use std::sync::OnceLock;

    static COUNTER: AtomicU64 = AtomicU64::new(0);
    // `RandomState` is seeded from the OS RNG, so a hash off a fresh one
    // is effectively a random per-process u64. Computed once and reused.
    static RAND_BASE: OnceLock<u64> = OnceLock::new();
    let base = *RAND_BASE.get_or_init(|| {
        let mut hasher = RandomState::new().build_hasher();
        hasher.write_u8(0);
        hasher.finish()
    });

    let seq = COUNTER.fetch_add(1, Ordering::Relaxed);
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_nanos());
    format!("{base:016x}.{}.{seq}.{nanos}", std::process::id())
}

/// Whether `s` carries an FFmpeg protocol scheme that would make the
/// muxer open a non-local destination.
///
/// libavformat selects the AVIO protocol from the prefix before the
/// first `:` (when that `:` comes before any `/`). We reject any such
/// scheme of two or more characters: this covers plain `http://...` and
/// the nested wrappers libavformat resolves from the prefix, e.g.
/// `tee:http://host/out` or `concat:a.mp4|b.mp4`, which carry an inner
/// `://` (or none) and so slipped past a `://`-only check. A single-
/// letter scheme is a Windows drive (`C:\out.mp4`), and a path whose
/// first `:`-or-`/` is a `/` (or that has neither) is local; both are
/// allowed.
fn has_url_scheme(s: &str) -> bool {
    let Some(idx) = s.find([':', '/']) else {
        return false;
    };
    if s.as_bytes()[idx] != b':' {
        return false;
    }
    let scheme = &s[..idx];
    scheme.len() >= 2
        && scheme.starts_with(|c: char| c.is_ascii_alphabetic())
        && scheme
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'+' | b'-' | b'.'))
}

fn partial_path_for(final_path: &Path, nonce: &str) -> PathBuf {
    // libavformat picks the muxer from the file extension, so the
    // marker has to land BEFORE the extension: `out.mp4` becomes
    // `out.partial.<nonce>.mp4`, never `out.mp4.partial`. Files without
    // an extension fall back to a trailing `.partial.<nonce>` suffix.
    let stem = final_path.file_stem().map(OsString::from);
    let ext = final_path.extension();

    let mut name = match (stem, ext) {
        (Some(stem), Some(ext)) => {
            let mut n = stem;
            n.push(".partial.");
            n.push(nonce);
            n.push(".");
            n.push(ext);
            n
        }
        (Some(stem), None) => {
            let mut n = stem;
            n.push(".partial.");
            n.push(nonce);
            n
        }
        _ => OsString::from(format!(".partial.{nonce}")),
    };
    if name.is_empty() {
        name = OsString::from(format!(".partial.{nonce}"));
    }
    match final_path.parent() {
        Some(parent) if !parent.as_os_str().is_empty() => parent.join(name),
        _ => PathBuf::from(name),
    }
}

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

    fn tempdir() -> PathBuf {
        let base = std::env::temp_dir().join(format!(
            "exmpeg-atomic-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&base).unwrap();
        base
    }

    #[test]
    fn renames_partial_onto_destination_on_success() {
        let dir = tempdir();
        let final_path = dir.join("out.bin");
        let seen_partial = std::cell::RefCell::new(PathBuf::new());

        let result = run(final_path.to_str().unwrap(), |p| {
            *seen_partial.borrow_mut() = p.to_path_buf();
            let mut f = std::fs::File::create(p).unwrap();
            f.write_all(b"payload").unwrap();
            Ok::<_, NativeError>(())
        });

        let partial = seen_partial.into_inner();
        assert!(result.is_ok());
        assert!(final_path.exists());
        assert!(!partial.exists());
        // The partial is a `<stem>.partial.<nonce>.<ext>` sibling whose
        // extension is preserved last for muxer detection.
        let name = partial.file_name().unwrap().to_str().unwrap();
        assert!(name.starts_with("out.partial."));
        assert_eq!(partial.extension().unwrap(), "bin");
        assert_eq!(partial.parent(), final_path.parent());
        assert_eq!(std::fs::read(&final_path).unwrap(), b"payload");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn removes_partial_and_does_not_create_destination_on_error() {
        let dir = tempdir();
        let final_path = dir.join("out.bin");
        let seen_partial = std::cell::RefCell::new(PathBuf::new());

        let result: Result<(), NativeError> = run(final_path.to_str().unwrap(), |p| {
            *seen_partial.borrow_mut() = p.to_path_buf();
            let mut f = std::fs::File::create(p).unwrap();
            f.write_all(b"half-written").unwrap();
            Err(NativeError::new("decode_error", "boom"))
        });

        assert!(result.is_err());
        assert!(!final_path.exists());
        assert!(!seen_partial.into_inner().exists());
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn partial_path_is_unique_per_call() {
        let final_path = PathBuf::from("/tmp/out.mp4");
        let a = partial_path_for(&final_path, &partial_nonce());
        let b = partial_path_for(&final_path, &partial_nonce());

        // Two calls against the same destination never collide.
        assert_ne!(a, b);
        // The extension stays last so the muxer still detects the format.
        assert_eq!(a.extension().unwrap(), "mp4");
        assert_eq!(b.extension().unwrap(), "mp4");
        assert!(
            a.file_name()
                .unwrap()
                .to_str()
                .unwrap()
                .starts_with("out.partial.")
        );
    }

    #[test]
    fn extensionless_partial_keeps_nonce_suffix() {
        let partial = partial_path_for(&PathBuf::from("/tmp/out"), "X");
        assert_eq!(partial.file_name().unwrap(), "out.partial.X");
    }

    #[test]
    fn removes_partial_when_body_panics() {
        let dir = tempdir();
        let final_path = dir.join("out.bin");
        // The partial path carries a per-call random nonce, so capture the
        // actual one the closure was handed rather than predicting it.
        let seen_partial = std::cell::RefCell::new(PathBuf::new());

        // Mirror the production layering: `catch_unwind` wraps `run` a
        // frame up, so the guard must remove the partial during unwind.
        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            run(final_path.to_str().unwrap(), |p| {
                *seen_partial.borrow_mut() = p.to_path_buf();
                std::fs::write(p, b"half-written").unwrap();
                panic!("boom");
                #[allow(unreachable_code)]
                Ok::<_, NativeError>(())
            })
        }));

        assert!(caught.is_err(), "the panic should propagate out of run");
        assert!(
            !seen_partial.into_inner().exists(),
            "the partial must be cleaned up on panic"
        );
        assert!(!final_path.exists());
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn rejects_url_scheme_output_without_invoking_body() {
        for url in [
            "http://host/out.mp4",
            "https://host/out.mp4",
            "ftp://host/out.mp4",
            "rtmp://host/live",
            "tcp://host:9000",
            "file:///tmp/out.mp4",
            // Nested wrappers libavformat resolves from the prefix.
            "tee:http://host/out.mp4",
            "concat:a.mp4|b.mp4",
        ] {
            let result: Result<(), NativeError> = run(url, |_p| {
                panic!("body must not run for a URL output");
            });
            let err = result.unwrap_err();
            assert_eq!(err.r#type, "invalid_request", "for {url}");
        }
    }

    #[test]
    fn has_url_scheme_distinguishes_paths_from_protocols() {
        for path in [
            "/tmp/out.mp4",
            "out.mp4",
            "./rel/out.mkv",
            "C:\\Users\\out.mp4", // Windows drive letter (single-char scheme)
            "/has spaces/out.webm",
        ] {
            assert!(!has_url_scheme(path), "{path} should not be a URL");
        }
        for url in [
            "s3://bucket/key",
            "tee:http://host/out.mp4",
            "concat:a.mp4|b.mp4",
            "subfile:,start,123,end,456,:/tmp/x.mp4",
        ] {
            assert!(has_url_scheme(url), "{url} should be flagged as a protocol");
        }
        assert!(!has_url_scheme("://nohost")); // empty scheme
        assert!(!has_url_scheme("1http://x")); // scheme must start alpha
    }
}