gunnar_common/

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;
}