crates/gunnar-ui-server/src/sse.rs
//! The live feed: gunnar's event stream, re-published as server-sent events.
//!
//! One subscription to gunnar, fanned out to every open browser. gunnar's bus
//! is a broadcast ring, so a second subscriber would be a second thing to fall
//! behind for no benefit.
//!
//! **The shared fan-out is the box administrator's.** The one stream is opened
//! on the console's own key and carries every repository on the box, so only an
//! administrator's connection is handed it. Everybody else's feed is their OWN
//! `Events.Watch`, signed with their own key, and gunnar decides what is in it
//! (thin layer, owner 2026-10-04). See [`events`].
//!
//! # Losing events loudly
//!
//! gunnar tells a subscriber that has fallen behind exactly how much it missed,
//! with an `EventGap`. It now also answers the two questions a RECONNECT asks,
//! and the pump asks both rather than guessing:
//!
//! * **"where did I leave off?"** — `WatchRequest.from_seq`. The pump keeps the
//! sequence it last forwarded and sends it back, so everything gunnar still
//! holds past that point is replayed instead of being silently skipped. If
//! the cursor has fallen out of gunnar's ring, gunnar itself says so with an
//! `EventGap` carrying a MEASURED `dropped` — which is why this file no
//! longer synthesises a gap of its own. The old one was arithmetic on two
//! sequence numbers seen either side of a reconnect; it could not tell a
//! restart from a hole, and after a restart it produced nothing at all
//! because the new numbers were LOWER.
//!
//! * **"are these numbers even from the same server run?"** — `StreamOpened`,
//! the first message of every stream, carries an epoch. When it changes, the
//! cursor is void and the honest thing to show is that the feed restarted,
//! not a fabricated count of what was missed. Nothing buffered those events;
//! nobody can say how many there were.
//!
//! An SSE `Last-Event-ID` still cannot help a browser here, so the client does
//! not resume: it re-reads. The resume is between this process and gunnar.
//!
//! # A browser that opened at three in the morning saw nothing at all
//!
//! Both resume mechanisms above are about a RECONNECT of this process. Neither
//! did anything for the thing an operator actually does, which is open the
//! console. The overview said, in one panel, "Events 147 published · the ring
//! holds 1 024", and in the panel beside it "Nothing has happened since the
//! page was opened" — both true, and together an accusation: gunnar was holding
//! every one of those 147 events and the console was throwing them away.
//!
//! It threw them away twice over, and both halves are fixed here:
//!
//! * **The fan-out had no memory.** `broadcast::Sender::subscribe` starts a
//! receiver at the tail, so a tab opened one second after a push never learned
//! of it. The hub now keeps the last [`BACKLOG`] rendered lines and hands a
//! new connection that backlog before its receiver — see
//! [`EventHub::subscribe_with_backlog`] for why the two must be taken under
//! one lock, which is the whole correctness argument.
//!
//! * **The pump started at gunnar's live tail.** Its first `WatchRequest` sent
//! `from_seq: 0`, which gunnar reads as "start live" and not as "I have seen
//! nothing" — so after a restart of THIS process, gunnar's ring was ignored
//! and the backlog began empty however much gunnar was still holding. See
//! [`FROM_THE_RING`].
use std::collections::VecDeque;
use std::convert::Infallible;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use axum::extract::State;
use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use futures_util::stream::Stream;
use futures_util::StreamExt as _;
use http::{HeaderMap, StatusCode};
use serde::Serialize;
use serde_json::json;
use tokio::sync::broadcast;
use crate::app::AppState;
use crate::auth::{session_cookie, Webauthn};
use crate::grpc::Gunnar;
use crate::pb;
/// How many items a slow browser may fall behind before it is told it lost some.
const FANOUT_CAPACITY: usize = 512;
/// How long to wait before re-subscribing after the stream to gunnar ends.
const RECONNECT_DELAY: Duration = Duration::from_secs(3);
/// The longest wait between two attempts while gunnar cannot be subscribed to.
const RECONNECT_CEILING: Duration = Duration::from_secs(60);
/// While a run of failures lasts with the same reason, it is said again this
/// often — and no more.
const RESAY_EVERY: Duration = Duration::from_secs(300);
/// **How the pump re-subscribes after a failure: the wait, and what to say.**
///
/// A gunnar that is gone (a main lost and laid again) or that does not know
/// the console's key yet (a fresh main before its restore) is a state that
/// lasts minutes, and asking every 3 s made it a WARN on this box AND a
/// `grpc.auth.refused` WARN on gunnar every time (release-3 estate-box: 7+/min
/// on both). The wait doubles from [`RECONNECT_DELAY`] to
/// [`RECONNECT_CEILING`]; the run is said once, again when its reason changes
/// or every [`RESAY_EVERY`], and its end once.
#[derive(Debug, Default)]
pub(crate) struct Resubscribe {
/// Attempts in a row that opened no stream.
failed: u32,
/// When this run of failures began.
since: Option<std::time::Instant>,
/// The reason said last, and when.
said: Option<(tonic::Code, std::time::Instant)>,
}
impl Resubscribe {
/// One attempt failed with `code` (`what` is its message): how long to
/// wait, and the line to say, if any — `(true, …)` at WARN.
pub(crate) fn failed(&mut self, code: tonic::Code, what: &str, now: std::time::Instant) -> (Duration, Option<(bool, String)>) {
self.attempt(code, what, now, "cannot subscribe to gunnar's event stream")
}
/// **A stream that opened and then ended** with `code`: `delivered` says
/// whether it carried any event. One that carried none is a failed attempt
/// of the run already going (main stopped by an upgrade: the relay accepts
/// the call and the stream ends at once); one that did starts a run of its
/// own ([`Resubscribe::opened`] ended the last). Either way the run backs
/// off and is said once, as [`Resubscribe::failed`] is.
pub(crate) fn ended(&mut self, code: tonic::Code, what: &str, delivered: bool, now: std::time::Instant) -> (Duration, Option<(bool, String)>) {
let _ = delivered;
self.attempt(code, what, now, "the event stream from gunnar ended")
}
fn attempt(&mut self, code: tonic::Code, what: &str, now: std::time::Instant, first: &str) -> (Duration, Option<(bool, String)>) {
self.failed = self.failed.saturating_add(1);
let since = *self.since.get_or_insert(now);
let line = match self.said {
Some((c, at)) if c == code && now.duration_since(at) < RESAY_EVERY => None,
Some((c, _)) if c == code => Some((
true,
format!(
"still cannot subscribe to gunnar's event stream ({code:?}) after {} attempt(s) over {} s: {what}",
self.failed,
now.duration_since(since).as_secs()
),
)),
// A key gunnar does not know YET (a fresh primary before its principals are
// installed, a restored one before its restart) is a wait, said at INFO.
_ => Some((code != tonic::Code::Unauthenticated, format!("{first} ({code:?}): {what} — asking again at a slowing pace, said again only if this changes"))),
};
if line.is_some() {
self.said = Some((code, now));
}
let wait = RECONNECT_DELAY.saturating_mul(1u32 << (self.failed - 1).min(5)).min(RECONNECT_CEILING);
(wait, line)
}
/// A stream opened: the end of a run of failures, said once.
pub(crate) fn opened(&mut self, now: std::time::Instant) -> Option<String> {
let failed = std::mem::take(&mut self.failed);
let since = self.since.take();
self.said = None;
(failed > 0).then(|| {
format!(
"subscribed again to gunnar's event stream after {failed} failed attempt(s) over {} s",
since.map(|s| now.duration_since(s).as_secs()).unwrap_or(0)
)
})
}
}
/// **How many rendered lines the hub keeps for a browser that has not connected
/// yet.**
///
/// Two hundred, which is exactly what the console keeps in view
/// (`gunnar_ui_web::feed::KEEP`): handing a new tab more than it will ever
/// display would be bytes spent to be truncated on arrival, and handing it
/// fewer would leave a page that scrolls to a floor it did not reach on its
/// own. The bound is on the number of items and not on their size because a
/// [`FeedItem`] is a handful of short strings — the summaries are one line by
/// construction, they are built by [`render`] and nothing else, and the one
/// field that could carry a repository's own text is a scrubbed URL.
///
/// It is a bound and not a target. This is a convenience buffer in a console,
/// not a log: gunnar's ring is the deeper history and is itself not durable.
const BACKLOG: usize = 200;
/// **The `from_seq` that means "whatever you still hold".**
///
/// `WatchRequest.from_seq` is "the sequence of the last event I processed", and
/// 0 is spelled, deliberately and with a test of its own on gunnar's side, as
/// "start at the live tail". So the wire has no way for a subscriber that has
/// processed NOTHING to say so: the number it honestly holds is 0, and 0 asks
/// for the opposite of what it wants.
///
/// 1 is as close as the protocol gets. gunnar replays every buffered event with
/// `seq > 1`, which is the whole ring bar its very first event, and reports a
/// measured `EventGap` for anything between 1 and the oldest it still holds —
/// so nothing is silently skipped and nothing is invented. The cost is exactly
/// one event per run of gunnar, seq 1, which is unreachable from here.
///
/// That is a gap in the protocol and not in this file. Closing it would take an
/// additive field on `WatchRequest` — `optional uint64 from_seq`, where the
/// absent case keeps today's live-tail meaning and an explicit 0 means "from
/// the beginning of the ring". Until gunnar grows one, this constant is the
/// honest approximation and this comment is the record of what it costs.
const FROM_THE_RING: u64 = 1;
/// **How many live feeds one session may hold at once.**
///
/// Generous for a person and cheap to hit for a script: a console open in five
/// tabs is five, and a reconnect loop that never closes the old one is eight
/// within a second. Eight rather than one, because an SSE stream is per TAB and
/// a browser that reconnects can briefly hold two of them for the same tab —
/// a cap of one would break the ordinary case it is supposed to survive.
pub const PER_SESSION_FEEDS: usize = 8;
/// **How many live feeds this console will hold in total.**
///
/// Every connection pins a `broadcast::Receiver`, and a receiver that falls
/// behind holds its slot in a ring of [`FANOUT_CAPACITY`] items until it is
/// read or dropped. The per-session cap alone does not bound that: sessions are
/// minted by registration, and the anonymous half of this console cannot open a
/// feed but an invited viewer with a script can open many. This is the number
/// that keeps the memory a stranger-with-an-account can make this process hold
/// a number rather than a function of how many accounts exist.
const TOTAL_FEEDS: usize = 512;
/// How often an open stream re-checks that its session is still valid.
///
/// A stream is one HTTP request, authorised once at connect. Without this,
/// revoking every session would leave live feeds running until their tabs
/// closed, and the revocation would be a lie.
const SESSION_RECHECK: Duration = Duration::from_secs(20);
/// One line of the console's feed.
#[derive(Debug, Clone, Serialize)]
pub struct FeedItem {
/// gunnar's server-wide sequence number, or 0 for a gap this process made.
pub seq: u64,
pub at_unix_ms: i64,
/// A lowercase name: `ref_updated`, `repo_created`, `gap`, and so on.
pub kind: String,
/// `<namespace>/<repo>`, when the event names one.
pub repo: Option<String>,
/// The one-line human summary the console shows.
pub summary: String,
/// Set on a gap: how many events are known missing, when that is known.
pub dropped: Option<u64>,
}
/// The fan-out.
#[derive(Clone)]
pub struct EventHub {
tx: broadcast::Sender<FeedItem>,
/// How many feeds are open, per session and altogether.
///
/// On the hub rather than on `App` because the thing it bounds is the fan-
/// out — a connection costs a `broadcast::Receiver` — and because the hub
/// is already the one handle every construction site passes.
live: Arc<Mutex<Live>>,
/// The last [`BACKLOG`] lines, so a browser that opens now is not told the
/// forge has been idle when it has not.
///
/// Its own lock rather than a field on [`Live`]: this one is taken on the
/// publish path, which runs per event on a busy box, and the census lock is
/// taken per connection. Sharing them would put every push behind the same
/// mutex as every tab open for no reason either of them asked for.
recent: Arc<Mutex<VecDeque<FeedItem>>>,
/// The pump feeding this hub ([`EventHub::spawn`]); `None` for a detached
/// hub, and after [`EventHub::stop`].
pump: Arc<Mutex<Option<tokio::task::AbortHandle>>>,
/// gunnar's highest sequence number this hub has published — where a
/// person's own feed asks gunnar to replay from, so it opens with the same
/// recent history an administrator's does (narrowed by gunnar, not here).
latest: Arc<std::sync::atomic::AtomicU64>,
}
/// The live-feed census: one count per session, and the total.
///
/// The key is the blake3 of the session cookie, not the cookie: the value is a
/// credential, this map outlives any one request, and a count needs an identity
/// and not a secret. Sessions with no feeds open are removed rather than left
/// at zero, so the map is the size of what is connected and not of who has ever
/// connected.
#[derive(Default)]
struct Live {
per_session: std::collections::HashMap<[u8; 32], usize>,
total: usize,
}
/// **One open feed's place in the census, given back when the feed ends.**
///
/// A guard and not a pair of counter calls, because the ways an SSE stream ends
/// are not all `return`s: the browser closes the tab, the process drops the
/// response, the session re-check breaks the loop, a write fails. `Drop` covers
/// every one of them, including the panic. A cap that leaked on any of those
/// paths would be a console that stops serving feeds after a while and needs a
/// restart to explain.
pub struct Feed {
hub: EventHub,
key: [u8; 32],
}
impl Drop for Feed {
fn drop(&mut self) {
let mut live = self.hub.live.lock().expect("the feed census");
live.total = live.total.saturating_sub(1);
if let Some(count) = live.per_session.get_mut(&self.key) {
*count = count.saturating_sub(1);
if *count == 0 {
live.per_session.remove(&self.key);
}
}
}
}
/// Why a feed was refused.
///
/// Two reasons and they are told apart on purpose: one is the caller's own
/// doing and is fixed by closing a tab, the other is the console's capacity and
/// is not the caller's fault at all. A single "too many" would send somebody
/// hunting through their own tabs for a limit they are nowhere near.
#[derive(Debug, PartialEq, Eq)]
pub enum NoFeed {
/// This session already holds [`PER_SESSION_FEEDS`].
ThisSession,
/// The console already holds [`TOTAL_FEEDS`].
ThisConsole,
}
impl std::fmt::Debug for EventHub {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("EventHub")
.field("subscribers", &self.tx.receiver_count())
.finish()
}
}
impl EventHub {
/// Create the hub and start pumping gunnar's stream into it.
pub fn spawn(gunnar: Gunnar) -> EventHub {
let (tx, _) = broadcast::channel(FANOUT_CAPACITY);
let hub = EventHub {
tx: tx.clone(),
live: Arc::new(Mutex::new(Live::default())),
recent: Arc::new(Mutex::new(VecDeque::with_capacity(BACKLOG))),
pump: Arc::new(Mutex::new(None)),
latest: Arc::new(std::sync::atomic::AtomicU64::new(0)),
};
let task = tokio::spawn(pump(gunnar, hub.clone()));
*hub.pump.lock().expect("the pump slot") = Some(task.abort_handle());
hub
}
/// **Stop pumping gunnar's stream into this hub** — the hub of a console
/// that was replaced (gunnar-front's `console::start` builds a new router,
/// with a new hub, every time the fed primary changes).
///
/// The pump task is aborted: its subscription is dropped (the stream to
/// that main is cancelled) and it never subscribes again. Feeds already
/// open on this hub stay open and hear nothing more; the browser's next
/// connection reaches the new console. Idempotent.
pub fn stop(&self) {
if let Some(task) = self.pump.lock().expect("the pump slot").take() {
task.abort();
}
}
/// Is a pump still feeding this hub?
pub fn pumping(&self) -> bool {
self.pump.lock().expect("the pump slot").as_ref().is_some_and(|t| !t.is_finished())
}
/// A hub with no source, for tests and for an API-only deployment.
pub fn detached() -> EventHub {
let (tx, _) = broadcast::channel(FANOUT_CAPACITY);
EventHub {
tx,
live: Arc::new(Mutex::new(Live::default())),
recent: Arc::new(Mutex::new(VecDeque::with_capacity(BACKLOG))),
pump: Arc::new(Mutex::new(None)),
latest: Arc::new(std::sync::atomic::AtomicU64::new(0)),
}
}
/// **Take a place in the census for one new feed, or say why not.**
///
/// `session` is whatever names this connection's session — the cookie for a
/// browser, and for the break-glass token the token itself, which is one
/// operator with one bucket. It is hashed on the way in and never stored.
///
/// # Errors
///
/// This session already holds [`PER_SESSION_FEEDS`], or the console already
/// holds [`TOTAL_FEEDS`]. The total is checked FIRST: when the console is
/// full, saying "you have too many tabs open" to somebody with one is a
/// lie, and the two are told apart so the message can be true.
pub fn admit(&self, session: &str) -> Result<Feed, NoFeed> {
let key = *blake3::hash(session.as_bytes()).as_bytes();
let mut live = self.live.lock().expect("the feed census");
if live.total >= TOTAL_FEEDS {
return Err(NoFeed::ThisConsole);
}
let held = live.per_session.entry(key).or_default();
if *held >= PER_SESSION_FEEDS {
return Err(NoFeed::ThisSession);
}
*held += 1;
live.total += 1;
Ok(Feed {
hub: self.clone(),
key,
})
}
/// gunnar's highest sequence number this hub has seen, 0 for none.
pub fn latest_seq(&self) -> u64 {
self.latest.load(std::sync::atomic::Ordering::Relaxed)
}
/// How many feeds are open right now, in total.
///
/// For the tests and for [`std::fmt::Debug`]: `receiver_count` counts the
/// pump's own subscription too and is not the same number.
pub fn live_feeds(&self) -> usize {
self.live.lock().expect("the feed census").total
}
/// **What the hub already has, and a receiver for everything after it.**
///
/// The one way to listen. There is deliberately no live-only door beside
/// it: the defect this replaced was a connection that started at the tail,
/// and a second spelling would be a second chance to open one by accident.
///
/// # The lock is held across `subscribe`, and that is the whole point
///
/// The join between "what the backlog returned" and "what the receiver
/// delivers" has to be exact, and the two obvious spellings are both wrong:
///
/// * snapshot, then subscribe — an event published in between is in neither
/// half, and the operator is shown a feed with a hole in it that nothing
/// reports, because nothing counted it;
/// * subscribe, then snapshot — an event published in between is in BOTH,
/// and the same push appears twice on an audit feed, which is worse than
/// a hole: a reader cannot tell it from two pushes.
///
/// Taking the backlog and the receiver under one lock that [`publish`] also
/// holds across its own `send` makes the split exact instead of likely:
/// every item in the returned vector was sent before this receiver existed,
/// so the receiver cannot repeat one, and every item sent after this
/// returns reaches the receiver. The same argument gunnar's own bus makes
/// about its `recent` deque, for the same reason.
///
/// [`publish`]: EventHub::publish
pub fn subscribe_with_backlog(&self) -> (Vec<FeedItem>, broadcast::Receiver<FeedItem>) {
let recent = self.recent.lock().expect("the feed backlog");
let rx = self.tx.subscribe();
(recent.iter().cloned().collect(), rx)
}
/// Publish an item, retaining it for connections that do not exist yet.
/// Used by the pump, and by tests.
pub fn publish(&self, item: FeedItem) {
if item.seq > 0 {
self.latest
.fetch_max(item.seq, std::sync::atomic::Ordering::Relaxed);
}
let mut recent = self.recent.lock().expect("the feed backlog");
if recent.len() == BACKLOG {
recent.pop_front();
}
recent.push_back(item.clone());
// Under the lock, and not after it: see `subscribe_with_backlog`. An
// error here only means nobody is listening.
let _ = self.tx.send(item);
}
}
/// Where this process had got to on gunnar's stream, and in which run.
///
/// The two halves travel together on purpose. A sequence number without the
/// epoch it was assigned in is the exact thing that made a restart
/// indistinguishable from a quiet hour.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Cursor {
pub epoch: String,
pub seq: u64,
}
/// **The cursor to send, given where this process has got to.**
///
/// A pure function because the interesting case is the one with no number in
/// it. `cursor` is `None` on the very first subscription, and its `seq` is 0
/// after an epoch change reset it — both mean "this process has processed
/// nothing in the run it is about to talk to", and both used to be sent as
/// `from_seq: 0`, which gunnar reads as "start live". That is how a console
/// restarted at noon could show an empty feed while gunnar sat on a ringful of
/// the morning's pushes.
///
/// [`FROM_THE_RING`] carries the reasoning and what the floor costs.
fn resume_from(cursor: Option<&Cursor>) -> u64 {
cursor.map(|c| c.seq).unwrap_or(0).max(FROM_THE_RING)
}
/// Subscribe to gunnar forever, resuming where the last connection stopped.
async fn pump(gunnar: Gunnar, hub: EventHub) {
let mut cursor: Option<Cursor> = None;
let mut again = Resubscribe::default();
loop {
let mut wait = RECONNECT_DELAY;
let stream = gunnar
.events()
.watch(pb::WatchRequest {
// Everything, unfiltered: the console filters in the browser, so
// one subscription serves every open tab whatever it is looking
// at. Attestation object ids stay off — they carry every new oid
// in a push and no view shows them.
include_attestation_object_ids: false,
// Ask to resume. gunnar replays what it still holds past this
// point and sends its own `EventGap` for whatever it does not,
// which is a measured number rather than one this process
// inferred from two sequence numbers.
from_seq: resume_from(cursor.as_ref()),
..Default::default()
})
.await;
match stream {
Ok(response) => {
// The run of failures ends at the first EVENT, not at the
// call's answer: while main is down (an upgrade, a recharge) the
// relay answers the call and the stream ends at once, and that
// is one more failed attempt (sse-upgrade-spam).
let mut delivered = false;
let mut events = response.into_inner();
let mut first = true;
while let Some(next) = events.next().await {
match next {
Ok(event) => {
if !delivered {
delivered = true;
if let Some(line) = again.opened(std::time::Instant::now()) {
tracing::info!("{line}");
}
}
let prologue = match event.payload.as_ref() {
Some(pb::event::Payload::StreamOpened(opened)) => Some(opened),
_ => None,
};
if first {
if let Some(item) = opening(cursor.as_ref(), prologue) {
hub.publish(item);
}
first = false;
}
if let Some(opened) = prologue {
// The prologue is bookkeeping, not news: it
// moves the cursor into this run and is not put
// on the console's feed. Anything worth showing
// about it was said by `opening` above.
//
// A different epoch resets the position to 0.
// Keeping the old number would be worse than
// having none: if this connection dropped
// before its first event, the next `from_seq`
// would name a sequence in this run that this
// process never saw, and gunnar would skip it.
let same_run =
cursor.as_ref().is_some_and(|c| c.epoch == opened.epoch);
cursor = Some(Cursor {
epoch: opened.epoch.clone(),
seq: if same_run {
cursor.as_ref().map(|c| c.seq).unwrap_or(0)
} else {
0
},
});
continue;
}
// Gaps and other synthesised markers carry seq 0 and
// must not drag the cursor backwards.
if event.seq > 0 {
match cursor.as_mut() {
Some(c) => c.seq = event.seq,
None => {
cursor = Some(Cursor {
epoch: String::new(),
seq: event.seq,
})
}
}
}
hub.publish(render(event));
}
Err(status) => {
let (w, line) = again.ended(status.code(), status.message(), delivered, std::time::Instant::now());
wait = w;
match line {
Some((true, line)) => tracing::warn!("{line}"),
Some((false, line)) => tracing::info!("{line}"),
None => tracing::debug!(%status, "the event stream from gunnar ended (said already)"),
}
break;
}
}
}
}
Err(status) => {
let (w, line) = again.failed(status.code(), status.message(), std::time::Instant::now());
wait = w;
match line {
Some((true, line)) => tracing::warn!("{line}"),
Some((false, line)) => tracing::info!("{line}"),
None => tracing::debug!(%status, "cannot subscribe to gunnar's event stream (said already)"),
}
}
}
tokio::time::sleep(wait).await;
}
}
/// What, if anything, the console must be told when a new subscription opens.
///
/// `None` means "say nothing", which is the right answer twice: on the very
/// first connection, because everything before the console started is history
/// it never claimed to have; and on a reconnect within the same epoch, because
/// the cursor was honoured and any real loss arrives as gunnar's own
/// `EventGap`, counted rather than guessed.
fn opening(previous: Option<&Cursor>, opened: Option<&pb::StreamOpened>) -> Option<FeedItem> {
let previous = previous?;
let break_line = |summary: String| FeedItem {
seq: 0,
at_unix_ms: crate::auth::now_secs() * 1000,
kind: "restart".to_owned(),
repo: None,
summary,
// **Deliberately none.** Nothing buffered the events on the other side
// of a restart, so there is no number to report. The gap this replaced
// printed one anyway, computed from sequence numbers belonging to two
// different runs.
dropped: None,
};
match opened {
Some(opened) if opened.epoch == previous.epoch => None,
Some(opened) => Some(break_line(format!(
"gunnar restarted: its event stream began again at sequence 1, so the console \
cannot line up what it had (sequence {}) with what it is being sent now. \
Whatever happened in between was not recorded anywhere. The lists have been \
re-read. (stream {} → {})",
previous.seq,
short_epoch(&previous.epoch),
short_epoch(&opened.epoch)
))),
// No prologue at all: an older gunnar, which cannot say which run it is
// in. The console must not claim continuity it has no evidence for.
None => Some(break_line(
"gunnar did not say which run of its event stream this is, so the console cannot \
confirm the feed continues where it left off. The lists have been re-read."
.to_owned(),
)),
}
}
/// Enough of an epoch to tell two apart in a feed line, without printing 28
/// hex characters at an operator.
fn short_epoch(epoch: &str) -> String {
if epoch.is_empty() {
return "unknown".to_owned();
}
epoch.chars().take(8).collect()
}
/// Turn one of gunnar's events into a feed line.
fn render(event: pb::Event) -> FeedItem {
use pb::event::Payload;
let path = |p: &Option<pb::RepoPath>| p.as_ref().map(|p| format!("{}/{}", p.namespace, p.repo));
let (kind, repo, summary, dropped) = match event.payload {
Some(Payload::RefUpdated(e)) => (
"ref_updated",
path(&e.repo),
format!(
"{} moved to {}{}{}",
e.ref_name,
short(&e.new_oid),
if e.forced { " (forced)" } else { "" },
by_credential(&e.credential, &e.key_fingerprint)
),
None,
),
Some(Payload::RefDeleted(e)) => (
"ref_deleted",
path(&e.repo),
format!(
"{} was deleted{}",
e.ref_name,
by_credential(&e.credential, &e.key_fingerprint)
),
None,
),
Some(Payload::PushRejected(e)) => (
"push_rejected",
path(&e.repo),
format!("{} was rejected by {}: {}", e.ref_name, e.gate, e.reason),
None,
),
Some(Payload::RepoCreated(e)) => (
"repo_created",
e.repo.as_ref().and_then(|r| path(&r.path)),
// HENRIK-2: the line names who created it (the event's principal), as the
// encryption line does; an event that names nobody keeps the plain words.
if e.principal_id.is_empty() { "the repository was created".to_owned() } else { format!("{} created the repository", e.principal_id) },
None,
),
Some(Payload::RepoDeleted(e)) => (
"repo_deleted",
path(&e.repo),
"the repository was deleted".to_owned(),
None,
),
// **The remote is scrubbed here, by the same function the REST path
// uses.** A mirror URL of the form `https://<token>@host/…` is an
// ordinary thing to find in `refs/gunnar/mirrors` — it is what `gunnar
// mirror add` takes from a shell — and this line put it verbatim in
// front of every session with the console open. `api::mirrors::scrub`
// has existed for exactly that since the panel did; the feed was
// written without it, which made the credential's audience "everyone
// logged in" rather than "whoever opened the mirror panel". Called
// rather than reimplemented, so the two renderings cannot drift.
Some(Payload::MirrorSynced(e)) => (
"mirror_synced",
path(&e.repo),
format!(
"{} references synced to {}",
e.refs_changed,
crate::api::mirrors::scrub(&e.remote_url)
),
None,
),
Some(Payload::KeyUsed(e)) => (
"key_used",
path(&e.repo),
format!("{} used {}", short_fingerprint(&e.fingerprint), e.operation),
None,
),
Some(Payload::PushAttestation(e)) => (
"push_attestation",
path(&e.repo),
format!(
"{} pushed {} objects ({})",
e.principal_name,
e.new_object_count,
bytes(e.new_object_bytes)
),
None,
),
Some(Payload::ReplicationFailed(e)) => (
"replication_failed",
path(&e.repo),
format!(
"replication to {} failed on attempt {}: {}{}",
e.target,
e.attempt,
e.reason,
if e.needs_full_resync {
" — a full resync is needed"
} else {
""
}
),
None,
),
Some(Payload::PullRequestOpened(e)) => (
"pull_request_opened",
path(&e.repo),
format!(
"{} opened pull request #{} \u{201c}{}\u{201d} ({} \u{2192} {})",
e.principal_id,
e.number,
e.title,
short_ref(&e.source_ref),
short_ref(&e.target_ref)
),
None,
),
Some(Payload::PullRequestMerged(e)) => (
"pull_request_merged",
path(&e.repo),
format!(
"{} merged pull request #{} \u{201c}{}\u{201d}: {} moved to {}",
e.principal_id,
e.number,
e.title,
short_ref(&e.target_ref),
short(&e.merge_oid)
),
None,
),
Some(Payload::PullRequestClosed(e)) => (
"pull_request_closed",
path(&e.repo),
format!(
"{} closed pull request #{} \u{201c}{}\u{201d} without merging",
e.principal_id, e.number, e.title
),
None,
),
// A repository moved between Private and Public. gunnar only publishes
// this when `from != to`, so the line can state the move flatly.
// Rendered through `api::visibility_name` rather than a second table of
// words here, one implementation, so the feed and the repository page cannot
// disagree about what a visibility is called.
Some(Payload::VisibilityChanged(e)) => (
"visibility_changed",
path(&e.repo),
format!(
"{} made it {} (was {})",
e.principal_id,
crate::api::visibility_name(e.to),
crate::api::visibility_name(e.from)
),
None,
),
// A repository's "encrypted" flag turned on or off (gunnar publishes it
// only when `from != to`, wiring tracer B 2026-10-08). Plain words: the
// feed says what the account admin decided, never how it is enforced.
Some(Payload::EncryptedChanged(e)) => (
"encrypted_changed",
path(&e.repo),
if e.to {
format!("{} turned encryption on", e.principal_id)
} else {
format!("{} turned encryption off", e.principal_id)
},
None,
),
// The pump consumes the prologue itself (it is the cursor's
// bookkeeping, not an event), so this arm is not normally reached. It
// exists because the alternative is falling through to "unknown", and a
// feed line reading "gunnar sent an event this console does not
// recognise" about the message the console depends on would be a lie
// that is hard to trace.
Some(Payload::StreamOpened(e)) => (
"stream_opened",
None,
format!(
"the feed connected to gunnar (stream {}, at sequence {})",
short_epoch(&e.epoch),
e.last_seq
),
None,
),
// **What this line may not say is WHY.** It read "the console fell
// behind and gunnar dropped n events", which was true of the only way a
// gap could reach this process when it was written: a live subscriber
// that could not keep up with the broadcast ring.
//
// It is not true of the other way, which [`FROM_THE_RING`] opened. A
// cold start asks gunnar for everything it still holds, and a gunnar
// that has published more than a ringful answers with a gap — measured,
// correct, and nothing whatever to do with this console keeping up. One
// `EventGap` carries both cases and carries nothing that tells them
// apart, so the line reports the measurement and leaves the cause to
// the operator, who can see the ring's size on the same page.
//
// The range is spelled two ways for the same reason. `resume_seq` is 0
// when the buffer holds nothing past the cursor at all, and subtracting
// one from it printed "sequence 2 to 0" — a range that runs backwards
// and names an event that cannot exist, since sequences start at 1.
Some(Payload::Gap(e)) => (
"gap",
None,
match e.resume_seq {
0 => format!(
"gunnar no longer holds {} events, from sequence {} onwards: its event \
ring is bounded and nothing else recorded them",
e.dropped,
e.last_delivered_seq + 1,
),
resume => format!(
"gunnar no longer holds {} events (sequence {} to {}): its event ring is \
bounded and nothing else recorded them",
e.dropped,
e.last_delivered_seq + 1,
resume - 1,
),
},
Some(e.dropped),
),
// A kind this console does not know yet. Say so rather than drop it: a
// silent omission in an audit feed is worse than an unfamiliar word.
None => (
"unknown",
None,
format!(
"gunnar sent an event this console does not recognise (kind {})",
event.kind
),
None,
),
};
FeedItem {
seq: event.seq,
at_unix_ms: event.at_unix_ms,
kind: kind.to_owned(),
repo,
summary,
dropped,
}
}
/// `refs/heads/x` → `x`, for feed lines that already say which repo.
fn short_ref(name: &str) -> &str {
name.strip_prefix("refs/heads/")
.or_else(|| name.strip_prefix("refs/tags/"))
.unwrap_or(name)
}
fn short(oid: &str) -> String {
oid.chars().take(10).collect()
}
fn short_fingerprint(fingerprint: &str) -> String {
match fingerprint.strip_prefix("SHA256:") {
Some(rest) => format!("SHA256:{}", rest.chars().take(8).collect::<String>()),
None => fingerprint.chars().take(16).collect(),
}
}
/// **Which credential moved a ref, as the feed says it** (GUN-014).
///
/// `RefUpdated.key_fingerprint` answers this for a push over SSH and is EMPTY
/// for every push over smart HTTP, because HTTP authenticates with a token and
/// there is no key to name. Until this existed the feed read only the
/// fingerprint, so an HTTP push was a ref that moved with nobody attached to
/// it — the attribution hole GUN-014 names. `RefUpdated.credential` is the
/// field that is always answerable: `key:SHA256:…` or `token:<hash prefix>`,
/// and empty only for a caller that presented none.
///
/// The fingerprint is the fallback and not the other way round, so a console
/// talking to a gunnar older than the field still names the SSH pusher it
/// always named. The leading space belongs to the phrase: the empty case must
/// add nothing at all, and a caller that wrote `"{} by {}"` would leave a
/// dangling "by" on the anonymous line.
///
/// A `token:` value is a HASH PREFIX, never the token — gunnar hashes it
/// before it reaches the wire — so printing it is naming the credential and
/// not disclosing it. It is shortened here for the same reason a fingerprint
/// is: a feed line is read at a glance.
fn by_credential(credential: &str, key_fingerprint: &str) -> String {
let named = match credential.split_once(':') {
Some(("key", fingerprint)) => short_fingerprint(fingerprint),
Some(("token", hash)) => format!("token {}", short_fingerprint(hash)),
// Empty: an older gunnar, or a caller that presented no credential at
// all. Fall back to the fingerprint, which is the answer for SSH.
_ if credential.is_empty() => short_fingerprint(key_fingerprint),
// A form this build has not learned. Naming the credential is the
// whole point, so it is shown — shortened like every other one, and
// never dropped, because a silent drop is the hole this closes.
_ => short_fingerprint(credential),
};
if named.is_empty() {
String::new()
} else {
format!(" by {named}")
}
}
fn bytes(count: u64) -> String {
const UNITS: &[&str] = &["B", "KiB", "MiB", "GiB", "TiB"];
let mut value = count as f64;
let mut unit = 0;
while value >= 1024.0 && unit + 1 < UNITS.len() {
value /= 1024.0;
unit += 1;
}
if unit == 0 {
format!("{count} B")
} else {
format!("{value:.1} {}", UNITS[unit])
}
}
/// `?repo=<namespace>/<name>`: one repository's feed (its Activity tab).
#[derive(Debug, Clone, Default, serde::Deserialize)]
pub struct Scope {
#[serde(default)]
pub repo: Option<String>,
}
/// **`?repo=`: one repository's feed** — a narrowing the page asked for, never
/// a permission. What a person may be told at all is gunnar's answer to their
/// own `Events.Watch` (see [`events`]); this only ever removes lines.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Filter {
/// One repository only (`?repo=`).
pub only: Option<String>,
}
impl Filter {
/// Is this line on the repository the page asked for? An item that names
/// no repository is the feed's own bookkeeping — a gap, a restart — and is
/// what tells a browser its lists are stale, so it always passes.
pub fn may_see(&self, item: &FeedItem) -> bool {
match (&item.repo, &self.only) {
(Some(repo), Some(only)) => repo == only,
_ => true,
}
}
}
/// `GET /events` — the browser's live feed.
///
/// Inside `require_auth`, so a caller is already resolved. The re-check below is
/// about what happens *after* that: this response streams for hours, and the
/// authorisation that opened it was decided once.
///
/// # Whose stream it is (thin layer, owner 2026-10-04)
///
/// * **The box's administrator** (an admin session, the break-glass token, the
/// open console) is handed the console's one shared subscription — the
/// whole server's feed, opened on the console's own key, which is what that
/// role stands for. That is the hub below and its backlog.
/// * **Everybody else** gets a stream of their OWN: an `Events.Watch` signed
/// with their own key (the guest's, for a session bound to no principal),
/// with no namespaces named — gunnar's "mine": every account they are inside
/// (an employee is inside their employer's), plus the accounts whose admin
/// made their events public, and only the repositories there they may read.
/// gunnar re-checks it every 30 s, gates and revocation included, and ends
/// the stream when it no longer holds.
///
/// Until this line the console decided that itself: one admin-keyed pump,
/// filtered by a string rule ("the repository's namespace equals the bound
/// principal's"), so an employee saw nothing of their employer, the
/// `events_public` setting did nothing, and nothing was re-checked.
pub async fn events(
State(state): State<AppState>,
caller: Option<axum::Extension<crate::auth::middleware::Caller>>,
axum::extract::Query(scope): axum::extract::Query<Scope>,
headers: HeaderMap,
) -> Response {
let admin = match caller.as_ref() {
Some(axum::Extension(c)) => c.role().allows(crate::role::Role::Admin),
// No caller past `require_auth` is the open console, where everybody
// is admin by how the process was started.
None => !state.passkey_enabled() && state.admin_token.is_none(),
};
let only = scope.repo.filter(|r| !r.trim().is_empty());
let filter = Filter { only };
let cookie = session_cookie(&headers);
// **The cap, before the subscription.** Taking a `broadcast::Receiver`
// (or a stream from gunnar) first and then refusing would have the refused
// connection hold the resource the cap is about while it unwound.
//
// The bucket is this session. A request with no cookie reaching here is the
// break-glass admin (nothing else gets past `require_auth` on this route),
// and it gets one bucket of its own rather than sharing the empty string
// with a future caller that has no cookie either.
let feed = match state
.events
.admit(cookie.as_deref().unwrap_or("the break-glass admin"))
{
Ok(feed) => feed,
Err(why) => return too_many(why),
};
let webauthn = state.webauthn.clone();
if !admin {
let Some(axum::Extension(caller)) = caller.as_ref() else {
return crate::api::ApiError(crate::grpc::Fail {
code: "permissiondenied".to_owned(),
message: "this console streams events only to a session it can name".to_owned(),
})
.into_response();
};
return own_feed(&state, caller.name(), feed, filter, webauthn, cookie).await;
}
// Taken HERE and not inside the generator below: anything published
// between the two would be missed by a receiver opened on the far side of
// it, and missed silently, since nothing on this path counts what it never
// saw. The backlog and the receiver are one indivisible pair — see
// `subscribe_with_backlog` — so they are taken before the connection can
// block on anything at all.
let (backlog, mut rx) = state.events.subscribe_with_backlog();
let gunnar = state.gunnar.clone();
let stream = async_stream::stream! {
// Moved in, so the census place is given back when this generator is
// dropped — which is what every way an SSE stream can end looks like
// from here, including the ones that are not a `return`.
let _feed = feed;
// The first bytes on the wire, before any event exists: a comment
// line. Without it a quiet server sends nothing for a whole
// keep-alive interval, and a client (or a probe) cannot tell an open,
// idle stream from a connection that hangs before its first byte.
yield Ok::<_, Infallible>(SseEvent::default().comment("connected"));
// The backlog, oldest first, before a single live item. Oldest first
// because the console prepends every line it is handed, so this is the
// order that leaves the newest at the top — the same order the live
// tail arrives in, which is what makes the join invisible on screen as
// well as correct on the wire.
for item in backlog {
if !filter.may_see(&item) {
continue;
}
let data = serde_json::to_string(&item).unwrap_or_else(|_| "{}".to_owned());
yield Ok::<_, Infallible>(SseEvent::default().event("feed").data(data));
}
let mut recheck = tokio::time::interval(SESSION_RECHECK);
recheck.tick().await;
loop {
tokio::select! {
item = rx.recv() => match item {
Ok(item) => {
if !filter.may_see(&item) {
continue;
}
let data = serde_json::to_string(&item)
.unwrap_or_else(|_| "{}".to_owned());
yield Ok::<_, Infallible>(SseEvent::default().event("feed").data(data));
}
// This browser fell behind the fan-out. Same contract as
// gunnar's own gap: say so, do not paper over it.
Err(broadcast::error::RecvError::Lagged(dropped)) => {
let gap = FeedItem {
seq: 0,
at_unix_ms: crate::auth::now_secs() * 1000,
kind: "gap".to_owned(),
repo: None,
summary: format!(
"this tab fell behind and missed {dropped} events; \
the lists have been re-read"
),
dropped: Some(dropped),
};
let data = serde_json::to_string(&gap).unwrap_or_default();
yield Ok(SseEvent::default().event("feed").data(data));
}
Err(broadcast::error::RecvError::Closed) => break,
},
_ = recheck.tick() => {
if !still_authorised(&gunnar, webauthn.as_deref(), cookie.as_deref()).await {
yield Ok(SseEvent::default()
.event("expired")
.data(json!({ "reason": "the session ended" }).to_string()));
break;
}
}
}
}
};
Sse::new(stream)
.keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
.into_response()
}
/// **A person's own feed: gunnar's `Events.Watch`, signed with their key.**
///
/// No namespaces are named, so gunnar answers with "mine" — the accounts this
/// person is inside, plus those whose events their admin made public, and only
/// the repositories there they may read. It asks gunnar to replay from where
/// the console's own pump has got to, less [`BACKLOG`], so the feed opens with
/// recent history — gunnar's narrowing of it, not a copy of the
/// administrator's backlog.
///
/// gunnar refusing the watch, or ending it later (a revoked key, a gate, a
/// removed grant), is said to the browser as `expired` with gunnar's reason.
async fn own_feed(
state: &AppState,
user: &str,
feed: Feed,
filter: Filter,
webauthn: Option<Arc<Webauthn>>,
cookie: Option<String>,
) -> Response {
let reader = match crate::api::tenant_reader(state, user).await {
Ok(reader) => reader,
Err(why) => return why.into_response(),
};
let from_seq = state
.events
.latest_seq()
.saturating_sub(BACKLOG as u64)
.max(FROM_THE_RING);
let watched = reader
.gunnar
.events()
.watch(pb::WatchRequest {
include_attestation_object_ids: false,
from_seq,
// Empty: gunnar's "mine" for a reader that is not the server's.
namespaces: Vec::new(),
..Default::default()
})
.await;
let mut events = match watched {
Ok(response) => response.into_inner(),
Err(status) => return crate::api::ApiError::from(status).into_response(),
};
let gunnar = state.gunnar.clone();
let stream = async_stream::stream! {
let _feed = feed;
yield Ok::<_, Infallible>(SseEvent::default().comment("connected"));
let mut recheck = tokio::time::interval(SESSION_RECHECK);
recheck.tick().await;
loop {
tokio::select! {
next = events.next() => match next {
Some(Ok(event)) => {
// The prologue is bookkeeping for a resuming pump, not
// news for a page.
if matches!(event.payload, Some(pb::event::Payload::StreamOpened(_))) {
continue;
}
let item = render(event);
if !filter.may_see(&item) {
continue;
}
let data = serde_json::to_string(&item).unwrap_or_else(|_| "{}".to_owned());
yield Ok::<_, Infallible>(SseEvent::default().event("feed").data(data));
}
Some(Err(status)) => {
yield Ok(SseEvent::default()
.event("expired")
.data(json!({ "reason": status.message() }).to_string()));
break;
}
None => break,
},
_ = recheck.tick() => {
if !still_authorised(&gunnar, webauthn.as_deref(), cookie.as_deref()).await {
yield Ok(SseEvent::default()
.event("expired")
.data(json!({ "reason": "the session ended" }).to_string()));
break;
}
}
}
}
};
Sse::new(stream)
.keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
.into_response()
}
/// Is the session that opened this stream still good?
///
/// With passkey off there is nothing to check and the console is open anyway.
async fn still_authorised(gunnar: &crate::grpc::Gunnar, webauthn: Option<&Webauthn>, cookie: Option<&str>) -> bool {
let Some(webauthn) = webauthn else {
return true;
};
match cookie {
Some(c) => webauthn.session(gunnar, c).await.is_some(),
None => false,
}
}
/// The 429 a refused feed gets, in the words of whichever limit refused it.
///
/// `Retry-After: 5`, because both reasons clear on their own: a tab closes, or
/// somebody else's feed ends. A refusal with no retry hint reads to a client
/// library as "never", and the console would rather have the reconnect.
fn too_many(why: NoFeed) -> Response {
let message = match why {
NoFeed::ThisSession => format!(
"this session already has {PER_SESSION_FEEDS} live feeds open; close a console \
tab and this one will connect"
),
NoFeed::ThisConsole => format!(
"this console is already streaming {TOTAL_FEEDS} live feeds and will not open \
another; nothing is wrong with your session, try again in a moment"
),
};
(
StatusCode::TOO_MANY_REQUESTS,
[(http::header::RETRY_AFTER, "5")],
message,
)
.into_response()
}
/// `GET /events` when the feed is not available.
pub async fn unavailable() -> Response {
(
StatusCode::SERVICE_UNAVAILABLE,
"the event feed is not connected",
)
.into_response()
}
/// Boxing helper so the handler's opaque stream type stays out of signatures.
pub type FeedStream = std::pin::Pin<Box<dyn Stream<Item = Result<SseEvent, Infallible>> + Send>>;
#[cfg(test)]
mod tests {
use super::*;
fn cursor(epoch: &str, seq: u64) -> Cursor {
Cursor {
epoch: epoch.to_owned(),
seq,
}
}
fn opened(epoch: &str, last_seq: u64) -> pb::StreamOpened {
pb::StreamOpened {
epoch: epoch.to_owned(),
last_seq,
..Default::default()
}
}
/// A visibility flip reads as a MOVE — both ends named, in the right
/// order — and not as a bare "it is public now".
///
/// The kind arrived with gunnar's 2026-09 protos and the compiler forced an
/// arm for it; what the compiler cannot force is that the arm says anything
/// true. `from` and `to` are the same type and adjacent in the message, so
/// crossing them produces a perfectly compiling line that reverses the
/// meaning of every such event in the audit feed. Seen red by writing the
/// arm with the two swapped: *"the line must say what it became: private
/// made it private (was public)"*.
/// An "encrypted" flip says which way it went, in plain words (wiring
/// tracer B 2026-10-08: gunnar publishes EncryptedChanged beside its audit row).
#[test]
fn an_encryption_flip_says_on_or_off() {
let on = render(pb::Event {
seq: 13,
at_unix_ms: 1_700_000_000_000,
kind: pb::EventKind::EncryptedChanged as i32,
payload: Some(pb::event::Payload::EncryptedChanged(pb::EncryptedChanged {
repo: Some(pb::RepoPath { namespace: "acme".to_owned(), repo: "plans".to_owned() }),
from: false,
to: true,
principal_id: "alice".to_owned(),
})),
});
assert_eq!(on.kind, "encrypted_changed");
assert_eq!(on.repo.as_deref(), Some("acme/plans"));
assert_eq!(on.summary, "alice turned encryption on");
let off = render(pb::Event {
seq: 14,
at_unix_ms: 1_700_000_000_001,
kind: pb::EventKind::EncryptedChanged as i32,
payload: Some(pb::event::Payload::EncryptedChanged(pb::EncryptedChanged {
repo: Some(pb::RepoPath { namespace: "acme".to_owned(), repo: "plans".to_owned() }),
from: true,
to: false,
principal_id: "alice".to_owned(),
})),
});
assert_eq!(off.summary, "alice turned encryption off");
}
/// HENRIK-2 (rickardtest 2026-10-26: the owner read "repo created henrik/Konsult" and
/// could not tell who made it): **a created repository names its creator**, the
/// event's `principal_id`, as every other repository line does. A gunnar that sends
/// none keeps the plain words, never an empty name.
#[test]
fn a_created_repository_names_who_created_it() {
let created = |principal_id: &str| {
render(pb::Event {
seq: 15,
at_unix_ms: 1_700_000_000_002,
kind: pb::EventKind::RepoCreated as i32,
payload: Some(pb::event::Payload::RepoCreated(pb::RepoCreated {
repo: Some(pb::Repo {
path: Some(pb::RepoPath { namespace: "henrik".to_owned(), repo: "Konsult".to_owned() }),
..Default::default()
}),
principal_id: principal_id.to_owned(),
})),
})
};
let named = created("alica");
assert_eq!(named.kind, "repo_created");
assert_eq!(named.repo.as_deref(), Some("henrik/Konsult"));
assert_eq!(named.summary, "alica created the repository");
assert_eq!(created("").summary, "the repository was created", "no creator sent: the plain words");
}
#[test]
fn a_visibility_flip_names_both_ends_the_right_way_round() {
let item = render(pb::Event {
seq: 12,
at_unix_ms: 1_700_000_000_000,
kind: pb::EventKind::VisibilityChanged as i32,
payload: Some(pb::event::Payload::VisibilityChanged(
pb::VisibilityChanged {
repo: Some(pb::RepoPath {
namespace: "alice".to_owned(),
repo: "linux".to_owned(),
}),
from: pb::Visibility::Public as i32,
to: pb::Visibility::Private as i32,
principal_id: "alice".to_owned(),
},
)),
});
assert_eq!(item.kind, "visibility_changed");
assert_eq!(item.repo.as_deref(), Some("alice/linux"));
assert!(
item.summary.contains("made it private"),
"the line must say what it became: {}",
item.summary
);
assert!(
item.summary.contains("was public"),
"…and what it was, or a reader cannot tell a tightening from a leak: {}",
item.summary
);
assert!(
item.summary.starts_with("alice"),
"…and who did it: {}",
item.summary
);
}
#[test]
fn the_first_connection_says_nothing() {
// Everything before the console started is history it never had, so
// there is nothing to announce — whatever the epoch is.
assert!(opening(None, Some(&opened("run-a", 4096))).is_none());
assert!(opening(None, None).is_none());
}
#[test]
fn a_reconnect_within_the_same_run_says_nothing() {
// The cursor was sent and honoured. If anything was actually lost,
// gunnar says so itself with an `EventGap` carrying a MEASURED count —
// this process must not add a second, guessed one on top of it.
assert!(opening(Some(&cursor("run-a", 10)), Some(&opened("run-a", 4096))).is_none());
}
#[test]
fn a_restarted_gunnar_is_reported_as_a_restart_and_not_as_a_count() {
// The defect this replaced: two sequence numbers from different runs
// were subtracted from each other. After a restart the new number is
// LOWER, so the old code reported nothing at all — the console silently
// believed the feed was continuous.
let item = opening(Some(&cursor("run-a", 9000)), Some(&opened("run-b", 1)))
.expect("a changed epoch must be reported");
assert_eq!(item.kind, "restart", "not a gap: nothing counted this");
assert_eq!(
item.dropped, None,
"no number may be reported here. Nothing buffered the events on the other side of \
a restart, so any count would be invented"
);
assert!(
item.summary.contains("restarted"),
"the operator must be told what happened: {}",
item.summary
);
assert!(
item.summary.contains("9000"),
"…and where the console had got to: {}",
item.summary
);
}
#[test]
fn an_epoch_that_did_not_change_is_not_reported_as_one_that_did() {
// Both directions on the same axis, or "always report a restart" would
// pass the test above and the feed would cry wolf on every reconnect.
let same = opening(Some(&cursor("run-a", 5)), Some(&opened("run-a", 5)));
let changed = opening(Some(&cursor("run-a", 5)), Some(&opened("run-a-longer", 5)));
assert!(same.is_none());
assert!(
changed.is_some(),
"`run-a` and `run-a-longer` are different epochs; a prefix comparison would miss it"
);
}
#[test]
fn a_server_that_reports_no_epoch_is_not_assumed_continuous() {
// An older gunnar cannot say which run it is in. Claiming continuity
// would be a guess with no evidence, so the console says it cannot
// confirm — and re-reads.
let item = opening(Some(&cursor("run-a", 10)), None).expect("cannot be silent here");
assert_eq!(item.kind, "restart");
assert_eq!(item.dropped, None);
assert!(
item.summary.contains("cannot confirm"),
"the line must say what is unknown rather than assert a fact: {}",
item.summary
);
}
#[test]
fn an_epoch_is_shortened_but_stays_distinguishable() {
assert_eq!(short_epoch("0123456789abcdef"), "01234567");
assert_eq!(short_epoch(""), "unknown");
}
/// ★ **A mirror's credential does not ride out on the feed.**
///
/// The REST path has scrubbed `remote_url` since the mirror panel existed;
/// this line rendered the same field verbatim, to every session with the
/// console open, for every sync of every mirror on the box.
///
/// Red before: the summary carried `ghp_livetoken`, and with it a bearer
/// credential for somebody's whole account on the forge. The second half is
/// the control — the HOST must survive, or the feed stops saying which
/// remote it synced with.
#[test]
fn a_mirror_sync_does_not_put_its_credential_on_the_feed() {
let item = render(pb::Event {
seq: 5,
at_unix_ms: 1,
kind: pb::EventKind::MirrorSynced as i32,
payload: Some(pb::event::Payload::MirrorSynced(pb::MirrorSynced {
repo: Some(pb::RepoPath {
namespace: "nordisk".into(),
repo: "gunnar".into(),
}),
remote_url: "https://x-access-token:ghp_livetoken@github.com/nordisk/gunnar.git"
.into(),
refs_changed: 3,
..Default::default()
})),
});
assert!(
!item.summary.contains("ghp_livetoken"),
"the feed carried a live forge token: {}",
item.summary
);
assert!(
item.summary.contains("github.com/nordisk/gunnar.git"),
"…and it must still say which remote it synced with: {}",
item.summary
);
assert!(item.summary.contains('3'), "{}", item.summary);
}
#[test]
fn an_unrecognised_event_is_still_shown() {
let item = render(pb::Event {
seq: 7,
at_unix_ms: 1,
kind: 42,
payload: None,
});
assert_eq!(item.kind, "unknown");
assert!(item.summary.contains("42"));
}
#[test]
fn a_ref_update_names_the_repository_and_the_object() {
let item = render(pb::Event {
seq: 3,
at_unix_ms: 1,
kind: pb::EventKind::RefUpdated as i32,
payload: Some(pb::event::Payload::RefUpdated(pb::RefUpdated {
repo: Some(pb::RepoPath {
namespace: "vetra".into(),
repo: "gunnar".into(),
}),
ref_name: "refs/heads/main".into(),
new_oid: "0123456789abcdef0123".into(),
forced: true,
..Default::default()
})),
});
assert_eq!(item.repo.as_deref(), Some("vetra/gunnar"));
assert!(item.summary.contains("refs/heads/main"));
assert!(item.summary.contains("0123456789"));
assert!(item.summary.contains("forced"));
}
/// One `RefUpdated`, spelled once, so the four cases below differ only in
/// the two fields they are about.
fn moved(credential: &str, key_fingerprint: &str) -> FeedItem {
render(pb::Event {
seq: 3,
at_unix_ms: 1,
kind: pb::EventKind::RefUpdated as i32,
payload: Some(pb::event::Payload::RefUpdated(pb::RefUpdated {
repo: Some(pb::RepoPath {
namespace: "vetra".into(),
repo: "gunnar".into(),
}),
ref_name: "refs/heads/main".into(),
new_oid: "0123456789abcdef0123".into(),
key_fingerprint: key_fingerprint.into(),
credential: credential.into(),
..Default::default()
})),
})
}
/// ★ **A push over smart HTTP is attributed** (GUN-014).
///
/// `key_fingerprint` is empty for every HTTP push — HTTP authenticates
/// with a token and there is no key to name — and the feed read only that
/// field, so a ref that moved over HTTP moved with nobody attached to it.
/// `RefUpdated.credential` has carried the answer since 2026-09-11 and
/// this console did not read it.
///
/// Seen RED before `by_credential`: the line was
/// `refs/heads/main moved to 0123456789` and held no `9f2c` at all.
#[test]
fn an_http_push_is_named_by_its_token_and_not_left_anonymous() {
let item = moved("token:9f2c4ab1d3", "");
assert!(
item.summary.contains("9f2c4ab1d3"),
"an HTTP push is unattributed on the feed: {}",
item.summary
);
assert!(
item.summary.contains("token"),
"…and the line must say it was a token, not imply a key: {}",
item.summary
);
}
/// ★ **An SSH push is named from the same field**, so there is one reader
/// and not two. The fingerprint is deliberately ABSENT here: a gunnar that
/// sends `credential` sends it for SSH too, and a console that only
/// managed to name a key when the older field was also filled in would be
/// green on today's wire and red on tomorrow's.
#[test]
fn an_ssh_push_is_named_from_the_credential_alone() {
let item = moved("key:SHA256:abcdefghijklmnop", "");
assert!(
item.summary.contains("SHA256:abcdefgh"),
"the key that pushed is not on the line: {}",
item.summary
);
assert!(
!item.summary.contains("key:"),
"the wire's scheme prefix is not for a reader: {}",
item.summary
);
}
/// ★ **The fallback is alive.** A gunnar older than the field sends no
/// `credential`, and the SSH pusher this console has always named must
/// still be named. Without this the "fix" would have swapped one
/// attribution hole for another, against a server that is still deployed.
#[test]
fn an_older_gunnar_is_still_read_through_the_fingerprint() {
let item = moved("", "SHA256:zyxwvutsrqpo");
assert!(
item.summary.contains("SHA256:zyxwvuts"),
"a gunnar with no `credential` lost the attribution it did send: {}",
item.summary
);
}
/// ★ **Nobody presented anything, so nothing is claimed** — and above all
/// no dangling "by". The identity value this whole set sits next to: every
/// assertion above would pass on a renderer that appended " by " to
/// everything.
#[test]
fn a_ref_that_moved_with_no_credential_says_nothing_about_one() {
let item = moved("", "");
assert_eq!(item.summary, "refs/heads/main moved to 0123456789");
}
/// ★ **A deletion is attributed by the same reader.** `RefDeleted` carries
/// the field for the same reason `RefUpdated` does, and a branch
/// disappearing is the event an operator is most likely to be looking for
/// somebody to attach to.
#[test]
fn a_deletion_names_the_credential_that_made_it() {
let item = render(pb::Event {
seq: 4,
at_unix_ms: 1,
kind: pb::EventKind::RefDeleted as i32,
payload: Some(pb::event::Payload::RefDeleted(pb::RefDeleted {
repo: Some(pb::RepoPath {
namespace: "vetra".into(),
repo: "gunnar".into(),
}),
ref_name: "refs/heads/doomed".into(),
credential: "token:1a2b3c4d".into(),
..Default::default()
})),
});
assert_eq!(item.kind, "ref_deleted");
assert!(
item.summary.contains("1a2b3c4d"),
"a deleted ref is unattributed: {}",
item.summary
);
}
#[test]
fn gunnars_own_gap_keeps_its_numbers() {
let item = render(pb::Event {
seq: 100,
at_unix_ms: 1,
kind: pb::EventKind::Gap as i32,
payload: Some(pb::event::Payload::Gap(pb::EventGap {
dropped: 12,
last_delivered_seq: 80,
resume_seq: 93,
epoch: "run-a".into(),
})),
});
assert_eq!(item.kind, "gap");
assert_eq!(item.dropped, Some(12));
assert!(
item.summary.contains("81 to 92"),
"the line must name the hole's own edges, not the cursor's: {}",
item.summary
);
}
/// ★ **A gap does not accuse the console of falling behind.**
///
/// This line used to open "the console fell behind", which was the only way
/// a gap could arrive when it was written. [`FROM_THE_RING`] added a second
/// way: a cold start asks gunnar for its whole ring, and a gunnar that has
/// published more than a ringful answers with a gap that says nothing about
/// this console's speed. The two are the same message on the wire, so the
/// rendering may not pick one cause and assert it.
#[test]
fn a_gap_reports_the_measurement_and_not_a_cause_it_cannot_know() {
let cold_start = render(pb::Event {
seq: 0,
at_unix_ms: 1,
kind: pb::EventKind::Gap as i32,
payload: Some(pb::event::Payload::Gap(pb::EventGap {
// What gunnar answers a `from_seq: 1` with when its ring has
// long since rolled: 1 998 events published and evicted.
dropped: 1998,
last_delivered_seq: FROM_THE_RING,
resume_seq: 2000,
epoch: "run-a".into(),
})),
});
assert!(
!cold_start.summary.contains("fell behind"),
"nothing here fell behind; gunnar's ring is bounded and this console had just \
started: {}",
cold_start.summary
);
assert!(
cold_start.summary.contains("1998"),
"…and the measured count must survive the rewording: {}",
cold_start.summary
);
assert!(
cold_start.summary.contains("2 to 1999"),
"…as must both edges of the hole: {}",
cold_start.summary
);
}
/// A gap with nothing left on the far side of it names one edge, not a
/// range that runs backwards.
///
/// `resume_seq` is 0 when gunnar holds nothing past the cursor — reachable
/// on a cold start against a gunnar whose ring is empty — and subtracting
/// one from it printed "sequence 2 to 0", naming an event that cannot exist
/// since sequences start at 1.
#[test]
fn a_gap_with_no_far_edge_does_not_invent_one() {
let item = render(pb::Event {
seq: 0,
at_unix_ms: 1,
kind: pb::EventKind::Gap as i32,
payload: Some(pb::event::Payload::Gap(pb::EventGap {
dropped: 40,
last_delivered_seq: 1,
resume_seq: 0,
epoch: "run-a".into(),
})),
});
assert!(
!item.summary.contains(" to 0"),
"sequences start at 1, so there is no event 0 to name: {}",
item.summary
);
assert!(
item.summary.contains("from sequence 2 onwards"),
"the open-ended case says so rather than guessing an upper edge: {}",
item.summary
);
assert_eq!(item.dropped, Some(40));
}
#[test]
fn the_prologue_is_never_shown_as_an_unrecognised_event() {
// It is normally consumed by the pump. If it ever reaches `render`, the
// console must name it rather than print "an event this console does
// not recognise" about the one message its cursor depends on.
let item = render(pb::Event {
seq: 0,
at_unix_ms: 1,
kind: pb::EventKind::StreamOpened as i32,
payload: Some(pb::event::Payload::StreamOpened(pb::StreamOpened {
epoch: "0123456789ab".into(),
last_seq: 42,
..Default::default()
})),
});
assert_eq!(item.kind, "stream_opened");
assert_ne!(item.kind, "unknown");
assert!(item.summary.contains("01234567"), "{}", item.summary);
}
#[tokio::test]
async fn a_stream_without_passkey_stays_open() {
let gunnar = crate::grpc::Gunnar::connect_lazy("http://127.0.0.1:1", std::sync::Arc::new(crate::signer::Signer::new(ssh_key::PrivateKey::from(ssh_key::private::Ed25519Keypair::from_seed(&[5; 32]))).expect("a signer"))).expect("a lazy channel");
assert!(still_authorised(&gunnar, None, None).await);
}
#[test]
fn byte_counts_read_as_sizes() {
assert_eq!(bytes(512), "512 B");
assert_eq!(bytes(1536), "1.5 KiB");
assert_eq!(bytes(5 * 1024 * 1024), "5.0 MiB");
}
#[test]
fn a_fingerprint_is_shortened_but_still_looks_like_one() {
assert_eq!(
short_fingerprint("SHA256:abcdefghijklmnop"),
"SHA256:abcdefgh"
);
}
/// One feed line, identified only by its sequence number: the tests below
/// are about which lines arrive and in what order, never about what one
/// says.
fn line(seq: u64) -> FeedItem {
FeedItem {
seq,
at_unix_ms: 0,
kind: "repo_created".into(),
repo: Some("a/b".into()),
summary: format!("made {seq}"),
dropped: None,
}
}
/// Everything a receiver has waiting, as sequence numbers.
///
/// `try_recv` rather than `recv`, so a test that expects nothing more does
/// not have to prove a negative by waiting for a timeout. A `Lagged` here
/// would mean a test published past [`FANOUT_CAPACITY`] and is a bug in the
/// test, not a result: it is named rather than swallowed.
fn drained(rx: &mut broadcast::Receiver<FeedItem>) -> Vec<u64> {
let mut seen = Vec::new();
loop {
match rx.try_recv() {
Ok(item) => seen.push(item.seq),
Err(broadcast::error::TryRecvError::Empty)
| Err(broadcast::error::TryRecvError::Closed) => return seen,
Err(broadcast::error::TryRecvError::Lagged(n)) => {
panic!("the fixture outran the fan-out by {n}; publish fewer than {FANOUT_CAPACITY}")
}
}
}
}
/// gunnar whose event stream ends at once, every time — counted.
#[derive(Clone, Default)]
struct CountingMain(Arc<std::sync::atomic::AtomicUsize>);
#[tonic::async_trait]
impl pb::events_server::Events for CountingMain {
async fn watch(&self, _: tonic::Request<pb::WatchRequest>) -> Result<tonic::Response<tonic::codegen::BoxStream<pb::Event>>, tonic::Status> {
self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(tonic::Response::new(Box::pin(tokio_stream::iter(Vec::<Result<pb::Event, tonic::Status>>::new()))))
}
}
/// ★ **A stopped hub subscribes to nobody** (stinkies, follow-up of
/// recharge-main-front-cut-off): gunnar-front's `console::start` builds a
/// new console — and a new hub — every time the fed primary changes, and
/// the replaced hub's pump went on re-subscribing to the OLD main for the
/// life of the process.
#[tokio::test]
async fn a_stopped_hub_stops_subscribing() {
let main = CountingMain::default();
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let at = l.local_addr().unwrap();
tokio::spawn(
tonic::transport::Server::builder()
.add_service(pb::events_server::EventsServer::new(main.clone()))
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(l)),
);
let signer = Arc::new(
crate::signer::Signer::new(ssh_key::PrivateKey::from(ssh_key::private::Ed25519Keypair::from_seed(&[9u8; 32]))).expect("a fixed seed makes a signer"),
);
let hub = EventHub::spawn(Gunnar::connect_lazy(&at.to_string(), signer).expect("an address"));
let t = std::time::Instant::now();
while main.0.load(std::sync::atomic::Ordering::SeqCst) == 0 && t.elapsed() < Duration::from_secs(10) {
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(main.0.load(std::sync::atomic::Ordering::SeqCst) >= 1, "the hub subscribed");
assert!(hub.pumping());
hub.stop();
hub.stop();
assert!(!hub.pumping(), "stopped, and stopping again is harmless");
let at_stop = main.0.load(std::sync::atomic::Ordering::SeqCst);
tokio::time::sleep(RECONNECT_DELAY * 3).await;
assert_eq!(main.0.load(std::sync::atomic::Ordering::SeqCst), at_stop, "a stopped hub subscribed to gunnar again");
}
#[tokio::test]
async fn a_published_item_reaches_a_subscriber() {
let hub = EventHub::detached();
let (_backlog, mut rx) = hub.subscribe_with_backlog();
hub.publish(line(1));
assert_eq!(rx.recv().await.expect("delivered").seq, 1);
}
/// ★ **A browser that connects is told what happened before it did.**
///
/// The reported defect, at its smallest: the overview showed "Events 147
/// published · the ring holds 1 024" beside "Nothing has happened since the
/// page was opened". A `broadcast::Receiver` starts at the tail, so every
/// line published before a tab opened was gone, and the console said so in
/// words that blamed the forge for being idle.
///
/// Red before the hub retained anything: the backlog was `[]` and the
/// operator was shown three pushes' worth of nothing.
#[test]
fn a_new_feed_is_handed_what_happened_before_it_connected() {
let hub = EventHub::detached();
for seq in 1..=3 {
hub.publish(line(seq));
}
let (backlog, _rx) = hub.subscribe_with_backlog();
assert_eq!(
backlog.iter().map(|i| i.seq).collect::<Vec<_>>(),
vec![1, 2, 3],
"a connection must start with the history the hub still holds, oldest first"
);
}
/// ★ **The seam: nothing is delivered twice and nothing falls between.**
///
/// The backlog and the receiver are two sources for one feed, and the join
/// between them is where an audit feed gets quietly wrong. Both directions
/// are asserted, because each failure alone looks like a working feed: a
/// repeat renders as two pushes that never happened, and a hole renders as
/// a push that did happen and is nowhere, counted by nothing.
#[test]
fn the_backlog_and_the_live_tail_join_with_no_repeat_and_no_hole() {
let hub = EventHub::detached();
for seq in 1..=3 {
hub.publish(line(seq));
}
let (backlog, mut rx) = hub.subscribe_with_backlog();
for seq in 4..=6 {
hub.publish(line(seq));
}
let live = drained(&mut rx);
assert_eq!(
live,
vec![4, 5, 6],
"the receiver repeated the backlog: 1..3 were sent before it existed"
);
let mut joined = backlog.iter().map(|i| i.seq).collect::<Vec<_>>();
joined.extend(live);
assert_eq!(
joined,
vec![1, 2, 3, 4, 5, 6],
"what the operator sees is the two halves in order, with every event once"
);
}
/// ★ **The same seam, with a publisher running while the connection opens.**
///
/// The test above passes on an implementation that snapshots the backlog
/// and then subscribes as two separate steps, because nothing publishes in
/// between when the test is the only thread. On a live box the pump is that
/// thread, and it is publishing during every tab open.
///
/// Both broken orders are caught by one assertion, which is why the
/// assertion is contiguity rather than an expected list:
///
/// * snapshot then subscribe — an event published between them is in
/// neither half, so the joined sequence skips a number;
/// * subscribe then snapshot — an event published between them is in both,
/// so the joined sequence repeats one.
///
/// A race is detected and not proved, so this runs the window repeatedly
/// and aligns the two threads on a barrier rather than hoping they overlap.
/// Seen red both ways round, by building each broken order against this
/// same body: releasing the lock before `tx.subscribe()` failed on round 3
/// with "110 then 112", a hole; taking the receiver before the snapshot
/// failed on round 12 with "107 then 107", a repeat. Neither survived
/// `ROUNDS`, which is why that number is what it is.
#[test]
fn the_seam_holds_while_the_pump_is_publishing_into_it() {
use std::sync::Barrier;
// Below `FANOUT_CAPACITY`, so a receiver in this test can never lag and
// a failure is always the seam rather than the fixture.
const BEFORE: u64 = 100;
const AFTER: u64 = 300;
// Enough rounds that a broken seam fails long before the last one, and
// few enough that the suite stays a suite. The number was 200 and cost
// most of half a minute on a loaded box for no extra detection.
const ROUNDS: u32 = 48;
for round in 0..ROUNDS {
let hub = EventHub::detached();
let barrier = Arc::new(Barrier::new(2));
let publisher = std::thread::spawn({
let hub = hub.clone();
let barrier = Arc::clone(&barrier);
move || {
for seq in 1..=BEFORE {
hub.publish(line(seq));
}
// Both threads leave here together, so the subscription is
// opened while the publishing is under way rather than
// before it starts or after it finishes.
barrier.wait();
for seq in BEFORE + 1..=BEFORE + AFTER {
hub.publish(line(seq));
std::thread::yield_now();
}
}
});
barrier.wait();
let (backlog, mut rx) = hub.subscribe_with_backlog();
publisher.join().expect("the publisher panicked");
let mut seen = backlog.iter().map(|i| i.seq).collect::<Vec<_>>();
seen.extend(drained(&mut rx));
for pair in seen.windows(2) {
assert_eq!(
pair[1],
pair[0] + 1,
"round {round}: the join is not exact — {} then {} (a repeat or a hole \
across the backlog/live boundary)",
pair[0],
pair[1]
);
}
assert_eq!(
seen.last().copied(),
Some(BEFORE + AFTER),
"round {round}: the receiver missed the end of the stream",
);
}
}
/// The backlog is a bounded buffer, and it keeps the NEWEST lines.
///
/// The bound is the point — this process must not grow because gunnar is
/// busy and nobody is watching — and which end it drops from is the part a
/// `pop_back` would get wrong while still passing a length assertion.
#[test]
fn the_backlog_is_bounded_and_keeps_the_newest() {
let hub = EventHub::detached();
for seq in 1..=(BACKLOG as u64 + 50) {
hub.publish(line(seq));
}
let (backlog, _rx) = hub.subscribe_with_backlog();
assert_eq!(backlog.len(), BACKLOG);
assert_eq!(
backlog.first().map(|i| i.seq),
Some(51),
"the oldest lines are the ones to drop"
);
assert_eq!(
backlog.last().map(|i| i.seq),
Some(BACKLOG as u64 + 50),
"…and the newest are the ones an operator opened the page for"
);
}
/// ★ **A pump that has processed nothing asks gunnar for its ring.**
///
/// `from_seq: 0` is not "I have seen nothing"; gunnar reads it as "start at
/// the live tail" and has a test of its own that keeps it that way. Sending
/// the number this process honestly holds therefore asked for the opposite
/// of what it wanted, and a console restarted at noon showed an empty feed
/// while gunnar sat on a ringful of the morning's pushes.
///
/// The last case is the control: a cursor that means something is sent
/// unchanged, or a reconnect would replay everything gunnar holds on top of
/// what this process already forwarded.
#[test]
fn a_pump_with_nothing_processed_asks_for_the_ring_and_not_for_the_tail() {
assert_ne!(
resume_from(None),
0,
"the first subscription must not ask for the live tail: 0 is how gunnar is told \
to forget its ring"
);
assert_eq!(resume_from(None), FROM_THE_RING);
assert_eq!(
resume_from(Some(&cursor("run-a", 0))),
FROM_THE_RING,
"an epoch change resets the cursor to 0, which means the same thing: this process \
has processed nothing in the run it is about to talk to"
);
assert_eq!(
resume_from(Some(&cursor("run-a", 900))),
900,
"a real cursor is sent unchanged, or a reconnect replays what was already shown"
);
}
/// **L1 (estate-box check 3, release-3 recharge)**: while gunnar is gone
/// (a main lost and laid again) or does not know the console's key yet (a
/// fresh main before its restore), the console re-subscribed every 3 s and
/// said `cannot subscribe to gunnar's event stream` at WARN every time —
/// 7+/min on vm1, and every attempt was a `grpc.auth.refused` WARN on the
/// new main too. A run of failures now backs off (3 s doubling to a
/// minute) and is said ONCE, again only when the reason changes or every
/// five minutes while it lasts, and its end once at INFO.
#[test]
fn a_gunnar_that_cannot_be_subscribed_to_is_asked_less_and_said_once() {
let t0 = std::time::Instant::now();
let mut b = Resubscribe::default();
let (mut at, mut attempts, mut said) = (Duration::ZERO, 0u32, Vec::new());
// Two minutes gone, then two minutes that do not know the key.
while at < Duration::from_secs(240) {
let code = if at < Duration::from_secs(120) { tonic::Code::Unavailable } else { tonic::Code::Unauthenticated };
let (wait, line) = b.failed(code, "x", t0 + at);
if at < Duration::from_secs(60) {
attempts += 1;
}
said.extend(line);
at += wait;
}
assert!(attempts <= 6, "{attempts} attempts in the first minute (the gate's WARN rate is > 6/min)");
assert_eq!(said.len(), 2, "once per reason: {said:#?}");
assert!(said[1].1.contains("Unauthenticated") || said[1].1.contains("credentials"), "{said:#?}");
assert!(said[0].0 && !said[1].0, "gunnar gone is a WARN; a key it does not know yet is a wait, at INFO: {said:#?}");
let back = b.opened(t0 + at).expect("the end of the run is said");
assert!(back.contains("subscribed again"), "{back}");
assert!(b.opened(t0 + at).is_none(), "said once");
// A stream that opened resets the wait.
assert_eq!(b.failed(tonic::Code::Unavailable, "x", t0 + at).0, RECONNECT_DELAY);
}
/// **The upgrade's gate (lifecycle lf10, gate@upgrade RED, 2026-10-01)**:
/// while main is stopped, re-installed and booted by `estate upgrade`, the
/// front's channel to it still accepts the call (the relay answers) and the
/// stream then ends at once — `the event stream from gunnar ended … h2
/// protocol error` at WARN on EVERY attempt, 7 in 60 s, a WARN-spam offence
/// on the front during a PLANNED outage. A stream that ends before it
/// delivered anything is a failed attempt like a refused subscribe: it
/// backs off, is said once (WARN, gunnar is gone), again only when the
/// reason changes or every five minutes, and the run's end once at INFO.
#[test]
fn a_stream_that_ends_before_any_event_is_a_failed_attempt_said_once() {
let t0 = std::time::Instant::now();
let mut b = Resubscribe::default();
let (mut at, mut attempts, mut said) = (Duration::ZERO, 0u32, Vec::new());
// Main's upgrade: ten minutes of streams that open and end at once.
while at < Duration::from_secs(600) {
let (wait, line) = b.ended(tonic::Code::Unknown, "h2 protocol error: error reading a body from connection", false, t0 + at);
if at < Duration::from_secs(60) {
attempts += 1;
}
said.extend(line);
at += wait;
}
assert!(attempts <= 6, "{attempts} attempts in the first minute (the gate's WARN rate is > 6/min)");
assert!(said.len() <= 2 && said.iter().all(|(warn, _)| *warn), "once, and again after five minutes at most: {said:#?}");
assert!(said[0].1.contains("ended"), "{said:#?}");
let back = b.opened(t0 + at).expect("the end of the run is said");
assert!(back.contains("subscribed again"), "{back}");
// A stream that DID deliver and then ended starts a run of its own: said once.
let (w, line) = b.ended(tonic::Code::Unknown, "h2 protocol error", true, t0 + at);
assert_eq!(w, RECONNECT_DELAY);
assert!(line.is_some_and(|(warn, l)| warn && l.contains("ended")));
assert!(b.ended(tonic::Code::Unknown, "h2 protocol error", false, t0 + at + RECONNECT_DELAY).1.is_none(), "the same run, said already");
}
}