Add brokerd serve: startup, both sockets, and the expiry thread

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/brokerd/src/serve.rs

@@ -0,0 +1,199 @@
+//! `brokerd serve`: bring the broker up, listen on both sockets, and keep it running.
+
+use std::fs::{DirBuilder, Permissions};
+use std::io;
+use std::os::unix::fs::{DirBuilderExt, PermissionsExt};
+use std::os::unix::net::UnixListener;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::mpsc;
+use std::time::Duration;
+
+use crate::admin;
+use crate::audit::{AuditError, Writer};
+use crate::broker::{self, Broker};
+use crate::config::Config;
+use crate::ledger::Ledger;
+use crate::runner::Runtime;
+use crate::state::StateStore;
+use proto::{ChainFailure, Timestamp};
+
+/// A failure while `brokerd serve` is starting up.
+#[derive(Debug)]
+pub enum ServeError {
+    /// The audit chain would not open. Displayed as the audit error's own text.
+    Audit(AuditError),
+    /// A directory could not be prepared, or made private.
+    Dir(PathBuf, io::Error),
+    /// A socket could not be bound, or made private.
+    Socket(PathBuf, io::Error),
+}
+
+impl std::fmt::Display for ServeError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ServeError::Audit(e) => write!(f, "{e}"),
+            ServeError::Dir(path, e) => write!(f, "cannot prepare {}: {e}", path.display()),
+            ServeError::Socket(path, e) => write!(f, "cannot listen on {}: {e}", path.display()),
+        }
+    }
+}
+
+impl std::error::Error for ServeError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            ServeError::Audit(e) => Some(e),
+            ServeError::Dir(_, e) | ServeError::Socket(_, e) => Some(e),
+        }
+    }
+}
+
+/// The result of `start`: a running broker with both sockets open.
+pub struct Started {
+    pub broker: Arc<Broker>,
+    pub recovered: bool,
+    pub accepted: Option<ChainFailure>,
+    tools: UnixListener,
+    admin: UnixListener,
+}
+
+/// Start the broker: open the audit chain, bind both sockets, and build the ledger and broker.
+pub fn start(
+    cfg: Config,
+    accept_break: bool,
+    runtime: Box<dyn Runtime>,
+    log: Arc<dyn Fn(&str) + Send + Sync>,
+) -> Result<Started, ServeError> {
+    // The audit lock comes first: if the chain will not open, no directory or socket is made.
+    let opened = Writer::open(&cfg.audit_dir(), accept_break).map_err(ServeError::Audit)?;
+
+    // Both sockets, in the order the task gives.
+    let tools = listen(&cfg.broker_socket())?;
+    let admin = listen(&cfg.admin_socket())?;
+
+    // The ledger and broker share the one log, each through its own box.
+    let log1: Box<dyn Fn(&str) + Send + Sync> = {
+        let log = Arc::clone(&log);
+        Box::new(move |line: &str| log(line))
+    };
+    let ledger = Ledger::new(
+        Box::new(opened.writer),
+        StateStore::new(&cfg.state_dir()),
+        log1,
+    );
+    let log2: Box<dyn Fn(&str) + Send + Sync> = {
+        let log = Arc::clone(&log);
+        Box::new(move |line: &str| log(line))
+    };
+    let broker = Broker::new(cfg, ledger, runtime, log2);
+
+    Ok(Started {
+        broker: Arc::new(broker),
+        recovered: opened.recovered,
+        accepted: opened.accepted.map(|b| *b),
+        tools,
+        admin,
+    })
+}
+
+/// Bind one socket: prepare its directory at 0700, make it private, remove any stale
+/// socket, bind, then make the socket itself private at 0600.
+fn listen(socket: &Path) -> Result<UnixListener, ServeError> {
+    let dir = match socket.parent() {
+        Some(parent) => parent.to_path_buf(),
+        None => PathBuf::from("/"),
+    };
+
+    // 1. The directory, at 0700.
+    DirBuilder::new()
+        .recursive(true)
+        .mode(0o700)
+        .create(&dir)
+        .map_err(|e| ServeError::Dir(dir.clone(), e))?;
+    // 2. Make it private, always, even when it was already there at 0755.
+    std::fs::set_permissions(&dir, Permissions::from_mode(0o700))
+        .map_err(|e| ServeError::Dir(dir.clone(), e))?;
+
+    // 3. A stale socket is removed; a missing one is fine.
+    match std::fs::remove_file(socket) {
+        Ok(()) => {}
+        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
+        Err(e) => return Err(ServeError::Socket(socket.to_path_buf(), e)),
+    }
+
+    // 4. Bind, then 5. make the socket private.
+    let listener =
+        UnixListener::bind(socket).map_err(|e| ServeError::Socket(PathBuf::from(socket), e))?;
+    std::fs::set_permissions(socket, Permissions::from_mode(0o600))
+        .map_err(|e| ServeError::Socket(PathBuf::from(socket), e))?;
+    Ok(listener)
+}
+
+/// Which handler the accept thread runs for each accepted stream.
+#[derive(Copy, Clone)]
+enum Which {
+    Tools,
+    Admin,
+}
+
+/// Run the broker: start the expiry thread, one accept thread per socket, then wait for the
+/// first listener error.
+impl Started {
+    pub fn run(self) -> io::Result<()> {
+        let Started {
+            broker,
+            tools,
+            admin,
+            ..
+        } = self;
+
+        // The expiry thread: once a second, expire the approvals whose time has run out.
+        {
+            let broker = Arc::clone(&broker);
+            std::thread::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::<io::Error>();
+
+        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() {
+            Ok(e) => Err(e),
+            Err(_) => Err(io::Error::other("a listener thread ended")),
+        }
+    }
+}
+
+/// Accept streams from one listener forever, spawning a handler thread per stream.
+fn accept_loop(
+    listener: UnixListener,
+    broker: Arc<Broker>,
+    tx: mpsc::Sender<io::Error>,
+    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);
+                }
+            }
+        }
+    });
+}