# 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

```rust
#[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.