crates/monetize-server/src/bus.rs
//! **The event bus behind `Events.Watch`.**
//!
//! One `tokio::sync::broadcast` ring, one monotonic sequence counter, a per-run
//! epoch, and a per-subscriber [`EventStream`] that turns the channel's lag
//! signal into an explicit [`pb::EventGap`] on the wire.
//!
//! # Provenance
//!
//! This is gunnar's §8.1 bus (`gunnar-grpc/src/bus.rs`), adopted deliberately
//! rather than re-derived. It is a separate repository, so this is a port and
//! not a shared crate — the one place in this seam where one implementation
//! could not be
//! obeyed by construction. What was copied, and what monetize does differently,
//! is written down here so the divergence is a decision and not a drift:
//!
//! | question | gunnar's answer | monetize |
//! |---|---|---|
//! | slow subscriber | bounded ring, drop the OLDEST, tell them | same |
//! | how it is told | synthesised `EventGap`, never filtered out | same |
//! | replay | best effort within the same ring, `from_seq` | same |
//! | restart | per-run `epoch` in a `StreamOpened` prologue | same |
//! | publish failing | infallible; a push must not fail for lack of watchers | same, and the stake is higher: a PAYMENT must not |
//! | filter axes | namespace / repo / kind | product / tenant / kind |
//! | payload shaping | strips attestation object ids unless asked | none — no monetize payload is large enough to be worth an opt-in |
//! | timestamps | `int64 at_unix_ms` | `uint64`, which is what every other monetize message uses |
//!
//! # Why a lagging subscriber is told, not buffered
//!
//! A console tab that is asleep 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. The alternative — an unbounded
//! queue per subscriber — turns one asleep laptop into an OOM on a box that is
//! also holding a bankgiro ledger. It is not available here by construction, not
//! by discipline.
//!
//! # What a subscriber does with a gap
//!
//! [`pb::EventGap`] carries `dropped`, the last sequence known delivered, and
//! `resume_seq`. A console that sees one re-reads state over `Entitlement.List`
//! / `Ledger.Query` rather than trusting its incremental view. That is why the
//! gap names both edges of the hole — and why `monetize-ui-server` falls back to
//! polling rather than sitting quietly on a stream it can no longer trust.
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 monetize_proto::{self as pb, event, EventKind};
use tokio::sync::broadcast;
use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::Stream;
/// The ring size a server uses when the config says nothing. 512 events is a
/// couple of minutes of a busy monetize and a few days of a quiet one, at a few
/// hundred bytes each — tens of kilobytes, which is the point: it is a number
/// chosen so the bus never shows up in a memory graph.
pub const DEFAULT_EVENT_BUFFER: usize = 512;
/// Milliseconds since the Unix epoch. Saturating, because an appliance whose
/// clock has not been set must still be able to emit events.
pub fn now_unix_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis().min(u64::MAX as u128) as u64)
.unwrap_or(0)
}
/// A snapshot of the bus, for logs and tests.
#[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,
}
/// A value that is different in every run of the process.
///
/// No new dependency for it: the wall clock in nanoseconds, the process id, and
/// a counter so two buses built in the same process are still distinguishable.
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}"))
}
/// The publish side of `Events.Watch`.
///
/// Publishing is infallible and non-blocking from the caller's point of view.
/// `Ledger::append` calls it with a redb transaction already committed; a
/// payment must never fail, or wait, because nobody was reading.
#[derive(Debug)]
pub struct EventBus {
tx: broadcast::Sender<Arc<pb::Event>>,
seq: AtomicU64,
published: AtomicU64,
dropped: Arc<AtomicU64>,
capacity: usize,
/// Identifies this run. Minted once in [`EventBus::new`] and never mutated:
/// a restart is a new process, a new bus, a 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. There is no durable log here.
recent: Mutex<VecDeque<Arc<pb::Event>>>,
}
impl EventBus {
pub fn new(capacity: usize) -> Self {
let capacity = capacity.max(1);
let (tx, _rx) = broadcast::channel(capacity);
EventBus {
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, never blocks. With no subscribers the event is dropped on
/// the floor: this is a notification channel, and the durable record of a
/// purchase is the ledger, not this.
pub fn publish(&self, payload: event::Payload) -> Arc<pb::Event> {
// Sequence assignment, retention and broadcast under one lock. Not for
// the counter's sake — it is atomic — but so a subscription opening
// concurrently sees a consistent split: everything in `recent` was
// already sent, everything sent after it reaches the new receiver.
// Without that a resuming subscriber could be handed an 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(pb::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 at the live tail, with no prologue. What an in-process
/// subscriber wants: it started with the server and has no cursor.
pub fn subscribe(&self, filter: EventFilter) -> EventStream {
self.open(filter, None)
}
/// Subscribe the way `Events.Watch` does: a [`pb::StreamOpened`] prologue
/// first, then whatever `from_seq` can still be replayed, then the live
/// tail. `from_seq == 0` means "no cursor, start live" — and still earns the
/// prologue, because a client with no cursor yet is exactly the one that
/// most needs to be told which epoch it is about to start counting in.
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).
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<pb::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(pb::Event::new(
0,
now_unix_ms(),
event::Payload::StreamOpened(pb::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);
eprintln!(
"monetize-server: events resume gap dropped={dropped} from_seq={from_seq} oldest_buffered_seq={oldest} — the bus is a \
bounded ring with no log, so the subscriber is told what it lost rather than sold a replay it cannot have"
);
prologue.push_back(pb::Event::new(
0,
now_unix_ms(),
event::Payload::Gap(pb::EventGap {
dropped,
last_delivered_seq: from_seq,
resume_seq: replay.first().map(|e| e.seq).unwrap_or(0),
epoch: self.epoch.to_string(),
}),
));
}
// Filtered exactly as a live event is, so a replayed event and the
// same event delivered live are the same bytes (one implementation: one
// `EventFilter`).
for event in replay {
if filter.matches(&event) {
prologue.push_back((*event).clone());
}
}
}
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)
}
/// The sequence most recently assigned. 0 before anything is published.
/// **What the ring still holds**, oldest first — a read, for tests and
/// for a caller that wants the events it just caused without a stream.
pub fn recent(&self) -> Vec<Arc<pb::Event>> {
self.recent
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.cloned()
.collect()
}
pub fn last_seq(&self) -> u64 {
self.seq.load(Ordering::SeqCst)
}
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,
}
}
}
impl Default for EventBus {
fn default() -> Self {
EventBus::new(DEFAULT_EVENT_BUFFER)
}
}
/// Which events a subscriber wants. 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 products: Vec<String>,
pub tenants: Vec<String>,
pub kinds: Vec<EventKind>,
}
impl EventFilter {
pub fn from_request(req: &pb::WatchRequest) -> Self {
EventFilter {
products: req.products.clone(),
tenants: req.tenants.clone(),
kinds: req
.kinds
.iter()
.filter_map(|k| EventKind::try_from(*k).ok())
.filter(|k| *k != EventKind::Unspecified)
.collect(),
}
}
/// True when this subscriber should see `event`.
///
/// A gap and the prologue always match: both are facts about the STREAM
/// rather than about a product, 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: &pb::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;
}
// **A refused callback belongs to no product and no tenant**, and that
// is a fact about it rather than a field somebody forgot to fill in: it
// never got past `parse_callback`, so nothing in it names either, and
// this server will not take a stranger's word for one. It must
// therefore skip those two axes — a console watching one product would
// otherwise filter away the only notice it gets that the money for that
// product is at the vendor. The KIND axis still applies: a subscriber
// that asked for ledger rows alone asked for ledger rows alone.
if event.kind == EventKind::CallbackRefused as i32 {
return true;
}
// **A payment found at the vendor, likewise.** Its metadata may name a
// product, but that is the vendor repeating a label it was handed, not
// a fact this server established — and it is exactly the event a
// console scoped to that product must not be able to lose.
if event.kind == EventKind::PaidAtVendor as i32 {
return true;
}
if !self.products.is_empty() {
match event.product() {
Some(p) if self.products.iter().any(|w| w == p) => {}
_ => return false,
}
}
if !self.tenants.is_empty() {
match event.tenant() {
Some(t) if self.tenants.iter().any(|w| w == t) => {}
_ => return false,
}
}
true
}
}
/// One subscriber's view of the bus.
///
/// Pull-driven: nothing is copied out of the ring until the consumer polls,
/// which is what keeps the memory bound in the ring where it belongs and makes
/// the lag path deterministic enough to test.
pub struct EventStream {
inner: BroadcastStream<Arc<pb::Event>>,
filter: EventFilter,
/// Everything that must go out before the live tail, in order: the
/// `StreamOpened` prologue, a resume gap if the cursor fell out of the ring,
/// then the replayed events. Empty for a live-only subscription.
prologue: VecDeque<pb::Event>,
/// The event that follows a gap, held back one poll so the gap goes first
/// and can name the sequence the stream resumes at.
queued: Option<pb::Event>,
last_seq: u64,
dropped_total: Arc<AtomicU64>,
epoch: Arc<str>,
}
impl EventStream {
fn gap(&mut self, dropped: u64, resume_seq: u64) -> pb::Event {
let last_delivered_seq = self.last_seq;
pb::Event::new(
0,
now_unix_ms(),
event::Payload::Gap(pb::EventGap {
dropped,
last_delivered_seq,
resume_seq,
epoch: self.epoch.to_string(),
}),
)
}
}
impl Stream for EventStream {
type Item = Result<pb::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((*event).clone())));
}
// 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);
eprintln!(
"monetize-server: events subscriber lagged dropped={dropped} last_delivered_seq={} — the oldest events are gone and \
a gap marker is going out; memory is bounded by the bus ring, never by this subscriber",
this.last_seq
);
// A lag means the ring overwrote unread slots, so retained
// events are available RIGHT NOW. Take the first 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((*event).clone());
}
}
return Poll::Ready(Some(Ok(gap)));
}
Poll::Ready(None) => return Poll::Ready(None),
Poll::Pending => return Poll::Pending,
}
}
}
}
/// Build the `LedgerAppended` payload for one committed row.
///
/// The one place a `monetize::ledger::Row` becomes an event (one implementation): the ledger
/// observer installed by [`crate::state::AppState`] and any test that wants the
/// same shape both call this, so the split of a reference into product and
/// tenant exists once.
pub fn ledger_event(row: &monetize::ledger::Row, open: bool) -> event::Payload {
let reference = row.step.reference().to_string();
// `<product>/<tenant>/<period>` — the tenant is everything between the first
// and last slash, because a tenant id has slashes of its own (`team/sub`).
let mut head = reference.splitn(2, '/');
let product = head.next().unwrap_or_default().to_string();
let tenant = head
.next()
.and_then(|rest| rest.rsplit_once('/'))
.map(|(t, _)| t.to_string())
.unwrap_or_default();
event::Payload::LedgerAppended(pb::LedgerAppended {
reference,
product,
tenant,
seq: row.seq,
at_unix_ms: row.at_unix_ms,
step: row.step.name().to_string(),
open,
})
}
#[cfg(test)]
mod tests {
use super::*;
fn changed(product: &str, tenant: &str) -> event::Payload {
event::Payload::EntitlementChanged(pb::EntitlementChanged {
product: product.into(),
tenant: tenant.into(),
order: "gunnar/team/sub/2026-09-03+pack_bytes=10737418240".into(),
from_state: pb::State::Free as i32,
had_previous: true,
to_state: pb::State::Paid as i32,
source: "operator".into(),
pushed: true,
})
}
#[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(changed("gunnar", "a")).seq, 1);
assert_eq!(bus.publish(changed("gunnar", "b")).seq, 2);
assert_eq!(bus.last_seq(), 2);
}
#[test]
fn a_published_event_kind_agrees_with_its_payload() {
let bus = EventBus::new(8);
let event = bus.publish(changed("gunnar", "a"));
assert!(event.kind_matches_payload());
assert_eq!(event.kind, EventKind::EntitlementChanged as i32);
}
/// (r) Measure, never assert: a filter that matched everything would pass a
/// happy-path test.
/// Both directions on every axis.
#[test]
fn filter_matches_are_not_vacuous() {
let bus = EventBus::new(8);
let event = bus.publish(changed("gunnar", "team/sub"));
assert!(
EventFilter::default().matches(&event),
"an empty filter watches everything"
);
let hit = |f: EventFilter| f.matches(&event);
assert!(hit(EventFilter {
products: vec!["gunnar".into()],
..Default::default()
}));
assert!(
!hit(EventFilter {
products: vec!["holger".into()],
..Default::default()
}),
"the product axis must exclude"
);
assert!(hit(EventFilter {
tenants: vec!["team/sub".into()],
..Default::default()
}));
assert!(
!hit(EventFilter {
tenants: vec!["team".into()],
..Default::default()
}),
"a PREFIX of the tenant is not the tenant"
);
assert!(hit(EventFilter {
kinds: vec![EventKind::EntitlementChanged],
..Default::default()
}));
assert!(
!hit(EventFilter {
kinds: vec![EventKind::LedgerAppended],
..Default::default()
}),
"the kind axis must exclude"
);
// ANDed across axes: the right product with the wrong tenant is a miss.
assert!(!hit(EventFilter {
products: vec!["gunnar".into()],
tenants: vec!["other".into()],
..Default::default()
}));
}
#[test]
fn a_gap_and_the_prologue_survive_every_filter() {
let narrow = EventFilter {
products: vec!["nothing-matches-this".into()],
tenants: vec!["nor-this".into()],
kinds: vec![EventKind::UsageMeasured],
};
let gap = pb::Event::new(
0,
0,
event::Payload::Gap(pb::EventGap {
dropped: 3,
last_delivered_seq: 1,
resume_seq: 5,
epoch: "an-epoch".into(),
}),
);
assert!(
narrow.matches(&gap),
"a subscriber must never filter away its own gap marker"
);
let opened = pb::Event::new(
0,
0,
event::Payload::StreamOpened(pb::StreamOpened {
epoch: "an-epoch".into(),
..Default::default()
}),
);
assert!(
narrow.matches(&opened),
"…nor the prologue that says which run its sequence numbers belong to"
);
// The other direction, so neither assertion above is satisfied by a
// `matches` that says yes to everything.
assert!(!narrow.matches(&pb::Event::new(1, 0, changed("gunnar", "team/sub"))));
}
/// (r) An epoch pinned to a constant would satisfy every "the field is
/// populated" check and report no restart, ever.
#[test]
fn an_epoch_is_never_the_same_twice() {
let a = new_epoch();
let b = new_epoch();
assert_ne!(a, b, "two epochs minted in a row must differ");
assert!(!a.is_empty());
assert_ne!(
EventBus::new(4).epoch(),
EventBus::new(4).epoch(),
"two buses are two runs"
);
}
/// (r) The reference split. A tenant id contains a slash of its own
/// (`team/sub`), so a naive `split('/')` would call the tenant "team" and a
/// tenant filter would then match nothing a console ever asks for.
#[test]
fn a_ledger_event_splits_a_reference_into_product_and_the_whole_tenant() {
use monetize::ledger::Row;
let row = Row {
seq: 3,
at_unix_ms: 77,
step: monetize::Step::Released {
reference: "gunnar/team/sub/2026-09-04".into(),
reason: "expired".into(),
},
};
let event::Payload::LedgerAppended(e) = ledger_event(&row, false) else {
panic!("wrong payload")
};
assert_eq!(
(e.product.as_str(), e.tenant.as_str()),
("gunnar", "team/sub"),
"the tenant is everything between the first and last slash"
);
assert_eq!(
(e.seq, e.at_unix_ms, e.step.as_str(), e.open),
(3, 77, "Released", false)
);
// A single-segment tenant still works, and is a different answer.
let row = Row {
seq: 0,
at_unix_ms: 1,
step: monetize::Step::Declined {
reference: "holger/acme/2026-01-01".into(),
reason: "no".into(),
},
};
let event::Payload::LedgerAppended(e) = ledger_event(&row, false) else {
panic!("wrong payload")
};
assert_eq!(
(e.product.as_str(), e.tenant.as_str(), e.step.as_str()),
("holger", "acme", "Declined")
);
}
}