crates/gunnar-grpc/src/bus.rs
//! The event bus.
//!
//! One `tokio::sync::broadcast` channel, one monotonic sequence counter, and a
//! per-subscriber [`EventStream`] that turns the channel's lag signal into an
//! explicit [`EventGap`] on the wire.
//!
//! ## Why a lagging subscriber is told, not buffered
//!
//! A UI that stalls (a laptop asleep, a tab throttled, a korp panel behind a
//! slow link) must not be able to make the server grow. `broadcast` is a ring:
//! past capacity the *oldest* entries are overwritten and the slow receiver is
//! told how many it missed. That is the correct trade for a server that may be
//! PID 1 in a distroless image (the no-OS server rule): bounded memory, and honest about the loss.
//!
//! The alternative, an unbounded queue per subscriber, turns one asleep laptop
//! into an OOM. It is not available here by construction, not by discipline.
//!
//! ## What a subscriber does with a gap
//!
//! [`EventGap`] carries `dropped`, the last sequence number known delivered,
//! and `resume_seq`, the sequence the stream picks up at. A UI that sees one
//! re-fetches state over `Repos`/`Refs` rather than trusting its incremental
//! view. That is why the gap names both edges of the hole.
//!
//! ## Resuming, and the epoch that says whether a cursor means anything
//!
//! A subscriber that reconnects used to have two questions it could not answer:
//! "did I miss anything while I was away?" and "are these sequence numbers even
//! from the run I was talking to?". Both are answered here.
//!
//! * The bus keeps its most recent `capacity` events in [`EventBus::recent`],
//! alongside the broadcast ring, so [`EventBus::subscribe_from`] can replay
//! what a cursor asks for. It is the SAME bound — a resume can never make the
//! server hold more than a ringful, which is the property the whole design
//! rests on. There is no durable log and this is not one.
//! * Every subscription opened that way begins with a [`StreamOpened`]
//! prologue carrying an epoch minted once per [`EventBus`]. A restart is a
//! new bus and therefore a new epoch, and a cursor from a different epoch is
//! void — the server cannot detect that on its own, because seq 900 of the
//! last run and seq 900 of this one are the same number.
use std::collections::VecDeque;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::{SystemTime, UNIX_EPOCH};
use gunnar_proto::{event, Event, EventGap, EventKind, RepoPath, StreamOpened, WatchRequest};
use tokio::sync::broadcast;
use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::Stream;
/// Milliseconds since the Unix epoch. Saturating, because an appliance whose
/// clock has not been set yet must still be able to emit events.
pub fn now_unix_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis().min(i64::MAX as u128) as i64)
.unwrap_or(0)
}
/// A snapshot of the bus for `Admin.Stats`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BusStats {
pub subscribers: u64,
pub published: u64,
/// Total events lost to lag, summed over every subscriber. The number that
/// says the buffer is too small.
pub dropped: u64,
pub capacity: u64,
}
/// The publish side of the event stream.
///
/// `receive-pack` will publish `RefUpdated` / `RefDeleted` / `PushRejected` /
/// `PushAttestation` here once `gunnar-wire` lands; the control-plane services
/// in this crate already publish `RepoCreated` / `RepoDeleted`. Publishing is
/// deliberately infallible from the caller's point of view: a push must never
/// fail because nobody was watching.
#[derive(Debug)]
pub struct EventBus {
tx: broadcast::Sender<Arc<Event>>,
seq: AtomicU64,
published: AtomicU64,
dropped: Arc<AtomicU64>,
capacity: usize,
/// Identifies this run of this bus. Minted once, in [`EventBus::new`], and
/// never mutated: a restart is a new process, a new `EventBus` and
/// therefore a new epoch. See [`new_epoch`].
epoch: Arc<str>,
/// The last `capacity` events, so a `from_seq` can be replayed.
///
/// The same bound as the broadcast ring, deliberately: the resume path must
/// not be a second place the server can grow. `publish` holds this lock
/// across its `tx.send`, and a subscription takes it across its
/// `tx.subscribe`, which is what makes the split between "replayed" and
/// "live" exact — no event can be both, and none can fall between them.
recent: Mutex<VecDeque<Arc<Event>>>,
}
/// A value that is different in every run of the process.
///
/// No new dependency for this (the appliance crate budget): the wall clock
/// in nanoseconds, the process id, and a counter so two buses built in the same
/// process are still distinguishable. A restart moves the clock and usually the
/// pid; nothing here can repeat across boots short of a clock that never ticks.
fn new_epoch() -> Arc<str> {
static SEQUENCE: AtomicU64 = AtomicU64::new(0);
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
let pid = u64::from(std::process::id());
let n = SEQUENCE.fetch_add(1, Ordering::Relaxed);
Arc::from(format!("{nanos:016x}{pid:08x}{n:04x}"))
}
impl EventBus {
pub fn new(capacity: usize) -> Self {
let capacity = capacity.max(1);
let (tx, _rx) = broadcast::channel(capacity);
Self {
tx,
seq: AtomicU64::new(0),
published: AtomicU64::new(0),
dropped: Arc::new(AtomicU64::new(0)),
capacity,
epoch: new_epoch(),
recent: Mutex::new(VecDeque::with_capacity(capacity)),
}
}
/// Publish one event, returning it with its assigned sequence and time.
///
/// Never fails and never blocks. With no subscribers the event is dropped
/// on the floor. The event stream is a notification channel, not a log; the
/// durable record of a push is the attestation line (push attestation), not this.
pub fn publish(&self, payload: event::Payload) -> Arc<Event> {
// Sequence assignment, retention and broadcast all happen under one
// lock. Not for the counter's sake — it is atomic — but so that a
// subscription opening concurrently sees a consistent split: everything
// in `recent` was already sent, and everything sent after it will reach
// the new receiver. Without that, a resuming subscriber could be handed
// the same event twice, or neither copy.
let mut recent = self.recent.lock().unwrap_or_else(|e| e.into_inner());
let seq = self.seq.fetch_add(1, Ordering::SeqCst) + 1;
let event = Arc::new(Event::new(seq, now_unix_ms(), payload));
self.published.fetch_add(1, Ordering::Relaxed);
if recent.len() == self.capacity {
recent.pop_front();
}
recent.push_back(Arc::clone(&event));
let _ = self.tx.send(Arc::clone(&event));
event
}
/// Subscribe with a filter built from a `WatchRequest`.
///
/// **Live only, and no prologue**: the stream starts at the tail and the
/// first thing on it is the next event published. This is what the
/// in-process subscribers want — the replicator and the webhooks are
/// started once with the server and have no cursor to resume from.
///
/// The returned stream yields only events matching the filter, except
/// [`EventKind::Gap`], which is synthesised per subscriber and is never
/// filtered out. A subscriber that filtered away its own gap marker would
/// silently believe it had seen everything.
pub fn subscribe(&self, filter: EventFilter) -> EventStream {
self.open(filter, None)
}
/// Subscribe the way `Events.Watch` does: a [`StreamOpened`] prologue
/// first, then whatever `from_seq` can still be replayed, then the live
/// tail.
///
/// `from_seq` is the sequence of the last event the subscriber processed;
/// 0 means "no cursor, start live". The promise, and its limits, are
/// written on `WatchRequest.from_seq` in `events.proto` — this is the code
/// that keeps it:
///
/// * everything still buffered past the cursor is delivered before any
/// newly published event;
/// * a cursor older than the oldest buffered event gets an [`EventGap`]
/// FIRST, naming how many events fell out of the ring;
/// * a cursor at or past the newest event replays nothing, and that is not
/// a gap.
pub fn subscribe_from(&self, filter: EventFilter, from_seq: u64) -> EventStream {
self.open(filter, Some(from_seq))
}
/// The one subscription path both public forms route through (one implementation).
///
/// `resume` is `None` for a live-only in-process subscriber and `Some` for
/// a `Watch` stream, where `Some(0)` still earns the prologue: a client
/// with no cursor yet is exactly the client that most needs to be told
/// which epoch it is about to start counting in.
fn open(&self, filter: EventFilter, resume: Option<u64>) -> EventStream {
let recent = self.recent.lock().unwrap_or_else(|e| e.into_inner());
let rx = self.tx.subscribe();
let last_seq = self.seq.load(Ordering::SeqCst);
let mut prologue = VecDeque::new();
if let Some(from_seq) = resume {
let oldest = recent.front().map(|e| e.seq).unwrap_or(0);
let replay: Vec<Arc<Event>> = if from_seq == 0 {
Vec::new()
} else {
recent
.iter()
.filter(|e| e.seq > from_seq)
.cloned()
.collect()
};
// What fell out of the ring between the cursor and the oldest thing
// still held. `oldest == 0` means the buffer is empty, so anything
// published after the cursor is gone rather than merely evicted.
let dropped = if from_seq == 0 {
0
} else if oldest == 0 {
last_seq.saturating_sub(from_seq)
} else {
oldest.saturating_sub(from_seq).saturating_sub(1)
};
prologue.push_back(Event::new(
0,
now_unix_ms(),
event::Payload::StreamOpened(StreamOpened {
epoch: self.epoch.to_string(),
last_seq,
oldest_buffered_seq: oldest,
replaying: replay.len() as u64,
from_seq,
}),
));
if dropped > 0 {
self.dropped.fetch_add(dropped, Ordering::Relaxed);
tracing::warn!(
dropped,
from_seq,
oldest_buffered_seq = oldest,
"a subscriber asked to resume from before the oldest buffered event; the \
bus is a bounded ring and has no log, so it is being told what it lost \
rather than promised a replay it cannot have."
);
prologue.push_back(Event::new(
0,
now_unix_ms(),
event::Payload::Gap(EventGap {
dropped,
last_delivered_seq: from_seq,
// The first event the replay actually delivers. 0 when
// there is none left to deliver at all.
resume_seq: replay.first().map(|e| e.seq).unwrap_or(0),
epoch: self.epoch.to_string(),
}),
));
}
// Shaped and filtered exactly as a live event would be, so a
// replayed event and the same event delivered live are the same
// bytes (one implementation: one writer, `EventFilter`).
for event in replay {
if filter.matches(&event) {
prologue.push_back(filter.shape(&event));
}
}
}
drop(recent);
EventStream {
inner: BroadcastStream::new(rx),
filter,
prologue,
queued: None,
last_seq,
dropped_total: Arc::clone(&self.dropped),
epoch: Arc::clone(&self.epoch),
}
}
/// The epoch every stream from this bus reports. Different in every run.
pub fn epoch(&self) -> &str {
&self.epoch
}
/// The oldest sequence still replayable, or 0 when nothing is buffered.
pub fn oldest_buffered_seq(&self) -> u64 {
self.recent
.lock()
.unwrap_or_else(|e| e.into_inner())
.front()
.map(|e| e.seq)
.unwrap_or(0)
}
pub fn stats(&self) -> BusStats {
BusStats {
subscribers: self.tx.receiver_count() as u64,
published: self.published.load(Ordering::Relaxed),
dropped: self.dropped.load(Ordering::Relaxed),
capacity: self.capacity as u64,
}
}
/// The sequence number most recently assigned. 0 before anything is
/// published.
pub fn last_seq(&self) -> u64 {
self.seq.load(Ordering::SeqCst)
}
}
impl Default for EventBus {
fn default() -> Self {
Self::new(crate::config::DEFAULT_EVENT_BUFFER)
}
}
/// Which events a subscriber wants (the event stream: filterable by namespace / repo /
/// kind).
///
/// Empty on an axis means "no restriction on that axis"; across axes the
/// filters are ANDed. An entirely empty filter watches everything.
#[derive(Debug, Clone, Default)]
pub struct EventFilter {
pub namespaces: Vec<String>,
pub repos: Vec<RepoPath>,
pub kinds: Vec<EventKind>,
/// Attestations carry every new object id and can be large, so the id list
/// is opt-in even for a subscriber that asked for attestations (push attestation).
pub include_attestation_object_ids: bool,
}
impl EventFilter {
pub fn from_request(req: &WatchRequest) -> Self {
Self {
namespaces: req.namespaces.clone(),
repos: req.repos.clone(),
kinds: req
.kinds
.iter()
.filter_map(|k| EventKind::try_from(*k).ok())
.filter(|k| *k != EventKind::Unspecified)
.collect(),
include_attestation_object_ids: req.include_attestation_object_ids,
}
}
/// True when this subscriber should see `event`.
///
/// A gap and the stream prologue always match: both are facts about the
/// STREAM rather than about a repository, and a subscriber that filtered
/// away its own gap — or the epoch it needs to read its own cursor by —
/// would silently believe it had seen everything. Everything else must pass
/// every non-empty axis.
pub fn matches(&self, event: &Event) -> bool {
if event.kind == EventKind::Gap as i32 || event.kind == EventKind::StreamOpened as i32 {
return true;
}
if !self.kinds.is_empty() && !self.kinds.iter().any(|k| *k as i32 == event.kind) {
return false;
}
if !self.namespaces.is_empty() {
match event.repo_path() {
Some(path) if self.namespaces.contains(&path.namespace) => {}
_ => return false,
}
}
if !self.repos.is_empty() {
match event.repo_path() {
Some(path)
if self
.repos
.iter()
.any(|r| r.namespace == path.namespace && r.repo == path.repo) => {}
_ => return false,
}
}
true
}
/// Strip anything this subscriber did not ask for. Today that is only the
/// attestation object-id list.
fn shape(&self, event: &Event) -> Event {
let mut event = event.clone();
if !self.include_attestation_object_ids {
if let Some(event::Payload::PushAttestation(att)) = event.payload.as_mut() {
// The COUNT stays. A subscriber must still be able to see that
// 12 000 objects arrived without being sent 12 000 oids.
att.new_object_oids.clear();
// And it must be able to tell THIS empty list from the one a
// push of zero objects carries, and from the one a store that
// never enumerated the set carries. Stripping without saying so
// would put a third meaning on the same empty field, which is
// the whole disease `new_object_oids_unknown` was added for.
// A reason already there is left alone: it is the truer one —
// the server had no list to strip.
if att.new_object_oids_unknown.is_empty() {
att.new_object_oids_unknown = "this subscription did not set \
WatchRequest.include_attestation_object_ids, so the object ids were \
stripped from this event; the count and the byte total are complete"
.to_owned();
}
}
}
event
}
}
/// One subscriber's view of the bus.
///
/// Pull-driven: nothing is copied out of the broadcast ring until the consumer
/// polls. That is what keeps the memory bound in the ring where it belongs,
/// and it is why the lag path is deterministic enough to test.
pub struct EventStream {
inner: BroadcastStream<Arc<Event>>,
filter: EventFilter,
/// Everything that must go out before the live tail, in order: the
/// `StreamOpened` prologue, then a resume gap if the cursor fell out of the
/// ring, then the replayed events. Empty for a live-only subscription.
prologue: VecDeque<Event>,
/// The event that follows a gap, held back for one poll so the gap can be
/// delivered first and can name the sequence the stream resumes at.
queued: Option<Event>,
last_seq: u64,
dropped_total: Arc<AtomicU64>,
epoch: Arc<str>,
}
impl EventStream {
fn gap(&mut self, dropped: u64, resume_seq: u64) -> Event {
let last_delivered_seq = self.last_seq;
Event::new(
0,
now_unix_ms(),
event::Payload::Gap(EventGap {
dropped,
last_delivered_seq,
resume_seq,
epoch: self.epoch.to_string(),
}),
)
}
}
impl Stream for EventStream {
type Item = Result<Event, tonic::Status>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
// The prologue outranks everything, including a lag gap: a client must
// learn the epoch before it is asked to interpret a sequence number.
if let Some(event) = this.prologue.pop_front() {
return Poll::Ready(Some(Ok(event)));
}
if let Some(event) = this.queued.take() {
return Poll::Ready(Some(Ok(event)));
}
loop {
match Pin::new(&mut this.inner).poll_next(cx) {
Poll::Ready(Some(Ok(event))) => {
this.last_seq = event.seq;
if this.filter.matches(&event) {
return Poll::Ready(Some(Ok(this.filter.shape(&event))));
}
// Filtered out: keep draining rather than returning
// Pending, or a subscriber with a narrow filter would stall
// behind traffic it does not want.
continue;
}
Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(dropped)))) => {
this.dropped_total.fetch_add(dropped, Ordering::Relaxed);
tracing::warn!(
dropped,
last_delivered_seq = this.last_seq,
"event subscriber lagged; dropping the oldest events and sending a gap \
marker. Memory is bounded by the bus ring, never by this subscriber."
);
// A lag means the ring overwrote unread slots, so retained
// events are available RIGHT NOW. Take the first one so the
// gap can name the resume point. If it is somehow not ready
// the gap still goes out, with resume_seq = 0 meaning
// "unknown". A gap must never be swallowed.
let resume = match Pin::new(&mut this.inner).poll_next(cx) {
Poll::Ready(Some(Ok(event))) => Some(event),
_ => None,
};
let resume_seq = resume.as_ref().map(|e| e.seq).unwrap_or(0);
let gap = this.gap(dropped, resume_seq);
if let Some(event) = resume {
this.last_seq = event.seq;
if this.filter.matches(&event) {
this.queued = Some(this.filter.shape(&event));
}
}
return Poll::Ready(Some(Ok(gap)));
}
Poll::Ready(None) => return Poll::Ready(None),
Poll::Pending => return Poll::Pending,
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use gunnar_proto::{RepoDeleted, RepoPath};
fn deleted(namespace: &str, repo: &str) -> event::Payload {
event::Payload::RepoDeleted(RepoDeleted {
repo: Some(RepoPath::new(namespace, repo)),
principal_id: String::new(),
})
}
#[test]
fn publish_assigns_monotonic_sequences_starting_at_one() {
let bus = EventBus::new(8);
assert_eq!(bus.last_seq(), 0, "nothing published yet");
assert_eq!(bus.publish(deleted("ns", "a")).seq, 1);
assert_eq!(bus.publish(deleted("ns", "b")).seq, 2);
assert_eq!(bus.last_seq(), 2);
}
#[test]
fn published_event_kind_agrees_with_its_payload() {
let bus = EventBus::new(8);
let event = bus.publish(deleted("ns", "a"));
assert!(event.kind_matches_payload());
assert_eq!(event.kind, EventKind::RepoDeleted as i32);
}
#[test]
fn filter_matches_are_not_vacuous() {
// Measure, never assert: a filter that matched everything would pass a happy-path test.
// Assert BOTH directions on every axis.
let bus = EventBus::new(8);
let event = bus.publish(deleted("alpha", "one"));
let empty = EventFilter::default();
assert!(empty.matches(&event), "an empty filter watches everything");
let ns_hit = EventFilter {
namespaces: vec!["alpha".into()],
..Default::default()
};
let ns_miss = EventFilter {
namespaces: vec!["beta".into()],
..Default::default()
};
assert!(ns_hit.matches(&event));
assert!(!ns_miss.matches(&event), "namespace filter must exclude");
let repo_hit = EventFilter {
repos: vec![RepoPath::new("alpha", "one")],
..Default::default()
};
let repo_miss = EventFilter {
repos: vec![RepoPath::new("alpha", "two")],
..Default::default()
};
assert!(repo_hit.matches(&event));
assert!(!repo_miss.matches(&event), "repo filter must exclude");
let kind_hit = EventFilter {
kinds: vec![EventKind::RepoDeleted],
..Default::default()
};
let kind_miss = EventFilter {
kinds: vec![EventKind::RefUpdated],
..Default::default()
};
assert!(kind_hit.matches(&event));
assert!(!kind_miss.matches(&event), "kind filter must exclude");
}
#[test]
fn a_gap_survives_every_filter() {
let gap = Event::new(
0,
0,
event::Payload::Gap(EventGap {
dropped: 3,
last_delivered_seq: 1,
resume_seq: 5,
epoch: "an-epoch".to_owned(),
}),
);
let narrow = EventFilter {
namespaces: vec!["nothing-matches-this".into()],
repos: vec![RepoPath::new("no", "such")],
kinds: vec![EventKind::KeyUsed],
include_attestation_object_ids: false,
};
assert!(
narrow.matches(&gap),
"a subscriber must never filter away its own gap marker"
);
let opened = Event::new(
0,
0,
event::Payload::StreamOpened(StreamOpened {
epoch: "an-epoch".to_owned(),
..Default::default()
}),
);
assert!(
narrow.matches(&opened),
"…nor the prologue that tells it which run its sequence numbers belong to"
);
// The other direction, so neither assertion above is satisfied by a
// `matches` that says yes to everything.
let ordinary = Event::new(
1,
0,
event::Payload::RepoDeleted(RepoDeleted {
repo: Some(RepoPath::new("nordisk", "gunnar")),
principal_id: String::new(),
}),
);
assert!(
!narrow.matches(&ordinary),
"this filter matches no repository and no kind, so an ordinary event must be excluded"
);
}
#[test]
fn an_epoch_is_never_the_same_twice() {
// Measure, never assert: an epoch pinned to a constant would satisfy every "the field
// is populated" check and report no restart ever.
let a = new_epoch();
let b = new_epoch();
assert_ne!(a, b, "two epochs minted in a row must differ");
assert!(!a.is_empty());
}
}