Add the ledger: the audit writer and session state behind one lock

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

crates/brokerd/tests/support/sink.rs

@@ -0,0 +1,78 @@
+//! An audit sink that fails on demand, and a log that tests can read. Do not edit.
+//!
+//! Included with `#[path = "support/sink.rs"] mod sink;`.
+
+#![allow(dead_code)] // each test file uses a different part of this module
+
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+
+use brokerd::audit::{AuditError, Writer};
+use brokerd::ledger::AuditSink;
+use proto::{AuditEvent, Timestamp};
+
+/// Switches shared between a test and its `Flaky` sink.
+#[derive(Clone, Default)]
+pub struct Switch {
+    fail: Arc<AtomicBool>,
+    panic: Arc<AtomicBool>,
+    attempts: Arc<AtomicUsize>,
+}
+
+impl Switch {
+    /// Every append from now on fails, without writing anything.
+    pub fn fail(&self, on: bool) {
+        self.fail.store(on, Ordering::SeqCst);
+    }
+    /// The next append panics, as a bug part-way through a write would.
+    pub fn panic_next(&self) {
+        self.panic.store(true, Ordering::SeqCst);
+    }
+    /// How many appends the ledger has asked for.
+    pub fn attempts(&self) -> usize {
+        self.attempts.load(Ordering::SeqCst)
+    }
+}
+
+/// A real `Writer` behind a switch.
+pub struct Flaky {
+    pub writer: Writer,
+    pub switch: Switch,
+}
+
+impl AuditSink for Flaky {
+    fn append(&mut self, time: Timestamp, event: AuditEvent) -> Result<u64, AuditError> {
+        self.switch.attempts.fetch_add(1, Ordering::SeqCst);
+        if self.switch.panic.swap(false, Ordering::SeqCst) {
+            panic!("a bug part-way through a write");
+        }
+        if self.switch.fail.load(Ordering::SeqCst) {
+            return Err(AuditError::Io {
+                what: "cannot write to the test log".to_string(),
+                source: std::io::Error::other("the disk is full"),
+            });
+        }
+        self.writer.append(time, event)
+    }
+}
+
+/// Collects every line a ledger or broker prints.
+#[derive(Clone, Default)]
+pub struct Lines(Arc<Mutex<Vec<String>>>);
+
+impl Lines {
+    pub fn sink(&self) -> Box<dyn Fn(&str) + Send + Sync> {
+        let lines = Arc::clone(&self.0);
+        Box::new(move |line| lines.lock().unwrap().push(line.to_string()))
+    }
+    pub fn all(&self) -> Vec<String> {
+        self.0.lock().unwrap().clone()
+    }
+    /// The lines that hold `text`.
+    pub fn with(&self, text: &str) -> Vec<String> {
+        self.all()
+            .into_iter()
+            .filter(|l| l.contains(text))
+            .collect()
+    }
+}