boxmaker/docs/plans/M3a/15-brokerd-serve.md

M3a task 15: brokerd serve

Branch: m3a (run git switch m3a; git status --short must be empty, otherwise stop) Commit subject: Add brokerd serve: startup, both sockets, and the expiry thread

Goal

brokerd serve --config <path> [--accept-break] takes the audit lock, checks the log, makes its two socket directories private, binds broker.sock and admin.sock, and serves each connection on its own thread, with one more thread expiring approvals every second. The runtime is Refusing (task 10). The audit lock comes first: it is what proves that a socket file left behind is stale and not another brokerd's. serve is the only module that starts threads.

Files

• Copy: crates/brokerd/tests/serve.rs
• Create: crates/brokerd/src/serve.rs
• Modify: crates/brokerd/src/main.rs (it holds the "not implemented" stub; replace all of it), crates/brokerd/src/lib.rs (add pub mod serve;), docs/implementer-log.md

Interfaces

#[derive(Debug)]
pub enum ServeError {
    Audit(AuditError),                   // Display: the AuditError's text
    Dir(PathBuf, std::io::Error),        // "cannot prepare {path}: {e}"
    Socket(PathBuf, std::io::Error),     // "cannot listen on {path}: {e}"
}                                        // + std::error::Error

pub struct Started {
    pub broker: Arc<Broker>,
    pub recovered: bool,                         // from audit::Opened
    pub accepted: Option<proto::ChainFailure>,   // from audit::Opened
    /* private: tools: UnixListener, admin: UnixListener */
}
pub fn start(cfg: Config, accept_break: bool, runtime: Box<dyn Runtime>,
             log: Arc<dyn Fn(&str) + Send + Sync>) -> Result<Started, ServeError>;
impl Started { pub fn run(self) -> std::io::Result<()>; }

start, in this order

1. audit::Writer::open(&cfg.audit_dir(), accept_break); Err(e) → Err(Audit(e)). Nothing else has happened yet: no directory, no socket.
2. listen(&cfg.broker_socket()), then listen(&cfg.admin_socket()); each error returned.
3. Ledger::new(Box::new(opened.writer), StateStore::new(&cfg.state_dir()), log1) and Broker::new(cfg, ledger, runtime, log2), where log1 and log2 are boxes that call the one Arc (let l = Arc::clone(&log); Box::new(move |line| l(line))).

listen(socket), every exit (the directory is socket.parent(), or / if it has none):

1. DirBuilder::new().recursive(true).mode(0o700).create(dir); Err → Dir.
2. std::fs::set_permissions(dir, Permissions::from_mode(0o700)) always, also when the directory was already there at 0755; Err → Dir.
3. std::fs::remove_file(socket): Ok or NotFound go on; any other Err → Socket.
4. UnixListener::bind(socket); Err → Socket.
5. set_permissions(socket, from_mode(0o600)); Err → Socket. Return the listener.

run

1. Spawn the expiry thread: forever, sleep(1 s), then admin::expire_due(&broker, Timestamp::now()).
2. Spawn one accept thread per listener. Each does: for every incoming() stream, stream?, then spawn a thread calling broker::handle (tools) or admin::handle (admin) with its own Arc clone. When incoming gives an error, the accept thread sends that Err on an mpsc channel.
3. run waits for the first message on that channel and returns it. Either listener failing stops the daemon. (If the channel is closed, return Err(io::Error::other("a listener thread ended")).) A send after the first has nobody to receive it; ignoring that result is correct, and a comment should say so.

main.rs

Usage is exactly usage: brokerd serve --config <path> [--accept-break]. The first argument must be serve; then --config <path> exactly once and --accept-break at most once, in any order. Anything else (no arguments, a missing value, a flag twice, an unknown word) → print the usage to stderr, exit 2.

1. Config::load(&path); Err(e) → brokerd: {e}, exit 1.
2. start(cfg, accept_break, Box::new(Refusing), Arc::new(|line: &str| eprintln!("{line}"))) (keep cfg.broker_socket() and cfg.admin_socket() first, for step 4). Err(ServeError::Audit(AuditError::NothingToAccept)) → brokerd: {e}, exit 2. Any other Err(e) → brokerd: {e}, exit 1. (AuditError's text already ends with its runbook pointer.)
3. recovered → eprintln!("{RECOVERED_NOTICE}"). accepted: Some(f) → audit: accepted the break at {file}:{line}: {what}.
4. brokerd: serving tools on {broker socket} and approvals on {admin socket}.
5. run(): Err(e) → brokerd: {e}, exit 1.

Steps

• 1. Copy. git switch m3a, then cp docs/plans/M3a/files/crates/brokerd/tests/serve.rs crates/brokerd/tests/
• 2. See the test fail. cargo test -p brokerd --test serve. Expected: it compiles (it runs the binary) and all 9 fail, because brokerd still says "not implemented"; one of them waits ten seconds for sockets that never appear.
• 3. Write serve.rs, add pub mod serve;, rewrite main.rs. Run cargo fmt --all.
• 4. See the tests pass. cargo test -p brokerd --test serve, five times. Expected: 9 passed every time.
• 5. Run all of brokerd. cargo test -p brokerd. Every suite passes.
• 6. Run the gate. make gate. Expected last line: gate: ok.
• 7. Log and commit. git add crates/brokerd docs/implementer-log.md && git commit

Done when

• serve reports 9 passed five runs in a row, and make gate prints gate: ok.

Stop and report if

• A test wants a socket bound, or a directory made, before the audit lock is taken.
• You need libc, unsafe or a signal handler. brokerd is stopped by being killed; the audit log is safe at any moment because every record is synced before anything acts on it.