diff --git a/crates/brokerd/src/args.rs b/crates/brokerd/src/args.rs index d6ba64c..657b9c6 100644 --- a/crates/brokerd/src/args.rs +++ b/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; diff --git a/crates/brokerd/src/audit.rs b/crates/brokerd/src/audit.rs index 95c1080..124e28e 100644 --- a/crates/brokerd/src/audit.rs +++ b/crates/brokerd/src/audit.rs @@ -191,12 +191,35 @@ fn tail_of_file_before_latest(dir: &Path) -> Result { } } +/// 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,21 +390,14 @@ impl Writer { return Err(AuditError::Stopped); } let target = target_for(&self.latest, time); - let (seq, hash) = match write_record( - &self.path, - &target, - false, - self.next_seq, - self.prev, - event, - time, - ) { - Ok(r) => r, - Err(e) => { - self.stopped = true; - return Err(e); - } - }; + let (seq, hash) = + match write_record(&self.path, &target, self.next_seq, self.prev, event, time) { + Ok(r) => r, + Err(e) => { + self.stopped = true; + return Err(e); + } + }; self.prev = hash; self.next_seq = seq.checked_add(1).unwrap_or(self.next_seq); if self.latest.as_deref() != Some(target.as_str()) { diff --git a/crates/brokerd/src/config.rs b/crates/brokerd/src/config.rs index 70f7a70..9d51e3e 100644 --- a/crates/brokerd/src/config.rs +++ b/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 { diff --git a/crates/brokerd/src/serve.rs b/crates/brokerd/src/serve.rs index 09379d3..b07eb10 100644 --- a/crates/brokerd/src/serve.rs +++ b/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 -/// only concerns one connection (the peer aborted it, a signal interrupted `accept`) is skipped; -/// any other stops the daemon through `tx`. +/// Accept streams from one listener forever, starting a handler thread per stream. When the +/// system is out of file descriptors or memory, the listener pauses and tries again instead of +/// 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, @@ -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), - Err(e) if one_connection_only(&e) => {} + Ok(stream) => { + 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, which: Which) { } } -/// 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 - ) +/// Errors from `accept` that mean "not now" rather than "never": out of file descriptors +/// (`EMFILE`, `ENFILE`, the same numbers on Linux and macOS) or out of memory. `accept` already +/// retries `EINTR` itself. +fn out_of_resources(e: &io::Error) -> bool { + const ENFILE: i32 = 23; + const EMFILE: i32 = 24; + matches!(e.raw_os_error(), Some(ENFILE | EMFILE)) || e.kind() == io::ErrorKind::OutOfMemory } diff --git a/crates/brokerd/tests/audit_edges.rs b/crates/brokerd/tests/audit_edges.rs index 8864423..778606f 100644 --- a/crates/brokerd/tests/audit_edges.rs +++ b/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); +} diff --git a/crates/brokerd/tests/serve_pointers.rs b/crates/brokerd/tests/serve_pointers.rs index 1fe5ff3..d88e853 100644 --- a/crates/brokerd/tests/serve_pointers.rs +++ b/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)); + } +} diff --git a/crates/proto/src/audit.rs b/crates/proto/src/audit.rs index 4d4c827..7aa090f 100644 --- a/crates/proto/src/audit.rs +++ b/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| date.get(range).and_then(|s| s.parse::().ok()); + matches!(number(5..7), Some(1..=12)) && matches!(number(8..10), Some(1..=31)) } diff --git a/crates/proto/tests/log_names.rs b/crates/proto/tests/log_names.rs index 0fb902f..9a56614 100644 --- a/crates/proto/tests/log_names.rs +++ b/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}"); } diff --git a/docs/runbook.md b/docs/runbook.md index a81a45e..f128ac0 100644 --- a/docs/runbook.md +++ b/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 -`brokerd: stopped serving: ` 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 +**What you see.** One of three lines, then this entry: + +- `brokerd: cannot accept on accept-broker (or accept-admin) for now, retrying: `. 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: `, 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 -reason other than one aborted connection, or a thread could not be started. The usual cause is a +**Why.** The system refused `brokerd` something it needs to serve: a new connection (`accept` +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.