boxmaker/docs/plans/M5b/files/crates/loopd/tests/instance.rs

//! One `loopd` per home (M5b spec, section 9a): a second `loopd serve` against the same home is
//! refused before it touches the socket or the server, and the first keeps serving; once the first
//! is killed, a new one removes the stale socket and serves. Every wait has a deadline, and every
//! process is killed and every path removed by a guard (tip T44). Do not edit.

mod support;

use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

use bxcore::{
    Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, WireError, read_frame,
    write_frame,
};
use support::{FakeServer, Home, Reply, tokenizer};

const POINTER: &str = "see docs/runbook.md#already-running";

fn config_file(home: &Home, server: &FakeServer) -> PathBuf {
    let text = format!(
        r#"
[infer]
socket = "{}"
model = "test-model"
[slots]
main = 0
background = 1
[expect]
template_sha256 = "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b"
n_ctx = 131072
slots = 2
[limits]
poll_ms = 40
liveness_ms = 500
retry_backoff_ms = [10]
[paths]
home = "{}"
"#,
        server.socket.display(),
        home.dir.display()
    );
    let path = home.dir.join("config.toml");
    std::fs::write(&path, text).unwrap();
    path
}

/// A server that passes the self-test twice: once for the first `loopd`, once for the one started
/// after it is killed.
fn healthy_routes(server: &FakeServer) {
    server.route("/props", vec![Reply::fixture("props")]);
    server.respond("/tokenize", tokenizer);
    server.route(
        "/v1/chat/completions",
        vec![
            Reply::fixture("tool_call"),
            Reply::fixture("turn1"),
            Reply::fixture("turn2"),
            Reply::fixture("tool_call"),
            Reply::fixture("turn1"),
            Reply::fixture("turn2"),
        ],
    );
}

fn socket(home: &Home) -> PathBuf {
    home.dir.join("run").join("loop").join("loop.sock")
}

/// A running `loopd serve`, killed and waited for when dropped. Declared after the `Home` and the
/// `FakeServer`, so it drops first.
struct Daemon(Child);

impl Drop for Daemon {
    fn drop(&mut self) {
        let _ = self.0.kill();
        let _ = self.0.wait();
    }
}

impl Daemon {
    fn running(&mut self) -> bool {
        matches!(self.0.try_wait(), Ok(None))
    }
}

/// `loopd serve` with its stderr in `<home>/<name>.err`, for the messages of a failing test.
fn start(home: &Home, config: &Path, name: &str) -> Daemon {
    let err = std::fs::File::create(home.dir.join(format!("{name}.err"))).unwrap();
    let child = Command::new(env!("CARGO_BIN_EXE_loopd"))
        .args(["serve", "--config"])
        .arg(config)
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::from(err))
        .spawn()
        .unwrap();
    Daemon(child)
}

fn stderr_of(home: &Home, name: &str) -> String {
    std::fs::read_to_string(home.dir.join(format!("{name}.err"))).unwrap_or_default()
}

/// Runs `loopd serve` to its end, at most 20 s (then it is killed), and gives its exit code
/// (`None` if it had to be killed) and its stderr.
fn run_to_end(config: &Path) -> (Option<i32>, String) {
    let mut child = Daemon(
        Command::new(env!("CARGO_BIN_EXE_loopd"))
            .args(["serve", "--config"])
            .arg(config)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap(),
    );
    let deadline = Instant::now() + Duration::from_secs(20);
    let mut code = None;
    while Instant::now() < deadline {
        if let Ok(Some(status)) = child.0.try_wait() {
            code = status.code();
            break;
        }
        std::thread::sleep(Duration::from_millis(20));
    }
    let _ = child.0.kill();
    let _ = child.0.wait();
    let mut stderr = String::new();
    if let Some(mut pipe) = child.0.stderr.take() {
        let _ = std::io::Read::read_to_string(&mut pipe, &mut stderr);
    }
    (code, stderr)
}

/// True when `loopd` answers on `socket`: a turn that resumes a session that does not exist gets
/// the error frame for it. Any other answer, or none within 5 s, is false.
fn answers(socket: &Path) -> bool {
    let Ok(mut stream) = UnixStream::connect(socket) else {
        return false;
    };
    let limit = Some(Duration::from_secs(5));
    if stream.set_read_timeout(limit).is_err() || stream.set_write_timeout(limit).is_err() {
        return false;
    }
    let turn = Envelope {
        v: PROTOCOL_VERSION,
        id: 7,
        r#final: true,
        msg: Message::Turn(Turn {
            session: SessionId::new("probe").unwrap(),
            content: "are you there".to_string(),
            resume: true,
        }),
    };
    if write_frame(&mut stream, &turn).is_err() {
        return false;
    }
    matches!(
        read_frame(&mut stream),
        Ok(Envelope {
            msg: Message::Error(WireError {
                code: ErrorCode::NoSuchSession,
                ..
            }),
            ..
        })
    )
}

