bench/src/admission_log.rs
//! Reading the **admission gate's refusal counters off the server's own log**.
//!
//! The `vs_forge_*` concurrent arms used to emit no in-flight or
//! refusal column, so "N clients, nothing refused" could not be told apart from
//! "the gate was never asked to bind". This module is the reader that closes
//! that gap, and it does so by the same means every other gunnar-only telemetry
//! the forge harness reports already uses (`RepackTelemetry` off `gunnar gc`,
//! the storage-engine off `store.engine_selected`): it parses a **line the
//! server already writes**, rather than adding a second metrics path.
//!
//! # The line
//!
//! When the admission gate refuses a transfer at capacity, each of the three
//! front doors logs the *same* fields from the *same* `gunnar_ssh::Admission`
//! gate:
//!
//! * `gunnar-http` — `event="http.transfer.refused"` (`service.rs::refuse_admission`)
//! * `gunnar-ssh` — `event="git.transfer.refused"` (`server.rs`)
//! * `gunnar-grpc` — `event="grpc.transfer.refused"` (`grpc_gate.rs`)
//!
//! carrying `in_flight`, `peak_in_flight`, `limit`, `reserved_bytes` and
//! `refused_total`. The two this column reports are:
//!
//! * `refused_total` — which is exactly `gunnar_ssh::AdmissionStats::refused`,
//! the *cumulative* count refused at capacity, and
//! * `peak_in_flight` — `gunnar_ssh::AdmissionStats::peak_in_flight`, the most
//! transfers ever concurrently in flight, which at a binding ceiling settles
//! on the ceiling.
//!
//! The forge harness clones over smart-HTTP, so the line it reads is the HTTP
//! one; the parser matches all three `*.transfer.refused` events so the same
//! reader serves an SSH or gRPC arm unchanged.
//!
//! # Why a refusal line and not a periodic stat
//!
//! `peak_in_flight` below a binding ceiling is not surfaced by *any* line the
//! server writes today, and surfacing it would be a `crates/` change (a stats
//! endpoint or a periodic log), which is out of scope for the bench crate. So this
//! reader reports what the log *does* carry: a refusal line's counters. That is
//! sufficient for the one thing the column exists to make visible — that the
//! gate **binds**: below the ceiling there are no refusal lines and
//! [`AdmissionRefusals::refused`] reads `0`; at a binding ceiling the lines
//! appear and it reads the gate's own non-zero counter, with `peak_in_flight`
//! equal to the ceiling.
/// Strip ANSI/CSI escape sequences from a log line.
///
/// The `tracing` fmt formatter, when it decides it is writing to a terminal,
/// renders a field as `\x1b[3mpeak_in_flight\x1b[0m\x1b[2m=\x1b[0m3` — the name
/// italic, the `=` dim — so a plain search for `peak_in_flight=` finds nothing,
/// which reads exactly like a field that was never emitted. Container logs are
/// captured off a pipe (no tty), so the common case is *no* ANSI, but a run with
/// `RUST_LOG_STYLE=always` or a future formatter change would put it back; this
/// strips it either way so the parser sees `peak_in_flight=3` in both.
///
/// Shared with [`crate`]'s repack-telemetry reader rather than twinned: the same
/// escape shape bit both, and one stripper is one place to be right.
pub fn strip_ansi(s: &str) -> String {
let mut out = String::with_capacity(s.len());
let mut chars = s.chars().peekable();
while let Some(c) = chars.next() {
if c == '\u{1b}' {
// ESC begins a control sequence; consume up to and including the
// terminating letter (`m` for SGR, but be liberal — any final byte
// in the `@`..`~` range ends a CSI).
if chars.peek() == Some(&'[') {
chars.next();
}
for d in chars.by_ref() {
if ('\u{40}'..='\u{7e}').contains(&d) {
break;
}
}
continue;
}
out.push(c);
}
out
}
/// Read one unsigned field value out of an (already ANSI-stripped) line.
///
/// Finds `key=` and reads the leading run of digits after it. Returns `None`
/// when the key is absent, so a caller can tell "field not on this line" from
/// "field was zero" — the distinction the whole column turns on.
fn u64_field(line: &str, key: &str) -> Option<u64> {
let needle = format!("{key}=");
let rest = line.split(&needle).nth(1)?;
let end = rest
.find(|c: char| !c.is_ascii_digit())
.unwrap_or(rest.len());
if end == 0 {
return None;
}
rest[..end].parse::<u64>().ok()
}
/// The admission gate's refusal counters as read off the server's log.
///
/// Every field is the gate's *own* number (`gunnar_ssh::AdmissionStats`),
/// never a count this reader invents — a refusal here is only ever data the
/// server already committed to a log line.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct AdmissionRefusals {
/// The largest `refused_total` (== `gunnar_ssh::AdmissionStats::refused`)
/// seen across the scanned lines, i.e. the gate's cumulative
/// refused-at-capacity count as of the last refusal in this window. `0` when
/// the gate never refused within these lines.
pub refused: u64,
/// The largest `peak_in_flight`
/// (== `gunnar_ssh::AdmissionStats::peak_in_flight`) seen, the high-water
/// of concurrent transfers the gate reported. `0` when no refusal line
/// appeared (the peak is only carried on refusal lines).
pub peak_in_flight: u64,
/// How many `*.transfer.refused` lines these log lines carried — one per
/// refused transfer. A cross-check on [`refused`](Self::refused): the count
/// of lines and the delta of the cumulative counter move together.
pub refusal_lines: u64,
}
impl AdmissionRefusals {
/// Did the gate bind at all in this window?
pub fn bound(&self) -> bool {
self.refused > 0 || self.refusal_lines > 0
}
}
/// Scan drained container-log lines for the admission gate's refusal counters.
///
/// `lines` is what `draupnir`'s `drain_logs_split` handed back for one batch;
/// only `*.transfer.refused` lines contribute. `refused` and `peak_in_flight`
/// are taken as the maximum seen, because both are monotonic counters on the
/// server, so the maximum in a window is the value as of that window's last
/// refusal.
pub fn scan_refusals<S: AsRef<str>>(lines: &[S]) -> AdmissionRefusals {
let mut r = AdmissionRefusals::default();
for raw in lines {
let line = strip_ansi(raw.as_ref());
if !line.contains("transfer.refused") {
continue;
}
r.refusal_lines += 1;
if let Some(v) = u64_field(&line, "refused_total") {
r.refused = r.refused.max(v);
}
if let Some(v) = u64_field(&line, "peak_in_flight") {
r.peak_in_flight = r.peak_in_flight.max(v);
}
}
r
}
#[cfg(test)]
mod tests {
use super::*;
use gunnar_ssh::{Admission, OpClass};
use std::sync::Arc;
/// A smart-HTTP refusal line the way `tracing`'s default fmt layer writes it
/// (no ANSI, off a pipe), with the fields in the order
/// `gunnar-http::refuse_admission` emits them.
fn http_refusal_line(in_flight: u64, peak: u64, limit: u64, refused_total: u64) -> String {
format!(
"2026-09-01T10:00:00.000000Z WARN gunnar::git: smart-HTTP transfer REFUSED: the \
admission gate is full event=\"http.transfer.refused\" \
reason=\"admission-gate-full\" service=git-upload-pack principal=alice \
refusal=at-capacity in_flight={in_flight} peak_in_flight={peak} limit={limit} \
reserved_bytes=0 refused_total={refused_total}"
)
}
#[test]
fn no_refusal_lines_read_zero() {
// A quiet ramp: ordinary chatter, not one refusal. The column MUST read
// 0 — this is the "nothing refused" half, and a reader that could only
// ever emit 0 would pass here and fail the next test.
let quiet = [
"2026-09-01T10:00:00Z INFO gunnar::git: transfer admitted service=git-upload-pack",
"2026-09-01T10:00:01Z INFO gunnar::store: store.engine_selected engine=znippy",
];
let r = scan_refusals(&quiet);
assert_eq!(r, AdmissionRefusals::default());
assert!(!r.bound(), "no refusal line must not read as a bound gate");
}
#[test]
fn reads_the_refused_field_not_a_neighbour() {
// The line carries in_flight=3, peak_in_flight=3 and refused_total=7.
// A parser wired to `in_flight` (a plausible mistake) would report 3 as
// the refusal count; this asserts it reads 7, the refused_total, and the
// peak, and counts the one line.
let line = http_refusal_line(3, 3, 3, 7);
let r = scan_refusals(&[line]);
assert_eq!(r.refused, 7, "must read refused_total, not in_flight");
assert_eq!(r.peak_in_flight, 3);
assert_eq!(r.refusal_lines, 1);
assert!(r.bound());
}
#[test]
fn strips_ansi_field_names() {
// The same line with the field name italic and the `=` dim, as a tty
// formatter writes it. Without the stripper `refused_total=` is not a
// contiguous substring and the field reads as absent — a silent 0.
let ansi = format!(
"\u{1b}[33mWARN\u{1b}[0m gunnar::git event=\"http.transfer.refused\" \
\u{1b}[3mpeak_in_flight\u{1b}[0m\u{1b}[2m=\u{1b}[0m5 \
\u{1b}[3mrefused_total\u{1b}[0m\u{1b}[2m=\u{1b}[0m2"
);
let r = scan_refusals(&[ansi]);
assert_eq!(r.refused, 2);
assert_eq!(r.peak_in_flight, 5);
}
#[test]
fn max_across_a_window_of_lines() {
// A batch that refused several transfers logs several lines; the
// cumulative counter climbs. The window's value is the maximum.
let lines = vec![
http_refusal_line(3, 3, 3, 5),
http_refusal_line(3, 3, 3, 6),
http_refusal_line(3, 3, 3, 7),
];
let r = scan_refusals(&lines);
assert_eq!(r.refused, 7);
assert_eq!(r.peak_in_flight, 3);
assert_eq!(r.refusal_lines, 3);
}
/// The measured proof against the **real gate counter**, not a hand-written
/// string: drive a real [`Admission`] to a binding ceiling, read its own
/// `stats()`, render the refusal line from those numbers exactly as the
/// server does, and assert the parser reads back the gate's counters.
///
/// Red first, and it is a real red: below the ceiling the gate admits and
/// `refused()` is 0, there is no refusal line, and the parser reads 0 — a
/// column hard-wired to a non-zero value would fail the green half, and one
/// hard-wired to 0 (the exact "nothing refused" hollow state at issue)
/// fails the bound half.
#[test]
fn parsed_column_equals_the_real_admission_counter() {
const CEILING: usize = 3;
// Server-wide ceiling AND per-principal share both at CEILING: this
// test binds on the server-wide wall, the same wall the `vs_forge_*`
// ramp has to cross to refuse anything. `try_admit` takes `&Arc<Self>`.
let gate = Arc::new(Admission::new(Some(CEILING), Some(CEILING)));
// ── below the ceiling: admit CEILING transfers, refuse nothing ──────
let mut held = Vec::new();
for _ in 0..CEILING {
held.push(
// `""` for the namespace, which is what the API documents for a
// door that knows none: this test binds on the SERVER-WIDE
// ceiling, so charging the refusal to the box rather than
// inventing a tenant for it is the honest reading.
gate.try_admit("alice", "", OpClass::Clone)
.expect("a transfer under the ceiling is admitted"),
);
}
let under = gate.stats();
assert_eq!(under.refused(), 0, "nothing is refused below the ceiling");
// With no refusal line written, the reader reads 0 — the honest
// "nothing refused" the whole column exists to be able to say.
assert_eq!(scan_refusals::<String>(&[]).refused, under.refused());
// ── at the ceiling: the next transfer is refused at capacity ────────
let refusal = gate
.try_admit("alice", "", OpClass::Clone)
.expect_err("the CEILING+1'th transfer must be refused at capacity");
let _ = refusal;
let at = gate.stats();
assert!(at.refused() >= 1, "the gate must have counted the refusal");
assert_eq!(
at.peak_in_flight as usize, CEILING,
"peak settles on the ceiling"
);
// Render the line the server would have logged from THESE counters, and
// assert the reader recovers the gate's own numbers.
let line = http_refusal_line(
at.in_flight,
at.peak_in_flight,
at.limit as u64,
at.refused(),
);
let parsed = scan_refusals(&[line]);
assert_eq!(
parsed.refused,
at.refused(),
"the reported refusal count must equal AdmissionStats::refused()"
);
assert_eq!(
parsed.peak_in_flight, at.peak_in_flight,
"the reported in-flight peak must equal AdmissionStats::peak_in_flight"
);
assert!(parsed.bound());
drop(held);
}
}