crates/gunnar-grpc/src/service/events.rs
//! `Events.Watch`, the server-streaming feed (the event stream).
//!
//! Almost the whole implementation is "build a filter, subscribe to the bus,
//! box the stream". Everything interesting (bounded memory, the gap marker for
//! a lagging subscriber, the `StreamOpened` prologue and the `from_seq` replay)
//! lives in [`crate::bus`], because that behaviour has to be testable without a
//! gRPC connection in the way.
//!
//! # The exception: a stream outlives the request that opened it
//!
//! Every other RPC in the control plane is a request, a check and an answer, so
//! "authorise in the handler" is the whole of it. This one is a request, a
//! check, and then an open pipe that keeps delivering for as long as the client
//! keeps reading — hours, in the console's case. `caller`/`require` ran once,
//! when the pipe was opened, and after that the stream carried every event on
//! the server to whoever had once held a server-wide `Read`. Revoking the key,
//! deleting the principal, dropping the grant: all of them took effect for the
//! NEXT `Watch` and for none of the open ones.
//!
//! Nothing gunnar had would close it either. `Watch` is server-scoped on
//! purpose (see the handler), so this is the broadest read in the control plane
//! — pushes, deletions, repository names across every namespace, including the
//! ones the ex-subscriber never had access to.
//!
//! [`Revalidated`] is the answer, and the property it enforces is about DATA
//! rather than about time: **no event is handed to a client across a check
//! older than [`RECHECK`]**. See its own doc for why that is the shape rather
//! than a timer.
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use gunnar_proto::events_server::Events;
use gunnar_proto::{Event, WatchRequest};
use tokio_stream::Stream;
use tonic::{Request, Response, Status};
use crate::auth::{Access, Caller};
use crate::bus::EventFilter;
use crate::service::{EventsService, ServiceContext};
use gunnar_proto::GrantLevel;
/// **How stale the authorisation behind a delivered event may be.**
///
/// Thirty seconds. The cost of the check is one ACL read against a control
/// plane the same process is already reading on every RPC, and it is paid at
/// most twice a minute per subscriber, so the number could be smaller; it is
/// not, because a revocation that takes half a minute to reach an open pipe is
/// indistinguishable from one that takes no time at all to anybody who is not
/// holding a stopwatch, and a per-event check on a hot bus is a read per event.
pub const RECHECK: Duration = Duration::from_secs(30);
/// A `Watch` stream that keeps asking whether its subscriber may still watch.
///
/// # Why the check hangs off delivery and not off a timer
///
/// A timer would need a wakeup of its own, and would then be racing the thing
/// it guards: a revoked subscriber on a busy bus could be handed a hundred
/// events between the revocation and the tick. Checking on the delivery path
/// instead makes the guarantee a statement about what crosses the wire, which
/// is what the finding is about — **no event reaches the client unless an
/// authorisation no older than [`RECHECK`] said it could**. A quiet stream is
/// not re-checked and does not need to be: it is delivering nothing.
///
/// A failed check ends the stream, it does not skip an event. A subscriber
/// whose grant is gone must be told, in the status it was refused with, rather
/// than left holding a pipe that has silently stopped carrying — which is what
/// `EventFilter` validation up in the handler already refuses to do for a
/// mistyped namespace, for the same reason.
struct Revalidated {
inner: crate::bus::EventStream,
ctx: ServiceContext,
caller: Caller,
/// What this subscriber was let in on, re-asked at every check.
scope: WatchScope,
/// When the next delivery must re-check. Set to now at construction, so the
/// FIRST event out of a long replay is covered too.
next_check: Instant,
/// [`RECHECK`], as a field so the constant lives in one place and a test
/// can ask for a check on every delivery rather than sleeping for half a
/// minute to observe one.
recheck: Duration,
/// Set once the check has failed. A `Stream` must not be polled for more
/// after it has yielded its terminal error, and the refusal is terminal.
refused: bool,
}
/// **What a watcher was let in on** (owner, 2026-10-03: an account's emitted
/// events are not public by default, "such things has to be set").
#[derive(Debug, Clone, PartialEq, Eq)]
enum WatchScope {
/// A server reader: every event its filter lets through, as always.
Server,
/// Named accounts and nothing else — the request's `namespaces`, which for
/// this watcher are a BOUNDARY, not a filter. `outside` are the ones it is
/// not inside (watched because their events are public): an event about a
/// repository there reaches it only when it may read that repository, so a
/// private repository's events never leave the account.
Accounts {
names: Vec<String>,
outside: Vec<String>,
},
/// **The caller's own** (2026-10-04, the thin layer): a non-server reader
/// that named NO account. It cannot name the accounts it is inside — that
/// is gunnar's to know — so the boundary is decided per EVENT: an event
/// about a repository reaches it when the caller is inside that account
/// (`Read` over its namespace, gates honoured), or when the account's
/// events are public and the caller may read that repository. The
/// stream's own bookkeeping (`StreamOpened`, `EventGap`) reaches it; an
/// event about no repository reaches it only when it is a `KeyUsed` of the
/// caller's own principal.
Mine,
}
/// The one sentence for an account whose events are not shown to this caller —
/// the same whether the account exists or not, so a refusal tells an outsider
/// nothing about which names are taken.
fn not_shown(account: &str) -> Status {
Status::permission_denied(format!(
"Events.Watch: the events of {account} are shown only inside the account, unless its \
account admin makes them public"
))
}
impl WatchScope {
/// Decide (or re-decide) a watch. `Server` when the caller holds a
/// server-wide `Read`; otherwise every named account must be one the caller
/// is inside (`Read` over its namespace) or one whose events are public.
fn decide(
ctx: &ServiceContext,
caller: &Caller,
namespaces: &[String],
) -> Result<WatchScope, Status> {
let server = ctx.require(caller, Access::Server, GrantLevel::Read, "Events.Watch");
match server {
Ok(_) => return Ok(WatchScope::Server),
Err(status) if status.code() != tonic::Code::PermissionDenied => return Err(status),
// Naming no account: the caller's own. Let in on the caller's own
// account (`Read` over the namespace named after it) — which is
// true for any principal that exists and is not locked out — so a
// removed or gated principal is refused here, in gunnar's words,
// and the same question is the re-check below. A MEMBER has no
// account of his own (2026-10-07): his is the account that
// enrolled him, read on the `Read` enrolment gave him.
Err(_) if namespaces.is_empty() => {
let own = ctx
.employer_of(&caller.principal_id)?
.unwrap_or_else(|| caller.principal_id.clone());
ctx.require(
caller,
Access::namespace(&own),
GrantLevel::Read,
"Events.Watch",
)?;
return Ok(WatchScope::Mine);
}
Err(_) => {}
}
let mut outside = Vec::new();
for account in namespaces {
let inside = ctx
.control
.may(
&caller.principal_id,
&Access::namespace(account),
GrantLevel::Read,
)
.map_err(|err| {
Status::unavailable(format!("the identity table could not be read: {err}"))
})?;
if inside {
continue;
}
let public = ctx
.control
.namespace_for_outsider(&caller.principal_id, account)?
.is_some_and(|ns| ns.events_public);
if !public {
return Err(not_shown(account));
}
outside.push(account.clone());
}
Ok(WatchScope::Accounts {
names: namespaces.to_vec(),
outside,
})
}
/// Is `event` one this watcher may be handed? A server reader: yes. An
/// account watcher: yes, unless it is about a repository in an account it
/// is outside of and may not read.
fn admits(&self, ctx: &ServiceContext, caller: &Caller, event: &Event) -> bool {
if let WatchScope::Mine = self {
return Self::mine_admits(ctx, caller, event);
}
let WatchScope::Accounts { outside, .. } = self else {
return Self::server_admits(ctx, caller, event);
};
match event.repo_path() {
Some(path) if outside.contains(&path.namespace) => ctx
.control
.may(&caller.principal_id, &Access::repo(path), GrantLevel::Read)
.unwrap_or(false),
_ => true,
}
}
}
impl WatchScope {
/// **A server reader's per-event boundary** (2026-10-05, the operator
/// sees aggregates only): everything, except what happens inside a
/// customer's account its server grant does not reach — that it is shown
/// as any outsider is, only when the account admin made the events public.
/// A lookup that fails is a `false`.
fn server_admits(ctx: &ServiceContext, caller: &Caller, event: &Event) -> bool {
use gunnar_proto::event::Payload;
match event.repo_path() {
Some(path) => match ctx.shielded(caller, &path.namespace) {
Ok(false) => true,
Ok(true) => Self::mine_admits(ctx, caller, event),
Err(_) => false,
},
None => match event.payload.as_ref() {
Some(Payload::KeyUsed(used)) => {
match ctx.account_of_principal(&used.principal_id) {
Some(account) => !ctx.shielded(caller, &account).unwrap_or(true),
None => true,
}
}
_ => true,
},
}
}
/// [`WatchScope::Mine`]'s per-event boundary. A lookup that fails is a
/// `false`: an event withheld is recoverable, one leaked is not.
fn mine_admits(ctx: &ServiceContext, caller: &Caller, event: &Event) -> bool {
use gunnar_proto::event::Payload;
match event.payload.as_ref() {
Some(Payload::StreamOpened(_)) | Some(Payload::Gap(_)) => return true,
_ => {}
}
let Some(path) = event.repo_path() else {
return matches!(
event.payload.as_ref(),
Some(Payload::KeyUsed(used)) if used.principal_id == caller.principal_id
);
};
let inside = ctx
.control
.may(
&caller.principal_id,
&Access::namespace(&path.namespace),
GrantLevel::Read,
)
.unwrap_or(false);
if inside {
return true;
}
let public = ctx
.control
.namespace_for_outsider(&caller.principal_id, &path.namespace)
.ok()
.flatten()
.is_some_and(|ns| ns.events_public);
public
&& ctx
.control
.may(&caller.principal_id, &Access::repo(path), GrantLevel::Read)
.unwrap_or(false)
}
}
impl Stream for Revalidated {
type Item = Result<Event, Status>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
loop {
if this.refused {
return Poll::Ready(None);
}
// Poll first: a stream with nothing to hand over has nothing to check
// an authorisation FOR, and re-checking a parked subscriber twice a
// minute would be a control-plane read per idle console.
let next = match Pin::new(&mut this.inner).poll_next(cx) {
Poll::Ready(Some(item)) => item,
other => return other,
};
if Instant::now() >= this.next_check {
let rechecked = match &this.scope {
WatchScope::Server => this
.ctx
.require(
&this.caller,
Access::Server,
GrantLevel::Read,
"Events.Watch",
)
.map(|_| WatchScope::Server),
// The caller is still a principal here and not locked out
// — the question it was let in on. What each event may
// carry is decided per event, so a grant revoked
// mid-stream already stops the next delivery.
WatchScope::Mine => WatchScope::decide(&this.ctx, &this.caller, &[]).map(
|decided| match decided {
// A server grant arriving mid-stream does not
// widen an own watch either.
WatchScope::Server => WatchScope::Mine,
mine => mine,
},
),
WatchScope::Accounts { names, .. } => {
WatchScope::decide(&this.ctx, &this.caller, names).map(|decided| {
match decided {
// A grant arriving mid-stream does not widen an
// account watcher: its boundary is what it asked
// for (and a server reader may read every
// repository, so nothing is filtered away).
WatchScope::Server => WatchScope::Accounts {
names: names.clone(),
outside: Vec::new(),
},
accounts => accounts,
}
})
}
};
match rechecked {
Ok(scope) => this.scope = scope,
Err(status) => {
tracing::info!(
principal = %this.caller.principal_id,
"closing an Events.Watch stream: its subscriber may no longer watch"
);
this.refused = true;
return Poll::Ready(Some(Err(status)));
}
}
this.next_check = Instant::now() + this.recheck;
}
match &next {
Ok(event) if !this.scope.admits(&this.ctx, &this.caller, event) => continue,
_ => return Poll::Ready(Some(next)),
}
}
}
}
#[tonic::async_trait]
impl Events for EventsService {
type WatchStream = Pin<Box<dyn Stream<Item = Result<Event, Status>> + Send + 'static>>;
async fn watch(
&self,
request: Request<WatchRequest>,
) -> Result<Response<Self::WatchStream>, Status> {
// **A server reader watches everything** (its `repos`/`namespaces` are
// a FILTER). **Anybody else watches named accounts only**, and for them
// the `namespaces` are the BOUNDARY: each must be an account the caller
// is inside, or one whose account admin made its events public — and
// even then a private repository's events stay inside its account
// ([`WatchScope`]). An account's events are shown to nobody outside it
// by default (owner, 2026-10-03).
let caller = self.ctx.caller(&request)?;
let scope = WatchScope::decide(&self.ctx, &caller, &request.get_ref().namespaces)?;
let request = request.into_inner();
if let WatchScope::Accounts { names, .. } = &scope {
// A repository filter outside the named accounts would widen the
// boundary through the other axis.
if let Some(stray) = request.repos.iter().find(|r| !names.contains(&r.namespace)) {
return Err(not_shown(&stray.namespace));
}
}
// Validate every path in the filter rather than silently watching
// nothing: a UI that mistyped a namespace should get an error, not an
// eternally quiet stream it will read as "no activity".
for repo in &request.repos {
repo.validate()
.map_err(|e| Status::invalid_argument(format!("`repos`: {e}")))?;
}
for kind in &request.kinds {
gunnar_proto::EventKind::try_from(*kind)
.map_err(|_| Status::invalid_argument(format!("unknown event kind {kind}")))?;
}
let filter = EventFilter::from_request(&request);
tracing::debug!(
namespaces = filter.namespaces.len(),
repos = filter.repos.len(),
kinds = filter.kinds.len(),
from_seq = request.from_seq,
"new Events.Watch subscriber"
);
// `subscribe_from`, not `subscribe`: every Watch stream opens with the
// `StreamOpened` prologue, whether or not a cursor was sent. A client
// that starts with no cursor is the one that most needs to be told
// which epoch its first sequence number belongs to — it is going to
// reconnect eventually, and by then the answer would be gone.
// Wrapped, not returned bare: see [`Revalidated`]. `next_check` is
// `now`, so even the first event out of a `from_seq` replay crosses a
// fresh check rather than riding the one the handler just made — the
// replay can be long, and "authorised when you asked" is precisely the
// guarantee that was not enough.
Ok(Response::new(Box::pin(Revalidated {
inner: self.ctx.bus.subscribe_from(filter, request.from_seq),
ctx: self.ctx.clone(),
caller,
scope,
next_check: Instant::now(),
recheck: RECHECK,
refused: false,
})))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::seam::MemoryControlPlane;
use crate::{EventBus, ServerConfig};
use gunnar_proto::{event, RepoDeleted, RepoPath};
use std::sync::Arc;
use tokio_stream::StreamExt as _;
fn deleted(namespace: &str, repo: &str) -> event::Payload {
event::Payload::RepoDeleted(RepoDeleted {
repo: Some(RepoPath::new(namespace, repo)),
principal_id: String::new(),
})
}
/// A control plane holding one enrolled principal with a server-wide
/// `Read` — the standing this RPC needs, and the standing the test takes
/// away.
fn plane_with_a_watcher() -> (Arc<MemoryControlPlane>, String) {
use crate::seam::ControlPlane as _;
let plane = MemoryControlPlane::new();
// The principal has to be ENROLLED: `may` returns false for an id it
// does not know, so granting without adding would make both tests below
// pass for the wrong reason.
let principal = plane
.add_principal("watcher", "watcher", gunnar_proto::PrincipalKind::User)
.expect("enrol the watcher");
let plane = Arc::new(plane);
plane.grant_server(&principal.id, GrantLevel::Read);
(plane, principal.id)
}
fn caller(principal_id: &str) -> Caller {
Caller {
principal_id: principal_id.to_owned(),
fingerprint: "SHA256:watcher".to_owned(),
machine: false,
authenticated: true,
}
}
/// A stream that re-checks on every delivery, so a test can watch a
/// revocation take effect without sleeping out [`RECHECK`].
fn watching(ctx: &ServiceContext, principal_id: &str) -> Revalidated {
Revalidated {
inner: ctx.bus.subscribe_from(EventFilter::default(), 0),
ctx: ctx.clone(),
caller: caller(principal_id),
scope: WatchScope::Server,
next_check: Instant::now(),
recheck: Duration::ZERO,
refused: false,
}
}
/// Pull items until one is an event carrying a payload, and return whether
/// it arrived or the stream refused.
///
/// The prologue (`StreamOpened`) is an event too, and skipping it by name
/// rather than by counting keeps the assertions about the thing under test.
async fn next_delivery(stream: &mut Revalidated) -> Result<Event, Status> {
loop {
match stream.next().await {
Some(Ok(event)) if event.kind == gunnar_proto::EventKind::StreamOpened as i32 => {
continue
}
Some(item) => return item,
None => panic!("the stream ended without delivering or refusing"),
}
}
}
/// **A watcher whose authorisation is withdrawn stops being handed
/// events.**
///
/// `Events.Watch` is the one RPC in the control plane whose answer outlives
/// its request: `caller`/`require` ran when the pipe opened and the pipe
/// then carried every event on the server for as long as the client kept
/// reading. And it is the BROADEST read there is — the stream is
/// server-scoped on purpose, so it carries pushes, deletions and repository
/// names across every namespace, including ones the ex-subscriber never had
/// access to.
///
/// Both halves, because a stream that refused everything would satisfy the
/// second on its own: an event published while the grant holds is
/// DELIVERED, and one published after it is withdrawn is refused with a
/// status rather than silently dropped — a pipe that has quietly stopped
/// carrying reads as "no activity", which is the same failure the handler's
/// filter validation refuses for a mistyped namespace.
#[tokio::test]
async fn a_watcher_who_may_no_longer_watch_stops_being_handed_events() {
let (plane, watcher) = plane_with_a_watcher();
let ctx = ServiceContext::new(
plane.clone(),
Arc::new(EventBus::new(64)),
ServerConfig::default(),
);
let mut stream = watching(&ctx, &watcher);
ctx.bus.publish(deleted("alpha", "one"));
let event = next_delivery(&mut stream)
.await
.expect("a watcher who may watch is handed the event");
assert_eq!(
event.kind,
gunnar_proto::EventKind::RepoDeleted as i32,
"the fixture delivered something other than the published event"
);
// The grant goes. Nothing else changes — the stream is open, the bus is
// the same bus, and the subscriber is still reading.
plane.force_may(Some(false));
ctx.bus.publish(deleted("alpha", "two"));
let status = next_delivery(&mut stream)
.await
.expect_err("the stream kept delivering to a subscriber who may no longer watch");
assert_eq!(
status.code(),
tonic::Code::PermissionDenied,
"the stream ended, but not by refusing: {status:?}"
);
// Terminal: a refused stream must not be pollable for more. Without
// this the refusal could be one skipped event rather than a closed
// pipe, and the next event would flow.
assert!(
stream.next().await.is_none(),
"a refused Events.Watch stream handed out something after its refusal"
);
}
/// The identity-value guard for the test above.
///
/// A `Revalidated` that refused on every delivery would satisfy the refusal
/// half perfectly. This drives four events through a subscriber whose grant
/// never changes and requires all four.
#[tokio::test]
async fn a_watcher_who_still_may_keeps_being_handed_events() {
let (plane, watcher) = plane_with_a_watcher();
let ctx = ServiceContext::new(
plane.clone(),
Arc::new(EventBus::new(64)),
ServerConfig::default(),
);
let mut stream = watching(&ctx, &watcher);
for n in 0..4 {
ctx.bus.publish(deleted("alpha", &format!("repo-{n}")));
next_delivery(&mut stream)
.await
.unwrap_or_else(|status| panic!("delivery {n} was refused: {status:?}"));
}
}
}