//! 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 `/.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, 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, speaker: None, }), }; 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 `/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"); }