/// Waits up to 20 s for `loopd` to answer on `socket`.
fn wait_until_it_answers(socket: &Path) -> bool {
    let deadline = Instant::now() + Duration::from_secs(20);
    while Instant::now() < deadline {
        if answers(socket) {
            return true;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    false
}

fn inode(path: &Path) -> u64 {
    use std::os::unix::fs::MetadataExt;
    std::fs::symlink_metadata(path)
        .map(|m| m.ino())
        .unwrap_or(0)
}

#[test]
fn a_second_loopd_on_the_same_home_is_refused_and_the_first_keeps_serving() {
    let home = Home::new();
    let server = FakeServer::start();
    healthy_routes(&server);
    let config = config_file(&home, &server);
    let socket = socket(&home);
    let mut first = start(&home, &config, "first");
    assert!(
        wait_until_it_answers(&socket),
        "the first loopd never answered; stderr: {}",
        stderr_of(&home, "first")
    );
    let before = inode(&socket);
    let completions = server.requests_to("/v1/chat/completions").len();
    let tokenized = server.requests_to("/tokenize").len();

    let (code, stderr) = run_to_end(&config);
    let lock = home.dir.join("run").join("loopd.lock");
    let line = format!(
        "loopd: another loopd is running (it holds {}); {POINTER}",
        lock.display()
    );
    assert_eq!(code, Some(1), "the second loopd: {stderr}");
    assert!(
        stderr.lines().any(|l| l == line),
        "wanted the line\n{line}\nstderr: {stderr}"
    );
    assert!(
        !stderr.contains("selftest"),
        "the second ran no self-test: {stderr}"
    );
    assert_eq!(
        (
            server.requests_to("/v1/chat/completions").len(),
            server.requests_to("/tokenize").len()
        ),
        (completions, tokenized),
        "the second sent the server nothing"
    );

    assert!(first.running(), "the first loopd is still running");
    assert_eq!(inode(&socket), before, "the first's socket is still there");
    assert!(
        answers(&socket),
        "the first loopd still answers; its stderr: {}",
        stderr_of(&home, "first")
    );
}

#[test]
fn after_the_first_is_killed_a_new_loopd_removes_the_stale_socket_and_serves() {
    let home = Home::new();
    let server = FakeServer::start();
    healthy_routes(&server);
    let config = config_file(&home, &server);
    let socket = socket(&home);
    let mut first = start(&home, &config, "first");
    assert!(
        wait_until_it_answers(&socket),
        "the first loopd never answered; stderr: {}",
        stderr_of(&home, "first")
    );
    let _ = first.0.kill();
    let _ = first.0.wait();
    assert!(socket.exists(), "a killed loopd leaves its socket");
    assert!(!answers(&socket), "nobody serves the stale socket");

    let _next = start(&home, &config, "next");
    assert!(
        wait_until_it_answers(&socket),
        "the new loopd never answered; stderr: {}",
        stderr_of(&home, "next")
    );
    let lock = home.dir.join("run").join("loopd.lock");
    let mode = std::fs::metadata(&lock).map(|m| m.permissions().mode() & 0o777);
    assert_eq!(mode.ok(), Some(0o600), "the lock file, mode 0600");
    assert_eq!(
        server.requests_to("/v1/chat/completions").len(),
        6,
        "each of the two ran the self-test"
    );
}

/// Gives `<home>/run` back its owner's permissions when dropped, before the `Home` is removed.
struct Restore(PathBuf);

impl Drop for Restore {
    fn drop(&mut self) {
        let _ = std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700));
    }
}

#[test]
fn a_run_directory_that_cannot_be_written_stops_loopd_naming_the_error() {
    let home = Home::new();
    let server = FakeServer::start();
    healthy_routes(&server);
    let config = config_file(&home, &server);
    let run = home.dir.join("run");
    std::fs::create_dir(&run).unwrap();
    let _restore = Restore(run.clone());
    std::fs::set_permissions(&run, std::fs::Permissions::from_mode(0o500)).unwrap();

    let (code, stderr) = run_to_end(&config);
    let line = format!(
        "loopd: cannot take the lock {}: Permission denied (os error 13); {POINTER}",
        run.join("loopd.lock").display()
    );
    assert_eq!(code, Some(1), "{stderr}");
    assert!(
        stderr.lines().any(|l| l == line),
        "wanted the line\n{line}\nstderr: {stderr}"
    );
    assert!(
        server.requests().is_empty(),
        "loopd sent the server nothing"
    );
    assert!(!socket(&home).exists(), "no socket");
}