gunnar-grpc 2.0.0

//! 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 build runner 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());
    }
}