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) <noreply@anthropic.com>
This commit is contained in:
2026-09-22 21:07:59 -07:00
co-authored by Claude Opus 5.5
parent eed0a221ca
commit 6af89f7c60
7 changed files with 284 additions and 50 deletions
+1 -1
View File
@@ -6,7 +6,7 @@
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::fmt; use std::fmt;
/// Maximum length, in bytes, of a path.
/// Maximum length, in bytes, of a path or of a URL. /// Maximum length, in bytes, of a path or of a URL.
pub const MAX_PATH: usize = 4096; pub const MAX_PATH: usize = 4096;
/// Maximum length, in bytes, of a URL. /// Maximum length, in bytes, of a URL.
+1 -1
View File
@@ -301,7 +301,7 @@ fn pending(
ToolResponse::Failed { ToolResponse::Failed {
message: GONE.to_string(), 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 return None; // 5. the peer left at the last look; nothing ran
} }
+3 -3
View File
@@ -91,8 +91,7 @@ pub fn load(dir: &Path) -> Result<GrantSet, Vec<GrantProblem>> {
match entry { match entry {
Ok(entry) => names.push(entry.file_name().to_string_lossy().to_string()), Ok(entry) => names.push(entry.file_name().to_string_lossy().to_string()),
Err(error) => { Err(error) => {
// An entry that cannot be read is the same problem as the directory; keep going. // An entry that cannot be read is the same problem as the directory.
names.push(String::new());
return Err(vec![GrantProblem { return Err(vec![GrantProblem {
file: dir.to_string_lossy().to_string(), file: dir.to_string_lossy().to_string(),
line: None, line: None,
@@ -198,7 +197,8 @@ fn span_line(text: &str, error: &toml::de::Error) -> u64 {
Some(before) => before, Some(before) => before,
None => text, 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, None => 1,
} }
+31 -20
View File
@@ -1,5 +1,6 @@
//! `brokerd`: the broker role. `serve` takes the audit lock, binds both sockets, and serves. //! `brokerd`: the broker role. `serve` takes the audit lock, binds both sockets, and serves.
use std::ffi::OsString;
use std::path::PathBuf; use std::path::PathBuf;
use std::process::ExitCode; use std::process::ExitCode;
use std::sync::Arc; use std::sync::Arc;
@@ -10,10 +11,15 @@ use brokerd::runner::Refusing;
use brokerd::serve::{self, ServeError}; use brokerd::serve::{self, ServeError};
const USAGE: &str = "usage: brokerd serve --config <path> [--accept-break]"; const USAGE: &str = "usage: brokerd serve --config <path> [--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 { fn main() -> ExitCode {
// The first argument must be `serve`. // 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() { match args.next() {
Some(command) if command == "serve" => {} Some(command) if command == "serve" => {}
_ => return usage(), _ => return usage(),
@@ -29,7 +35,7 @@ fn main() -> ExitCode {
let cfg = match Config::load(&config_path) { let cfg = match Config::load(&config_path) {
Ok(cfg) => cfg, Ok(cfg) => cfg,
Err(e) => { Err(e) => {
eprintln!("brokerd: {e}"); eprintln!("brokerd: {e}\n{START_FAILED}");
return ExitCode::from(1); return ExitCode::from(1);
} }
}; };
@@ -49,10 +55,15 @@ fn main() -> ExitCode {
eprintln!("brokerd: {e}"); eprintln!("brokerd: {e}");
return ExitCode::from(2); 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}"); eprintln!("brokerd: {e}");
return ExitCode::from(1); return ExitCode::from(1);
} }
Err(e) => {
eprintln!("brokerd: {e}\n{START_FAILED}");
return ExitCode::from(1);
}
}; };
if started.recovered { if started.recovered {
@@ -73,7 +84,7 @@ fn main() -> ExitCode {
match started.run() { match started.run() {
Ok(()) => ExitCode::SUCCESS, Ok(()) => ExitCode::SUCCESS,
Err(e) => { Err(e) => {
eprintln!("brokerd: {e}"); eprintln!("brokerd: stopped serving: {e}\n{LISTENER_LOST}");
ExitCode::from(1) ExitCode::from(1)
} }
} }
@@ -86,8 +97,10 @@ fn usage() -> ExitCode {
} }
/// The `--config` and `--accept-break` flags after `serve`, or `Err(())` for a bad list. /// The `--config` and `--accept-break` flags after `serve`, or `Err(())` for a bad list.
fn parse_serve<I: Iterator<Item = String>>(mut args: I) -> Result<(PathBuf, bool), ()> { /// Flags are compared as `OsStr`, so one that is not UTF-8 is simply unknown; the path is kept as
let mut config: Option<String> = None; /// given.
fn parse_serve<I: Iterator<Item = OsString>>(mut args: I) -> Result<(PathBuf, bool), ()> {
let mut config: Option<OsString> = None;
let mut accept_break = false; let mut accept_break = false;
loop { loop {
match args.next() { match args.next() {
@@ -97,24 +110,22 @@ fn parse_serve<I: Iterator<Item = String>>(mut args: I) -> Result<(PathBuf, bool
None => Err(()), None => Err(()),
}; };
} }
Some(arg) => match arg.as_str() { Some(arg) if arg == "--config" => match args.next() {
"--config" => match args.next() { Some(path) => {
Some(path) => { if config.is_some() {
if config.is_some() {
return Err(());
}
config = Some(path);
}
None => return Err(()),
},
"--accept-break" => {
if accept_break {
return Err(()); 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(()),
} }
} }
} }
+56 -25
View File
@@ -3,7 +3,7 @@
use std::fs::{DirBuilder, Permissions}; use std::fs::{DirBuilder, Permissions};
use std::io; use std::io;
use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; 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::path::{Path, PathBuf};
use std::sync::Arc; use std::sync::Arc;
use std::sync::mpsc; use std::sync::mpsc;
@@ -148,21 +148,24 @@ impl Started {
} = self; } = self;
// The expiry thread: once a second, expire the approvals whose time has run out. // 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); let broker = Arc::clone(&broker);
std::thread::spawn(move || { std::thread::Builder::new()
loop { .name("expiry".to_string())
std::thread::sleep(Duration::from_secs(1)); .spawn(move || {
admin::expire_due(&broker, Timestamp::now()); 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. // One accept thread per socket, each with its own clone of the broker.
let (tx, rx) = mpsc::channel::<io::Error>(); let (tx, rx) = mpsc::channel::<io::Error>();
accept_loop(tools, Arc::clone(&broker), tx.clone(), Which::Tools); accept_loop(tools, Arc::clone(&broker), tx.clone(), Which::Tools)?;
accept_loop(admin, Arc::clone(&broker), tx, Which::Admin); accept_loop(admin, Arc::clone(&broker), tx, Which::Admin)?;
// The first listener error stops the daemon. // The first listener error stops the daemon.
match rx.recv() { 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( fn accept_loop(
listener: UnixListener, listener: UnixListener,
broker: Arc<Broker>, broker: Arc<Broker>,
tx: mpsc::Sender<io::Error>, tx: mpsc::Sender<io::Error>,
which: Which, which: Which,
) { ) -> io::Result<()> {
std::thread::spawn(move || { let name = match which {
for stream in listener.incoming() { Which::Tools => "accept-broker",
match stream { Which::Admin => "accept-admin",
Ok(stream) => { };
let broker = Arc::clone(&broker); std::thread::Builder::new()
std::thread::spawn(move || match which { .name(name.to_string())
Which::Tools => broker::handle(stream, &broker), .spawn(move || {
Which::Admin => admin::handle(stream, &broker), for stream in listener.incoming() {
}); match stream {
} Ok(stream) => serve_one(stream, &broker, which),
Err(e) => { Err(e) if one_connection_only(&e) => {}
// Once `run` has returned, nobody receives; dropping this is correct. Err(e) => {
let _ = tx.send(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<Broker>, 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
)
} }
+134
View File
@@ -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<I: IntoIterator<Item = S>, S: AsRef<OsStr>>(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}");
}
}
+58
View File
@@ -163,6 +163,64 @@ does not help and is not needed.
**Check.** `pgrep -a brokerd` shows one process. **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:
- `<config path>: …` — 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 <dir>: …` — the directory cannot be made or `chmod`ed:
`ls -ld <dir> "$(dirname <dir>)"`. A path that runs through a file, or a directory owned by
another user, gives this.
- `cannot listen on <socket>: …``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 <socket>`).
**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 <path>` 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: <error>` 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 ## broker-state-damaged
**What you see.** `brokerd` prints an error reading or writing **What you see.** `brokerd` prints an error reading or writing