diff --git a/crates/loopd/src/main.rs b/crates/loopd/src/main.rs index 4c22848..94df153 100644 --- a/crates/loopd/src/main.rs +++ b/crates/loopd/src/main.rs @@ -102,6 +102,18 @@ fn run_serve(path: &Path) -> ExitCode { return ExitCode::from(1); } + // Every startup notice comes before the socket is bound, so "serving on" is the last line + // before loopd serves: a supervisor or test that waits for it has seen all the others. + let port: Box = match &cfg.broker.socket { + Some(socket) => Box::new(BrokerPort::new( + socket.clone(), + Duration::from_millis(cfg.broker.timeout_ms), + )), + None => { + eprintln!("{}", not_configured_line()); + Box::new(NoBroker) + } + }; if let Some(parent) = socket.parent() && let Err(e) = std::fs::create_dir_all(parent) { @@ -121,16 +133,6 @@ fn run_serve(path: &Path) -> ExitCode { } eprintln!("loopd: serving on {}", socket.display()); - let port: Box = match &cfg.broker.socket { - Some(socket) => Box::new(BrokerPort::new( - socket.clone(), - Duration::from_millis(cfg.broker.timeout_ms), - )), - None => { - eprintln!("{}", not_configured_line()); - Box::new(NoBroker) - } - }; let memory = Box::new(MemoryPort::new(cfg.memory_socket(), MEMORY_WAIT)); let ctx = Arc::new(Context::new(cfg, client, port, Registry::m5a(), memory)); if let Err(e) = channel::serve(listener, ctx) { diff --git a/crates/loopd/tests/pointers.rs b/crates/loopd/tests/pointers.rs index f4f93ff..b7720f9 100644 --- a/crates/loopd/tests/pointers.rs +++ b/crates/loopd/tests/pointers.rs @@ -2,6 +2,8 @@ //! `scripts/check-runbook.sh` checks that the entries exist; these tests check that the messages //! name them. Do not edit. +#[path = "support/serving.rs"] +mod serving; mod support; use std::path::Path; @@ -162,20 +164,10 @@ fn healthy_routes(server: &FakeServer) { ); } -/// Waits for `loopd serve` to bind its socket, stops it, and returns what it printed. -fn stderr_once_serving(mut child: Child, socket: &Path) -> String { - let mut up = false; - for _ in 0..200 { - if socket.exists() { - up = true; - break; - } - std::thread::sleep(Duration::from_millis(50)); - } - child.kill().unwrap(); - let output = child.wait_with_output().unwrap(); - let stderr = String::from_utf8_lossy(&output.stderr).to_string(); - assert!(up, "the socket never appeared; stderr: {stderr}"); +/// Waits for `loopd serve` to print "serving on", stops it, and returns what it printed. +fn stderr_once_serving(child: Child) -> String { + let (up, stderr) = serving::stderr_once_serving(child); + assert!(up, "loopd never said it was serving; stderr: {stderr}"); stderr } @@ -218,8 +210,7 @@ fn serve_without_a_broker_says_so_once_with_the_entry() { let server = FakeServer::start(); healthy_routes(&server); let config = config_file(&home, &server, 2, ""); - let socket = home.dir.join("run/loop/loop.sock"); - let stderr = stderr_once_serving(serve(&config), &socket); + let stderr = stderr_once_serving(serve(&config)); let lines: Vec<&str> = stderr .lines() .filter(|l| l.contains("no tool broker is configured")) @@ -240,8 +231,7 @@ fn serve_with_a_broker_socket_does_not_say_it() { home.dir.join("run/loop-broker/broker.sock").display() ); let config = config_file(&home, &server, 2, &extra); - let socket = home.dir.join("run/loop/loop.sock"); - let stderr = stderr_once_serving(serve(&config), &socket); + let stderr = stderr_once_serving(serve(&config)); assert!(!stderr.contains("no tool broker"), "{stderr}"); assert!(stderr.contains("serving on"), "{stderr}"); } diff --git a/crates/loopd/tests/serve.rs b/crates/loopd/tests/serve.rs index b0313eb..42c32dc 100644 --- a/crates/loopd/tests/serve.rs +++ b/crates/loopd/tests/serve.rs @@ -1,10 +1,11 @@ //! Tests for the `loopd serve` command. Do not edit. +#[path = "support/serving.rs"] +mod serving; mod support; use std::os::unix::fs::PermissionsExt; use std::process::{Command, Stdio}; -use std::time::Duration; use support::{FakeServer, Home, Reply, tokenizer}; fn config_file(home: &Home, server: &FakeServer, expect_slots: u32) -> std::path::PathBuf { @@ -56,25 +57,16 @@ fn serve_runs_the_self_test_then_binds_the_socket_with_mode_0600() { healthy_routes(&server); let config = config_file(&home, &server, 2); let socket = home.dir.join("run").join("loop").join("loop.sock"); - let mut child = Command::new(env!("CARGO_BIN_EXE_loopd")) + let child = Command::new(env!("CARGO_BIN_EXE_loopd")) .args(["serve", "--config"]) .arg(&config) .stderr(Stdio::piped()) .spawn() .unwrap(); - let mut up = false; - for _ in 0..200 { - if socket.exists() { - up = true; - break; - } - std::thread::sleep(Duration::from_millis(50)); - } + // "serving on" comes after the mode is set; the socket stays after loopd is killed. + let (up, stderr) = serving::stderr_once_serving(child); + assert!(up, "loopd never said it was serving; stderr: {stderr}"); let mode = std::fs::metadata(&socket).map(|m| m.permissions().mode() & 0o777); - let _ = child.kill(); - let output = child.wait_with_output().unwrap(); - let stderr = String::from_utf8_lossy(&output.stderr); - assert!(up, "the socket never appeared; stderr: {stderr}"); assert_eq!(mode.unwrap(), 0o600, "{stderr}"); assert!(stderr.contains("selftest: ok"), "{stderr}"); assert!(stderr.contains("serving on"), "{stderr}"); diff --git a/crates/loopd/tests/support/serving.rs b/crates/loopd/tests/support/serving.rs new file mode 100644 index 0000000..2420882 --- /dev/null +++ b/crates/loopd/tests/support/serving.rs @@ -0,0 +1,63 @@ +//! Waiting for `loopd serve` to be up, by the line it prints last before it serves. +//! +//! The socket appears before `loopd` has set its mode and printed "serving on", so a test that +//! waits for the socket and then stops `loopd` can lose those lines (M5c-1 review, finding 7). +//! Every startup notice comes before "serving on", so once it is read, the rest has been printed. + +use std::io::{BufRead, BufReader, Read}; +use std::process::Child; +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +/// How long `loopd` may take to pass its self-test and bind, on a loaded machine. +const UP: Duration = Duration::from_secs(60); + +/// Reads `child`'s stderr until the "serving on" line, stops it, and returns whether that line came +/// and everything it printed. `child` must have been spawned with `stderr(Stdio::piped())`. +pub fn stderr_once_serving(mut child: Child) -> (bool, String) { + let stderr = child.stderr.take().expect("stderr is piped"); + let (tx, rx) = mpsc::channel::(); + let reader = std::thread::spawn(move || { + let mut lines = BufReader::new(stderr); + let mut line = String::new(); + loop { + line.clear(); + match lines.read_line(&mut line) { + Ok(0) | Err(_) => break, + Ok(_) => { + if tx.send(line.clone()).is_err() { + break; + } + } + } + } + let mut rest = String::new(); + let _ = lines.read_to_string(&mut rest); + rest + }); + let deadline = Instant::now() + UP; + let mut text = String::new(); + let mut up = false; + while let Some(left) = deadline.checked_duration_since(Instant::now()) { + match rx.recv_timeout(left) { + Ok(line) => { + up = line.starts_with("loopd: serving on "); + text.push_str(&line); + if up { + break; + } + } + Err(_) => break, + } + } + let _ = child.kill(); + let _ = child.wait(); + // The reader ends at the pipe's end once `loopd` is gone; join it before draining, so no line + // it sends after the drain is lost. + let rest = reader.join().unwrap_or_default(); + while let Ok(line) = rx.try_recv() { + text.push_str(&line); + } + text.push_str(&rest); + (up, text) +}