crates/gunnar-common/src/logqueue.rs
//! **The console queue**: the log lines gunnar writes to stderr go through ONE
//! bounded queue and ONE drainer thread, so no request ever waits for the
//! console to accept a byte.
//!
//! # Why (MEASURED 2026-08-30)
//!
//! On the appliance stderr is `/dev/console` = `ttyS0`, a 16550 UART that the
//! hypervisor paces at its declared 115200 baud — about 30 KB/s on the KVM
//! rig. A push narrates ~18 `INFO` lines of ~370 bytes each through the
//! `ssh.connection{…}:git.transfer{…}` span prefix, and every one of them was
//! written synchronously from the connection's own thread: `tracing`'s fmt
//! layer holds the writer while the kernel's serial driver waits for the
//! transmit register byte by byte. On the rig that was **8–50 ms per log
//! line**, between lines with no work in between; a push-create that the
//! same binary serves in 3 ms on a developer box took 161–228 ms, and eight
//! parallel clients got 6–7.6 ops/s because the UART is one device and every
//! thread queued on it. Stock git, with no console to talk to, created
//! repositories at 103 ops/s. The serial log's own byte rate over the phase
//! was 32.0 KB/s — the console was the clock.
//!
//! # What this does
//!
//! The fmt layer formats each event into a per-event buffer (the
//! [`MakeWriter`] contract: one writer per event, dropped when the event is
//! written). On drop the line is handed to a `sync_channel` of [`CAPACITY`]
//! lines and the thread carries on. A single drainer thread owns stderr and
//! writes lines in order.
//!
//! * `WARN` and `ERROR` **block until the queue has room** — never dropped.
//! The audit trail (`gunnar::audit`) is `WARN`, the dashboard's alert rows
//! are `WARN`, and one such line waiting a few milliseconds behind a
//! backlog is the right trade; a lost one is not.
//! * `INFO` and below are **dropped when the queue is full**, and counted.
//! The drainer says so on the console itself — `event="log.dropped"` with
//! the count — the next time it catches up, so a sampled console never
//! reads as a complete one. The dashboard is unaffected: it reads the
//! in-process tap ([`crate::logtap`]), not the console.
//!
//! The queue is used when stderr is **not** a terminal: PID 1 on the
//! appliance, a systemd unit under journald, a test harness with `RUST_LOG`
//! set. A human at a terminal keeps the synchronous writer: there the console
//! is fast, and `println!` output interleaving with the log in order matters
//! more than a few microseconds per line.
//!
//! # Exit
//!
//! A process that returns from `main` takes whatever is still queued with it,
//! so the two entry points call [`flush`] on their way out. It waits, bounded,
//! for the drainer to empty the queue.
use std::io::Write as _;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc::{sync_channel, Receiver, SyncSender, TrySendError};
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant};
use tracing_subscriber::fmt::MakeWriter;
/// Lines the queue holds before `INFO` starts being dropped. At ~370 bytes a
/// line this is ~1.5 MB at the very worst, preallocated by the channel once.
pub const CAPACITY: usize = 4096;
/// Lines the queue could not carry, since the process started.
static DROPPED: AtomicU64 = AtomicU64::new(0);
/// Lines that were written to the console, since the process started.
static WRITTEN: AtomicU64 = AtomicU64::new(0);
/// Lines currently queued (sent, not yet written). The flush waits on this.
static QUEUED: AtomicUsize = AtomicUsize::new(0);
static SENDER: OnceLock<SyncSender<Vec<u8>>> = OnceLock::new();
/// **The tap**: every console line handed to one more reader as it is made —
/// the VM's boot log that the installer collects (gunnar-server
/// `installer_listener`, 2026-09-24). A plain function, set once; unset, it
/// costs one atomic load per line.
static TAP: OnceLock<fn(&str)> = OnceLock::new();
/// Hand every console line to `f` as well. The first caller wins.
pub fn set_tap(f: fn(&str)) {
let _ = TAP.set(f);
}
/// **The console, when fd 2 is no longer it** ([`crate::stderr_capture`]):
/// the process's stderr became a pipe so that what libraries print with
/// `eprintln!` (monetize-server, its guests) reaches the tap as log lines,
/// and the drainer must then write to the console it was — never back into
/// the pipe it would read again.
static CONSOLE: OnceLock<std::fs::File> = OnceLock::new();
/// Name `console` as the device the drainer writes to. Before
/// [`crate::logging::init`]; the first caller wins.
pub fn set_console(console: std::fs::File) {
let _ = CONSOLE.set(console);
}
/// What the drainer writes to: the console [`set_console`] named, else stderr.
pub fn console_sink() -> Box<dyn std::io::Write + Send> {
match CONSOLE.get().and_then(|f| f.try_clone().ok()) {
Some(f) => Box::new(f),
None => Box::new(std::io::stderr()),
}
}
/// Write `text` straight to the console (a PID 1 shout), not through fd 2:
/// with stderr captured, fd 2 is the pipe and the line would be tapped twice.
pub fn console_write(text: &str) {
let _ = console_sink().write_all(text.as_bytes());
}
/// A line said OUTSIDE tracing (the boot narrator), handed to the tap too.
pub fn tap_line(line: &str) {
if let Some(f) = TAP.get() {
f(line);
}
}
/// Total `INFO`-and-below lines the console could not carry so far.
pub fn dropped() -> u64 {
DROPPED.load(Ordering::Relaxed)
}
/// Total lines written to the console so far.
pub fn written() -> u64 {
WRITTEN.load(Ordering::Relaxed)
}
/// Whether the queue is installed (stderr was not a terminal at init).
pub fn active() -> bool {
SENDER.get().is_some()
}
/// Wait, at most `timeout`, for every queued line to reach the console.
///
/// Called by the entry points on their way out; harmless when the queue is
/// not installed. Bounded because a console that has stopped accepting bytes
/// must not turn a clean exit into a hang.
pub fn flush(timeout: Duration) {
if SENDER.get().is_none() {
return;
}
let deadline = Instant::now() + timeout;
while QUEUED.load(Ordering::Acquire) > 0 && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(2));
}
let _ = std::io::stderr().flush();
}
/// Start the drainer and hand back the `MakeWriter` the fmt layer takes.
///
/// `sink` is what the drainer writes to — stderr in the process, a fake in the
/// tests. Installing twice is a no-op that returns the same queue; the second
/// sink is ignored.
pub fn install(sink: Box<dyn std::io::Write + Send>) -> QueueWriter {
let sender = SENDER.get_or_init(|| {
let (tx, rx) = sync_channel::<Vec<u8>>(CAPACITY);
// A32-14, stays a task: the console writer itself (a job's ALERT line is written through it); drained at `flush`.
std::thread::Builder::new()
.name("gunnar-console".to_owned())
.spawn(move || drain(rx, sink))
.expect("spawning the console drainer thread");
tx
});
QueueWriter {
sender: sender.clone(),
}
}
/// The drainer: one thread, owns the sink, writes lines in queue order and
/// reports the drops it catches up over.
fn drain(rx: Receiver<Vec<u8>>, mut sink: Box<dyn std::io::Write + Send>) {
let mut reported: u64 = 0;
while let Ok(line) = rx.recv() {
let _ = sink.write_all(&line);
WRITTEN.fetch_add(1, Ordering::Relaxed);
// Caught up (this was the last line queued): say how much the console
// missed, once per catch-up, on the console itself, BEFORE the count
// drops to zero — `flush` returns on zero and must see the report
// written. The line is short on purpose: it goes to the very device
// that could not keep up.
if QUEUED.load(Ordering::Acquire) == 1 {
let dropped = DROPPED.load(Ordering::Relaxed);
if dropped > reported {
let _ = writeln!(
sink,
"gunnar: the console could not carry {} log lines (INFO and below; {dropped} \
since start) — WARN rows and the dashboard are complete event=\"log.dropped\" \
dropped={} total={dropped}",
dropped - reported,
dropped - reported
);
reported = dropped;
}
let _ = sink.flush();
}
QUEUED.fetch_sub(1, Ordering::AcqRel);
}
let _ = sink.flush();
}
/// The `MakeWriter` half: hands the fmt layer one [`LineWriter`] per event.
#[derive(Clone)]
pub struct QueueWriter {
sender: SyncSender<Vec<u8>>,
}
impl<'a> MakeWriter<'a> for QueueWriter {
type Writer = LineWriter;
fn make_writer(&'a self) -> Self::Writer {
// Called without metadata only by code outside the fmt layer's event
// path; treat it as a line that must not be lost.
LineWriter {
sender: self.sender.clone(),
buf: Vec::with_capacity(512),
must_deliver: true,
}
}
fn make_writer_for(&'a self, meta: &tracing::Metadata<'_>) -> Self::Writer {
LineWriter {
sender: self.sender.clone(),
buf: Vec::with_capacity(512),
must_deliver: *meta.level() <= tracing::Level::WARN,
}
}
}
/// One formatted event. Bytes collect here; the line is queued on drop.
pub struct LineWriter {
sender: SyncSender<Vec<u8>>,
buf: Vec<u8>,
/// `WARN`/`ERROR`: wait for room rather than drop.
must_deliver: bool,
}
impl std::io::Write for LineWriter {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
self.buf.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
impl Drop for LineWriter {
fn drop(&mut self) {
if self.buf.is_empty() {
return;
}
if let Some(f) = TAP.get() {
f(&String::from_utf8_lossy(&self.buf));
}
let line = std::mem::take(&mut self.buf);
QUEUED.fetch_add(1, Ordering::AcqRel);
if self.must_deliver {
if self.sender.send(line).is_err() {
// The drainer is gone (process tearing down): nothing to do.
QUEUED.fetch_sub(1, Ordering::AcqRel);
}
return;
}
match self.sender.try_send(line) {
Ok(()) => {}
Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => {
QUEUED.fetch_sub(1, Ordering::AcqRel);
DROPPED.fetch_add(1, Ordering::Relaxed);
}
}
}
}
/// A sink for the tests: every byte written, behind a mutex, with an optional
/// per-write delay that plays the paced UART.
#[doc(hidden)]
pub struct SlowSink {
pub bytes: std::sync::Arc<Mutex<Vec<u8>>>,
pub delay: Duration,
}
impl std::io::Write for SlowSink {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
if !self.delay.is_zero() {
std::thread::sleep(self.delay);
}
self.bytes
.lock()
.unwrap_or_else(|e| e.into_inner())
.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
/// The one statement this module exists for: a console that takes 5 ms
/// per line costs the thread that logs nothing like 5 ms per line. WARN
/// lines all arrive; INFO lines past the queue's capacity are dropped and
/// counted; the drainer reports the drop on the console.
#[test]
fn a_slow_console_never_stalls_the_logger_and_loses_no_warn() {
let bytes = Arc::new(Mutex::new(Vec::new()));
let writer = install(Box::new(SlowSink {
bytes: Arc::clone(&bytes),
delay: Duration::from_millis(5),
}));
let info = tracing::Metadata::new(
"t",
"t",
tracing::Level::INFO,
None,
None,
None,
tracing::field::FieldSet::new(&[], tracing::callsite::Identifier(&CALLSITE)),
tracing::metadata::Kind::EVENT,
);
let warn = tracing::Metadata::new(
"t",
"t",
tracing::Level::WARN,
None,
None,
None,
tracing::field::FieldSet::new(&[], tracing::callsite::Identifier(&CALLSITE)),
tracing::metadata::Kind::EVENT,
);
let n = CAPACITY * 3;
let started = Instant::now();
for i in 0..n {
let mut w = writer.make_writer_for(&info);
writeln!(w, "info line {i}").unwrap();
}
let info_elapsed = started.elapsed();
for i in 0..8 {
let mut w = writer.make_writer_for(&warn);
writeln!(w, "WARN line {i}").unwrap();
}
// 12 288 INFO lines at 5 ms each would be a minute; the logger must be
// done in well under one second.
assert!(
info_elapsed < Duration::from_secs(1),
"logging {n} INFO lines to a 5 ms/line console took {info_elapsed:?}: the logger waited on the console"
);
let dropped_now = dropped();
assert!(
dropped_now > 0 && dropped_now < n as u64,
"expected some but not all INFO lines dropped past a {CAPACITY}-line queue, got {dropped_now} of {n}"
);
flush(Duration::from_secs(120));
let out = String::from_utf8_lossy(&bytes.lock().unwrap()).into_owned();
for i in 0..8 {
assert!(
out.contains(&format!("WARN line {i}\n")),
"WARN line {i} was lost"
);
}
assert!(
out.contains("event=\"log.dropped\""),
"the drainer must report on the console what the console missed"
);
assert_eq!(
written() as usize,
n - dropped_now as usize + 8,
"every line not dropped was written"
);
}
struct Callsite;
impl tracing::Callsite for Callsite {
fn set_interest(&self, _: tracing::subscriber::Interest) {}
fn metadata(&self) -> &tracing::Metadata<'_> {
unreachable!("the test never registers this callsite")
}
}
static CALLSITE: Callsite = Callsite;
}