brokerd: recover a torn line in place, real dates only, bounded ttl, EMFILE

From the independent review of task 23. A torn last line followed by an empty
later file had its recovery written into the later file, which broke the
chain for good; the line is now ended in its own file. The log-name rule
takes months 01 to 12 and days 01 to 31 only. [approvals] ttl_ms is limited
to a day, the longest loopd waits after a pending frame. Running out of file
descriptors or memory pauses the listener instead of stopping brokerd (the
errors the previous fix skipped do not occur on Linux). args.rs's doc fixed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

crates/brokerd/src/args.rs

@@ -6,8 +6,8 @@
 
 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.
 pub const MAX_URL: usize = 2048;

crates/brokerd/src/audit.rs

@@ -191,12 +191,35 @@ fn tail_of_file_before_latest(dir: &Path) -> Result<Hash32, AuditError> {
     }
 }
 
+/// End the last line of the log with a newline, in the file that holds it: the last log file that
+/// is not empty. A torn line is always there, even when an empty later file exists, and it must be
+/// ended in place, or the next record would join it on one line.
+fn end_last_line(dir: &Path) -> Result<(), AuditError> {
+    for name in log_files(dir)?.iter().rev() {
+        let path = dir.join(name);
+        let bytes = fs::read(&path).map_err(|e| io("read audit log", &path, e))?;
+        match bytes.last() {
+            None => continue,
+            Some(b'\n') => return Ok(()),
+            Some(_) => {
+                let mut file = fs::OpenOptions::new()
+                    .append(true)
+                    .open(&path)
+                    .map_err(|e| io("open audit log", &path, e))?;
+                file.write_all(b"\n").map_err(io::Error::other)?;
+                file.sync_all().map_err(io::Error::other)?;
+                return Ok(());
+            }
+        }
+    }
+    Ok(())
+}
+
 /// Append one record: serialise it, open the target file, write the line, and return the seq and
 /// the hash of the line it wrote. A fresh file is created mode 0600 and the directory synced.
 fn write_record(
     dir: &Path,
     target: &str,
-    newline_first: bool,
     seq: u64,
     prev: Hash32,
     event: AuditEvent,
@@ -216,9 +239,6 @@ fn write_record(
         .mode(0o600)
         .open(&path)
         .map_err(|e| io("open audit log", &path, e))?;
-    if newline_first {
-        file.write_all(b"\n").map_err(io::Error::other)?;
-    }
     file.write_all(json.as_bytes()).map_err(io::Error::other)?;
     file.write_all(b"\n").map_err(io::Error::other)?;
     file.sync_all().map_err(io::Error::other)?;
@@ -305,10 +325,12 @@ impl Writer {
             if !accept_break {
                 return Err(AuditError::Broken(Box::new(failure.clone())));
             }
+            if failure.tail_torn {
+                end_last_line(dir)?;
+            }
             let (seq, hash) = write_record(
                 dir,
                 latest.as_deref().unwrap_or(&day_name(now)),
-                failure.tail_torn,
                 failure.break_seq,
                 failure.break_prev,
                 AuditEvent::AcceptedBreak {
@@ -322,10 +344,12 @@ impl Writer {
         } else if accept_break {
             return Err(AuditError::NothingToAccept);
         } else if let Some(torn) = report.torn_tail.as_ref() {
+            if !torn.has_newline {
+                end_last_line(dir)?;
+            }
             let (seq, hash) = write_record(
                 dir,
                 latest.as_deref().unwrap_or(&day_name(now)),
-                !torn.has_newline,
                 torn.recovery_seq,
                 torn.recovery_prev,
                 AuditEvent::Recovery {
@@ -366,15 +390,8 @@ impl Writer {
             return Err(AuditError::Stopped);
         }
         let target = target_for(&self.latest, time);
-        let (seq, hash) = match write_record(
+        let (seq, hash) =
-            &self.path,
+            match write_record(&self.path, &target, self.next_seq, self.prev, event, time) {
-            &target,
-            false,
-            self.next_seq,
-            self.prev,
-            event,
-            time,
-        ) {
                 Ok(r) => r,
                 Err(e) => {
                     self.stopped = true;

crates/brokerd/src/config.rs

@@ -56,10 +56,15 @@ pub struct Config {
     pub approvals: Approvals,
 }
 
+/// The longest an approval may wait: a day, the longest `loopd` waits after a pending frame.
+pub const MAX_TTL_MS: u64 = 86_400_000;
+
 #[derive(Debug)]
 pub enum ConfigError {
     Read(PathBuf, std::io::Error),
     Parse(PathBuf, toml::de::Error),
+    /// The file parses, but a value is outside what brokerd accepts.
+    Invalid(PathBuf, String),
 }
 
 impl std::fmt::Display for ConfigError {
@@ -67,6 +72,7 @@ impl std::fmt::Display for ConfigError {
         match self {
             ConfigError::Read(path, err) => write!(f, "{}: {err}", path.display()),
             ConfigError::Parse(path, err) => write!(f, "{}: {err}", path.display()),
+            ConfigError::Invalid(path, why) => write!(f, "{}: {why}", path.display()),
         }
     }
 }
@@ -82,6 +88,17 @@ impl Config {
             std::fs::read_to_string(path).map_err(|e| ConfigError::Read(path.to_path_buf(), e))?;
         let config: Config =
             toml::from_str(&text).map_err(|e| ConfigError::Parse(path.to_path_buf(), e))?;
+        // `loopd` waits at most a day after a pending frame, so a longer approval would be given up
+        // by the caller while `bxctl approvals` still listed it.
+        if !(1..=MAX_TTL_MS).contains(&config.approvals.ttl_ms) {
+            return Err(ConfigError::Invalid(
+                path.to_path_buf(),
+                format!(
+                    "[approvals] ttl_ms is {}; it must be from 1 to {MAX_TTL_MS} (a day)",
+                    config.approvals.ttl_ms
+                ),
+            ));
+        }
         Ok(config)
     }
     pub fn broker_socket(&self) -> PathBuf {

crates/brokerd/src/serve.rs

@@ -197,9 +197,10 @@ impl Started {
     }
 }
 
-/// Accept streams from one listener forever, starting a handler thread per stream. An error that
+/// Accept streams from one listener forever, starting a handler thread per stream. When the
-/// only concerns one connection (the peer aborted it, a signal interrupted `accept`) is skipped;
+/// system is out of file descriptors or memory, the listener pauses and tries again instead of
-/// any other stops the daemon through `tx`.
+/// stopping: otherwise a peer holding many idle connections could stop the daemon, and the other
+/// socket with it. Any other error stops the daemon through `tx`.
 fn accept_loop(
     listener: UnixListener,
     broker: Arc<Broker>,
@@ -213,10 +214,22 @@ fn accept_loop(
     std::thread::Builder::new()
         .name(name.to_string())
         .spawn(move || {
+            let mut reported = false;
             for stream in listener.incoming() {
                 match stream {
-                    Ok(stream) => serve_one(stream, &broker, which),
+                    Ok(stream) => {
-                    Err(e) if one_connection_only(&e) => {}
+                        reported = false;
+                        serve_one(stream, &broker, which);
+                    }
+                    Err(e) if out_of_resources(&e) => {
+                        if !reported {
+                            broker.log(&format!(
+                                "brokerd: cannot accept on {name} for now, retrying: {e}\nsee docs/runbook.md#brokerd-listener-lost"
+                            ));
+                            reported = true;
+                        }
+                        std::thread::sleep(Duration::from_millis(200));
+                    }
                     Err(e) => {
                         // Once `run` has returned, nobody receives; dropping this is correct.
                         let _ = tx.send(e);
@@ -243,10 +256,11 @@ fn serve_one(stream: UnixStream, broker: &Arc<Broker>, which: Which) {
     }
 }
 
-/// Errors from `accept` that end one connection, not the listener.
+/// Errors from `accept` that mean "not now" rather than "never": out of file descriptors
-fn one_connection_only(e: &io::Error) -> bool {
+/// (`EMFILE`, `ENFILE`, the same numbers on Linux and macOS) or out of memory. `accept` already
-    matches!(
+/// retries `EINTR` itself.
-        e.kind(),
+fn out_of_resources(e: &io::Error) -> bool {
-        io::ErrorKind::ConnectionAborted | io::ErrorKind::Interrupted | io::ErrorKind::WouldBlock
+    const ENFILE: i32 = 23;
-    )
+    const EMFILE: i32 = 24;
+    matches!(e.raw_os_error(), Some(ENFILE | EMFILE)) || e.kind() == io::ErrorKind::OutOfMemory
 }

crates/brokerd/tests/audit_edges.rs

@@ -104,3 +104,31 @@ fn the_lock_file_outlives_the_writer() {
     assert!(dir.path.join(".lock").exists());
     Writer::open(&dir.path, false).expect("the lock was released with the writer");
 }
+
+/// A torn last line in one file, then an empty later file (created, never written). The torn line
+/// is ended in its own file and the recovery goes on the chain after it, so the log verifies and
+/// the next start is an ordinary one. Found by the independent review of task 23.
+#[test]
+fn a_torn_line_before_an_empty_later_file_is_recovered_in_place() {
+    let dir = TempDir::unmade("torn-then-empty");
+    let mut writer = Writer::open(&dir.path, false).unwrap().writer;
+    writer
+        .append(ts("2026-09-17T08:00:00.000Z"), denied(1))
+        .unwrap();
+    drop(writer);
+    let mut d1 = std::fs::read(dir.path.join(D1)).unwrap();
+    d1.extend_from_slice(br#"{"seq":1,"time":"2026-09-17T09"#);
+    std::fs::write(dir.path.join(D1), &d1).unwrap();
+    std::fs::write(dir.path.join(D2), b"").unwrap();
+
+    let opened = Writer::open(&dir.path, false).unwrap();
+    assert!(opened.recovered);
+    drop(opened);
+    let report = verify_dir(&dir.path).unwrap();
+    assert_eq!(report.failure, None, "{:?}", report.failure);
+    assert_eq!(report.recoveries.len(), 1);
+    assert!(std::fs::read(dir.path.join(D1)).unwrap().ends_with(b"\n"));
+
+    let again = Writer::open(&dir.path, false).expect("the next start is an ordinary one");
+    assert!(!again.recovered);
+}

crates/brokerd/tests/serve_pointers.rs

@@ -177,3 +177,27 @@ fn the_entries_exist() {
         assert!(runbook.lines().any(|l| l == entry), "{entry}");
     }
 }
+
+/// `[approvals] ttl_ms` is between 1 ms and a day: `loopd` waits at most a day after a pending
+/// frame, so a longer approval would be abandoned while `bxctl` still listed it.
+#[test]
+fn an_approval_ttl_outside_a_day_is_a_config_error() {
+    for ttl in ["0", "86400001", "18446744073709551615"] {
+        let dir = TempDir::new("ptr-ttl");
+        std::fs::create_dir_all(dir.path().join("grants")).unwrap();
+        let path = dir.write(
+            "brokerd.toml",
+            &format!(
+                "[paths]\nhome = \"{0}\"\ngrants = \"{0}/grants\"\n[approvals]\nttl_ms = {ttl}\n",
+                dir.path().display()
+            ),
+        );
+        let out = brokerd([
+            OsStr::new("serve"),
+            OsStr::new("--config"),
+            path.as_os_str(),
+        ]);
+        fails_with_pointer(&out);
+        assert!(stderr(&out).contains("ttl_ms"), "{ttl}: {}", stderr(&out));
+    }
+}

crates/proto/src/audit.rs

@@ -93,14 +93,20 @@ pub struct AuditRecord {
 }
 
 /// True if `name` is an audit log file: `YYYY-MM-DD.jsonl`, ASCII digits with `-` at positions 4
-/// and 7. `brokerd` and `bxctl audit verify` both decide with this, so they read the same files.
+/// and 7, a month from 01 to 12 and a day from 01 to 31. `brokerd` and `bxctl audit verify` both
+/// decide with this, so they read the same files.
 pub fn is_audit_log_name(name: &str) -> bool {
     let Some(date) = name.strip_suffix(".jsonl") else {
         return false;
     };
-    date.len() == 10
+    let shaped = date.len() == 10
         && date.bytes().enumerate().all(|(i, b)| match i {
             4 | 7 => b == b'-',
             _ => b.is_ascii_digit(),
-        })
+        });
+    if !shaped {
+        return false;
+    }
+    let number = |range: std::ops::Range<usize>| date.get(range).and_then(|s| s.parse::<u8>().ok());
+    matches!(number(5..7), Some(1..=12)) && matches!(number(8..10), Some(1..=31))
 }

crates/proto/tests/log_names.rs

@@ -5,7 +5,7 @@ use proto::is_audit_log_name;
 
 #[test]
 fn a_date_and_jsonl_is_a_log_file() {
-    for name in ["2026-09-17.jsonl", "0000-00-00.jsonl", "9999-12-31.jsonl"] {
+    for name in ["2026-09-17.jsonl", "0000-01-01.jsonl", "9999-12-31.jsonl"] {
         assert!(is_audit_log_name(name), "{name}");
     }
 }
@@ -27,6 +27,11 @@ fn every_other_name_is_not() {
         "12026-09-17.jsonl",
         "2026-09-17.JSONL",
         "２026-09-17.jsonl", // a full-width digit is not an ASCII digit
+        "2026-00-17.jsonl",  // no month 0
+        "2026-13-17.jsonl",
+        "2026-09-00.jsonl", // no day 0
+        "2026-09-32.jsonl",
+        "2026-99-99.jsonl",
     ] {
         assert!(!is_audit_log_name(name), "{name}");
     }

docs/runbook.md

@@ -198,13 +198,17 @@ socket, stop it; `brokerd` removes a stale socket file by itself.
 
 ## brokerd-listener-lost
 
-**What you see.** Either `brokerd` exits 1 after it had been serving, with
+**What you see.** One of three lines, then this entry:
-`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
+- `brokerd: cannot accept on accept-broker (or accept-admin) for now, retrying: <error>`. It keeps
+  running and tries every 200 ms; the line is printed once per episode.
+- `brokerd: cannot start a thread for a connection, so it was closed`. It keeps running.
+- `brokerd: stopped serving: <error>`, and it exits 1: any other failure of `accept`.
+ 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
+**Why.** The system refused `brokerd` something it needs to serve: a new connection (`accept`
-reason other than one aborted connection, or a thread could not be started. The usual cause is a
+failed) or a thread. 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.