nordisk/heimdall

crates/heimdall-plain/src/forklock.rs

// SPDX-License-Identifier: AGPL-3.0-only
// Copyright (C) 2026 Ignalina ApS and Vetra Automation AB

//! The fork guard against `ETXTBSY`. **Heimdall is a defensive sandbox.**
//!
//! A runner unpacks a job's root (its programs included) while other
//! threads of the same process start shims for other jobs. A shim started
//! by `fork` + `execve` inherits every open fd until its own `execve`
//! closes the `O_CLOEXEC` ones; if that window covers a program another
//! thread is still writing, that job's `execve` of the program fails with
//! `ETXTBSY` ("Text file busy"): a flaky, load-dependent setup error.
//!
//! The rule: everything that writes files a job will execute holds
//! [`writing`] (shared: unpacks run in parallel); everything that starts a
//! child process holds [`spawning`] (exclusive) until `spawn` returns
//! (`std::process::Command::spawn` returns after the child's `execve`).
//! So no child ever holds a write fd of a job's program.

use std::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};

static LOCK: RwLock<()> = RwLock::new(());

/// Hold while writing files a job may execute (shared).
pub fn writing() -> RwLockReadGuard<'static, ()> {
    LOCK.read().unwrap_or_else(|e| e.into_inner())
}

/// Hold while starting a child process (exclusive against [`writing`]).
pub fn spawning() -> RwLockWriteGuard<'static, ()> {
    LOCK.write().unwrap_or_else(|e| e.into_inner())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::mpsc;
    use std::time::Duration;

    /// The lock is process-wide: these tests must not run at the same time
    /// (a waiting spawning() also holds back new readers).
    static SERIAL: std::sync::Mutex<()> = std::sync::Mutex::new(());

    #[test]
    fn a_spawn_waits_while_a_job_file_is_being_written() {
        let _t = SERIAL.lock().unwrap_or_else(|e| e.into_inner());
        let w = writing();
        let (tx, rx) = mpsc::channel();
        let t = std::thread::spawn(move || {
            let _s = spawning();
            tx.send(()).unwrap();
        });
        assert!(
            rx.recv_timeout(Duration::from_millis(300)).is_err(),
            "spawning() returned while writing() was held"
        );
        drop(w);
        rx.recv_timeout(Duration::from_secs(5))
            .expect("spawning() after writing() ends");
        t.join().unwrap();
    }

    #[test]
    fn writers_do_not_wait_for_each_other() {
        let _t = SERIAL.lock().unwrap_or_else(|e| e.into_inner());
        let a = writing();
        let (tx, rx) = mpsc::channel();
        let t = std::thread::spawn(move || {
            let _b = writing();
            tx.send(()).unwrap();
        });
        rx.recv_timeout(Duration::from_secs(5))
            .expect("a second writer runs at once");
        drop(a);
        t.join().unwrap();
    }
}