From 6af89f7c608b6b72d73df6f2871b88837e1ea8c9 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Tue, 22 Sep 2026 21:07:59 -0700 Subject: [PATCH] brokerd: runbook pointers for startup failures, no thread panics, args_os M3a review findings 3, 5 (the cast), 6, 7 (brokerd), 11. A config, directory or socket failure at start now ends with docs/runbook.md#brokerd-start-failed, and losing a listener with #brokerd-listener-lost; both entries are new. Threads start through thread::Builder, so a refused thread is reported instead of silently killing a listener; an aborted connection no longer stops the daemon. brokerd reads args_os and keeps the config path as a path. The "requester went away" result is recorded at the time it happens. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/brokerd/src/args.rs | 2 +- crates/brokerd/src/broker.rs | 2 +- crates/brokerd/src/grants.rs | 6 +- crates/brokerd/src/main.rs | 51 ++++++---- crates/brokerd/src/serve.rs | 81 ++++++++++----- crates/brokerd/tests/serve_pointers.rs | 134 +++++++++++++++++++++++++ docs/runbook.md | 58 +++++++++++ 7 files changed, 284 insertions(+), 50 deletions(-) create mode 100644 crates/brokerd/tests/serve_pointers.rs diff --git a/crates/brokerd/src/args.rs b/crates/brokerd/src/args.rs index a691c2c..d6ba64c 100644 --- a/crates/brokerd/src/args.rs +++ b/crates/brokerd/src/args.rs @@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize}; use std::fmt; - +/// Maximum length, in bytes, of a path. /// Maximum length, in bytes, of a path or of a URL. pub const MAX_PATH: usize = 4096; /// Maximum length, in bytes, of a URL. diff --git a/crates/brokerd/src/broker.rs b/crates/brokerd/src/broker.rs index 7bfa837..468aa07 100644 --- a/crates/brokerd/src/broker.rs +++ b/crates/brokerd/src/broker.rs @@ -301,7 +301,7 @@ fn pending( ToolResponse::Failed { message: GONE.to_string(), }, - now, + Timestamp::now(), // now, not when it was decided: records keep time order ); return None; // 5. the peer left at the last look; nothing ran } diff --git a/crates/brokerd/src/grants.rs b/crates/brokerd/src/grants.rs index 2518e9c..c244fcd 100644 --- a/crates/brokerd/src/grants.rs +++ b/crates/brokerd/src/grants.rs @@ -91,8 +91,7 @@ pub fn load(dir: &Path) -> Result> { match entry { Ok(entry) => names.push(entry.file_name().to_string_lossy().to_string()), Err(error) => { - // An entry that cannot be read is the same problem as the directory; keep going. - names.push(String::new()); + // An entry that cannot be read is the same problem as the directory. return Err(vec![GrantProblem { file: dir.to_string_lossy().to_string(), line: None, @@ -198,7 +197,8 @@ fn span_line(text: &str, error: &toml::de::Error) -> u64 { Some(before) => before, None => text, }; - before.bytes().filter(|&b| b == b'\n').count() as u64 + 1 + let newlines = before.bytes().filter(|&b| b == b'\n').count(); + u64::try_from(newlines).map_or(u64::MAX, |n| n.saturating_add(1)) } None => 1, } diff --git a/crates/brokerd/src/main.rs b/crates/brokerd/src/main.rs index daf5239..bedebd1 100644 --- a/crates/brokerd/src/main.rs +++ b/crates/brokerd/src/main.rs @@ -1,5 +1,6 @@ //! `brokerd`: the broker role. `serve` takes the audit lock, binds both sockets, and serves. +use std::ffi::OsString; use std::path::PathBuf; use std::process::ExitCode; use std::sync::Arc; @@ -10,10 +11,15 @@ use brokerd::runner::Refusing; use brokerd::serve::{self, ServeError}; const USAGE: &str = "usage: brokerd serve --config [--accept-break]"; +/// Where the owner looks when brokerd cannot start: its config, a directory or a socket. +const START_FAILED: &str = "see docs/runbook.md#brokerd-start-failed"; +/// Where the owner looks when brokerd stops serving after it started. +const LISTENER_LOST: &str = "see docs/runbook.md#brokerd-listener-lost"; fn main() -> ExitCode { // The first argument must be `serve`. - let mut args = std::env::args().skip(1); + // `args_os`: a path need not be UTF-8, and `args` would panic on one that is not. + let mut args = std::env::args_os().skip(1); match args.next() { Some(command) if command == "serve" => {} _ => return usage(), @@ -29,7 +35,7 @@ fn main() -> ExitCode { let cfg = match Config::load(&config_path) { Ok(cfg) => cfg, Err(e) => { - eprintln!("brokerd: {e}"); + eprintln!("brokerd: {e}\n{START_FAILED}"); return ExitCode::from(1); } }; @@ -49,10 +55,15 @@ fn main() -> ExitCode { eprintln!("brokerd: {e}"); return ExitCode::from(2); } - Err(e) => { + // The audit errors carry their own entry; the rest are this one. + Err(ServeError::Audit(e)) => { eprintln!("brokerd: {e}"); return ExitCode::from(1); } + Err(e) => { + eprintln!("brokerd: {e}\n{START_FAILED}"); + return ExitCode::from(1); + } }; if started.recovered { @@ -73,7 +84,7 @@ fn main() -> ExitCode { match started.run() { Ok(()) => ExitCode::SUCCESS, Err(e) => { - eprintln!("brokerd: {e}"); + eprintln!("brokerd: stopped serving: {e}\n{LISTENER_LOST}"); ExitCode::from(1) } } @@ -86,8 +97,10 @@ fn usage() -> ExitCode { } /// The `--config` and `--accept-break` flags after `serve`, or `Err(())` for a bad list. -fn parse_serve>(mut args: I) -> Result<(PathBuf, bool), ()> { - let mut config: Option = None; +/// Flags are compared as `OsStr`, so one that is not UTF-8 is simply unknown; the path is kept as +/// given. +fn parse_serve>(mut args: I) -> Result<(PathBuf, bool), ()> { + let mut config: Option = None; let mut accept_break = false; loop { match args.next() { @@ -97,24 +110,22 @@ fn parse_serve>(mut args: I) -> Result<(PathBuf, bool None => Err(()), }; } - Some(arg) => match arg.as_str() { - "--config" => match args.next() { - Some(path) => { - if config.is_some() { - return Err(()); - } - config = Some(path); - } - None => return Err(()), - }, - "--accept-break" => { - if accept_break { + Some(arg) if arg == "--config" => match args.next() { + Some(path) => { + if config.is_some() { return Err(()); } - accept_break = true; + config = Some(path); } - _ => return Err(()), + None => return Err(()), }, + Some(arg) if arg == "--accept-break" => { + if accept_break { + return Err(()); + } + accept_break = true; + } + Some(_) => return Err(()), } } } diff --git a/crates/brokerd/src/serve.rs b/crates/brokerd/src/serve.rs index 447d633..cab7fac 100644 --- a/crates/brokerd/src/serve.rs +++ b/crates/brokerd/src/serve.rs @@ -3,7 +3,7 @@ use std::fs::{DirBuilder, Permissions}; use std::io; use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; -use std::os::unix::net::UnixListener; +use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::mpsc; @@ -148,21 +148,24 @@ impl Started { } = self; // The expiry thread: once a second, expire the approvals whose time has run out. + // `Builder` rather than `spawn`, which panics when the system refuses a thread. { let broker = Arc::clone(&broker); - std::thread::spawn(move || { - loop { - std::thread::sleep(Duration::from_secs(1)); - admin::expire_due(&broker, Timestamp::now()); - } - }); + std::thread::Builder::new() + .name("expiry".to_string()) + .spawn(move || { + loop { + std::thread::sleep(Duration::from_secs(1)); + admin::expire_due(&broker, Timestamp::now()); + } + })?; } // One accept thread per socket, each with its own clone of the broker. let (tx, rx) = mpsc::channel::(); - accept_loop(tools, Arc::clone(&broker), tx.clone(), Which::Tools); - accept_loop(admin, Arc::clone(&broker), tx, Which::Admin); + accept_loop(tools, Arc::clone(&broker), tx.clone(), Which::Tools)?; + accept_loop(admin, Arc::clone(&broker), tx, Which::Admin)?; // The first listener error stops the daemon. match rx.recv() { @@ -172,28 +175,56 @@ impl Started { } } -/// Accept streams from one listener forever, spawning a handler thread per stream. +/// Accept streams from one listener forever, starting a handler thread per stream. An error that +/// only concerns one connection (the peer aborted it, a signal interrupted `accept`) is skipped; +/// any other stops the daemon through `tx`. fn accept_loop( listener: UnixListener, broker: Arc, tx: mpsc::Sender, which: Which, -) { - std::thread::spawn(move || { - for stream in listener.incoming() { - match stream { - Ok(stream) => { - let broker = Arc::clone(&broker); - std::thread::spawn(move || match which { - Which::Tools => broker::handle(stream, &broker), - Which::Admin => admin::handle(stream, &broker), - }); - } - Err(e) => { - // Once `run` has returned, nobody receives; dropping this is correct. - let _ = tx.send(e); +) -> io::Result<()> { + let name = match which { + Which::Tools => "accept-broker", + Which::Admin => "accept-admin", + }; + std::thread::Builder::new() + .name(name.to_string()) + .spawn(move || { + for stream in listener.incoming() { + match stream { + Ok(stream) => serve_one(stream, &broker, which), + Err(e) if one_connection_only(&e) => {} + Err(e) => { + // Once `run` has returned, nobody receives; dropping this is correct. + let _ = tx.send(e); + return; + } } } - } + })?; + Ok(()) +} + +/// Hand one stream to its own thread. If the system refuses the thread, the stream is closed, which +/// the peer sees as the broker being unavailable, and the refusal is printed. +fn serve_one(stream: UnixStream, broker: &Arc, which: Which) { + let for_thread = Arc::clone(broker); + let started = std::thread::Builder::new().spawn(move || match which { + Which::Tools => broker::handle(stream, &for_thread), + Which::Admin => admin::handle(stream, &for_thread), }); + if let Err(e) = started { + broker.log(&format!( + "brokerd: cannot start a thread for a connection, so it was closed: {e}\nsee docs/runbook.md#brokerd-listener-lost" + )); + } +} + +/// Errors from `accept` that end one connection, not the listener. +fn one_connection_only(e: &io::Error) -> bool { + matches!( + e.kind(), + io::ErrorKind::ConnectionAborted | io::ErrorKind::Interrupted | io::ErrorKind::WouldBlock + ) } diff --git a/crates/brokerd/tests/serve_pointers.rs b/crates/brokerd/tests/serve_pointers.rs new file mode 100644 index 0000000..d9860bd --- /dev/null +++ b/crates/brokerd/tests/serve_pointers.rs @@ -0,0 +1,134 @@ +//! `brokerd serve` fails closed at startup with a runbook pointer for every reason it cannot start, +//! and never panics on its own command line. Found in the M3a review (findings 3 and 7). + +#[path = "support/tmp.rs"] +mod tmp; + +use std::ffi::OsStr; +use std::os::unix::ffi::OsStrExt; +use std::path::Path; +use std::process::{Command, Output}; + +use tmp::TempDir; + +const START_FAILED: &str = "see docs/runbook.md#brokerd-start-failed"; + +fn brokerd, S: AsRef>(args: I) -> Output { + Command::new(env!("CARGO_BIN_EXE_brokerd")) + .args(args) + .output() + .unwrap() +} + +fn stderr(out: &Output) -> String { + String::from_utf8_lossy(&out.stderr).into_owned() +} + +/// A config whose home is `dir`, with `sockets` as the `[sockets]` table's body. +fn config(dir: &TempDir, sockets: &str) -> std::path::PathBuf { + let home = dir.path().join("home"); + let grants = dir.path().join("grants"); + std::fs::create_dir_all(&grants).unwrap(); + std::fs::create_dir_all(&home).unwrap(); + dir.write( + "brokerd.toml", + &format!( + "[paths]\nhome = \"{}\"\ngrants = \"{}\"\n[sockets]\n{sockets}\n", + home.display(), + grants.display() + ), + ) +} + +fn fails_with_pointer(out: &Output) { + assert_eq!(out.status.code(), Some(1), "{}", stderr(out)); + assert!( + stderr(out).trim_end().ends_with(START_FAILED), + "{:?} must end with {START_FAILED:?}", + stderr(out) + ); +} + +#[test] +fn a_missing_config_names_the_entry() { + let dir = TempDir::new("ptr-missing"); + let out = brokerd([ + OsStr::new("serve"), + OsStr::new("--config"), + dir.path().join("nope.toml").as_os_str(), + ]); + fails_with_pointer(&out); +} + +#[test] +fn an_invalid_config_names_the_entry() { + let dir = TempDir::new("ptr-invalid"); + let path = dir.write("brokerd.toml", "[paths]\nhome = 3\n"); + let out = brokerd([ + OsStr::new("serve"), + OsStr::new("--config"), + path.as_os_str(), + ]); + fails_with_pointer(&out); +} + +#[test] +fn a_socket_that_cannot_be_bound_names_the_entry() { + let dir = TempDir::new("ptr-bind"); + let long = dir.path().join("s".repeat(120)).join("broker.sock"); + let path = config(&dir, &format!("broker = \"{}\"", long.display())); + fails_with_pointer(&brokerd([ + OsStr::new("serve"), + OsStr::new("--config"), + path.as_os_str(), + ])); +} + +#[test] +fn a_socket_directory_that_cannot_be_made_names_the_entry() { + let dir = TempDir::new("ptr-dir"); + let file = dir.write("a-file", ""); + let path = config( + &dir, + &format!("broker = \"{}\"", file.join("run/broker.sock").display()), + ); + fails_with_pointer(&brokerd([ + OsStr::new("serve"), + OsStr::new("--config"), + path.as_os_str(), + ])); +} + +/// A config path that is not UTF-8 is still a path: it is read (here: not found), not a panic. +#[test] +fn a_config_path_that_is_not_utf8_is_read_as_a_path() { + let dir = TempDir::new("ptr-os"); + let name = dir.path().join(OsStr::from_bytes(b"conf-\xff.toml")); + let out = brokerd([ + OsStr::new("serve"), + OsStr::new("--config"), + name.as_os_str(), + ]); + fails_with_pointer(&out); +} + +/// A flag that is not UTF-8 is not a flag brokerd knows: usage, exit 2, no panic. +#[test] +fn a_flag_that_is_not_utf8_is_a_usage_error() { + let out = brokerd([OsStr::new("serve"), OsStr::from_bytes(b"--\xff")]); + assert_eq!(out.status.code(), Some(2), "{}", stderr(&out)); + assert!(stderr(&out).starts_with("usage: brokerd serve")); + let out = brokerd([OsStr::from_bytes(b"\xff")]); + assert_eq!(out.status.code(), Some(2), "{}", stderr(&out)); +} + +#[test] +fn the_entries_exist() { + let runbook = std::fs::read_to_string( + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../docs/runbook.md"), + ) + .unwrap(); + for entry in ["## brokerd-start-failed", "## brokerd-listener-lost"] { + assert!(runbook.lines().any(|l| l == entry), "{entry}"); + } +} diff --git a/docs/runbook.md b/docs/runbook.md index 4bd97bb..9fa6e15 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -163,6 +163,64 @@ does not help and is not needed. **Check.** `pgrep -a brokerd` shows one process. +## brokerd-start-failed + +**What you see.** `brokerd` exits 1 at start, before it serves anything, with one line naming its +config file, a directory or a socket, then this entry. `loopd` then reports +[broker-unavailable](#broker-unavailable) for every tool call. + +**Why.** `brokerd` could not read or parse its config, could not create or make private (0700) the +directory a socket lives in, or could not bind a socket or make it private (0600). It will not +serve on a socket whose permissions it could not set, because those permissions are what keep other +programs off it. + +**Confirm.** The line says which: + +- `: …` — the file is missing, unreadable, or not valid TOML for `brokerd.toml` + (unknown keys are errors). Check it against `docs/specs/2026-09-18-m3a-decision-path.md`, + section 2, "Configuration". +- `cannot prepare : …` — the directory cannot be made or `chmod`ed: + `ls -ld "$(dirname )"`. A path that runs through a file, or a directory owned by + another user, gives this. +- `cannot listen on : …` — `path must be shorter than SUN_LEN` means the socket path is + longer than 107 bytes; `Address already in use` means something still listens there + (`ss -xlp | grep `). + +**Fix.** Correct the config, or the ownership of the directory, or choose a shorter socket path in +`[sockets]` (and the same path in `loopd`'s `[broker] socket`). If another process holds the +socket, stop it; `brokerd` removes a stale socket file by itself. + +**Check.** `brokerd serve --config ` prints `brokerd: serving tools on … and approvals on …`. + +## brokerd-listener-lost + +**What you see.** Either `brokerd` exits 1 after it had been serving, with +`brokerd: stopped serving: ` and this entry; or it keeps running and prints +`brokerd: cannot start a thread for a connection, so it was closed`. In both cases `loopd` reports +[broker-unavailable](#broker-unavailable) for the calls that were refused. + +**Why.** The system refused `brokerd` something it needs to serve: `accept` failed on a socket for a +reason other than one aborted connection, or a thread could not be started. The usual cause is a +limit: open files (`EMFILE`), processes or threads for the user, or memory. A connection that is +refused gets no decision, so nothing runs for it. + +**Confirm.** + +```sh +ulimit -n; ulimit -u +ls /proc/$(pgrep -x brokerd)/fd | wc -l # while it runs +ps -o nlwp= -p $(pgrep -x brokerd) # its thread count +``` + +Many threads usually means many connections waiting on approvals, or a client that opens +connections and never sends: look at `bxctl approvals` and at which process holds the sockets +(`ss -xp | grep broker`). + +**Fix.** Answer or let expire the pending approvals, stop whatever is flooding the socket, or raise +the limit. Then start `brokerd` again if it exited. + +**Check.** `brokerd` prints `serving tools on …`, and a tool call is decided again. + ## broker-state-damaged **What you see.** `brokerd` prints an error reading or writing