crates/monetize-server/tests/events.rs
//! **`Events.Watch`, and the three questions that decide whether it can be left
//! running.**
//!
//! The happy path — "a row was appended and the watcher got it" — is the easy
//! half, and it is asserted on the CONTENTS of the delivered event, not on
//! something having arrived: a stream that emitted an empty `Event` for every
//! row would satisfy "something arrived" and tell a console nothing.
//!
//! The half that matters is the other two:
//!
//! * a subscriber that stops reading must not be able to stall the ledger. The
//! writer here is the thing that moves money; it must finish whether or not a
//! browser tab is awake.
//! * the loss must be VISIBLE. A subscriber that silently misses events is
//! strictly worse than one that polls, because it looks live while being
//! wrong. The gap is proven by genuinely overflowing the ring, never by
//! constructing a gap message and asserting it exists.
use std::sync::Arc;
use std::time::Duration;
use monetize::testing::MockCloud;
use monetize::Step;
use monetize_product::State;
use monetize_proto::events_client::EventsClient;
use monetize_proto::{self as pb, event, EventKind};
use monetize_server::bus::{EventBus, EventFilter};
use monetize_server::registry::Registry;
use monetize_server::state::AppState;
use monetize_server::testing::{CallbackPay, TestServer};
use tokio_stream::StreamExt;
use tonic::metadata::MetadataValue;
use tonic::transport::Channel;
use tonic::Request;
const BEARER: &str = "test-bearer-token-of-sufficient-length";
/// Long enough that a missing event is a real absence, short enough that the
/// suite stays fast.
const QUIET: Duration = Duration::from_millis(300);
/// An `AppState` with a DELIBERATELY TINY ring, so the overflow path is reached
/// by publishing a dozen events instead of six hundred.
fn state_with_bus(capacity: usize) -> Arc<AppState> {
// No product is registered: every assertion here is about the LEDGER and the
// bus, and a product would only add a socket to the picture.
let mut state = AppState::in_memory(Registry::new().sells("gunnar", CallbackPay::new("s")).cloud(MockCloud::new()));
state.bus = Arc::new(EventBus::new(capacity));
// Re-point the ledger at the NEW bus. Without this the observer would still
// hold the default one and every assertion below would be about a bus
// nothing publishes to.
state.wire_events();
Arc::new(state)
}
fn declined(reference: &str) -> Step {
Step::Declined { reference: reference.into(), reason: "the vendor said no".into() }
}
/// The next event, or a NAMED failure. Every read goes through this: an event
/// that never arrives must fail the test by saying so, not by hanging until the
/// harness's sixty-second notice — a hang is indistinguishable from a slow box.
async fn next_within(stream: &mut monetize_server::bus::EventStream, what: &str) -> pb::Event {
tokio::time::timeout(Duration::from_secs(5), stream.next())
.await
.unwrap_or_else(|_| panic!("no event arrived within 5s, waiting for {what}"))
.expect("the stream must not end")
.expect("no status")
}
fn ledger_of(event: &pb::Event) -> &pb::LedgerAppended {
match event.payload.as_ref() {
Some(event::Payload::LedgerAppended(e)) => e,
other => panic!("expected a LedgerAppended, got {other:?}"),
}
}
/// The same bound, for the gRPC side of the stream.
async fn frame(stream: &mut tonic::Streaming<pb::Event>, what: &str) -> pb::Event {
tokio::time::timeout(Duration::from_secs(5), stream.next())
.await
.unwrap_or_else(|_| panic!("no frame arrived within 5s, waiting for {what}"))
.expect("the stream must not end")
.expect("no status")
}
fn gap_of(event: &pb::Event) -> &pb::EventGap {
match event.payload.as_ref() {
Some(event::Payload::Gap(gap)) => gap,
other => panic!("expected a gap marker, got {other:?}"),
}
}
// ---------------------------------------------------------------------------
// (a) DELIVERED, with its contents.
// ---------------------------------------------------------------------------
/// (r) A row appended while a subscriber is attached reaches it, and every field
/// the console renders is the row's own.
///
/// Asserting the reference, the step name, the seq, the split-out product and
/// tenant and the open flag — not `is_some()`. A bus that published a default
/// `LedgerAppended` for every append would pass an "something arrived" test and
/// would leave the console showing blank rows forever.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_ledger_row_appended_under_a_watcher_is_delivered_with_its_contents() {
let state = state_with_bus(16);
let mut stream = state.bus.subscribe(EventFilter::default());
state.ledger.append(1_700, declined("gunnar/team/sub/2026-09-04")).unwrap();
state.ledger.append(1_800, declined("holger/acme/2026-09-04")).unwrap();
let first = next_within(&mut stream, "the first appended row").await;
assert_eq!(first.seq, 1, "the first event of a run is seq 1");
assert_eq!(first.kind, EventKind::LedgerAppended as i32);
assert!(first.kind_matches_payload(), "kind and payload must not disagree");
let row = ledger_of(&first);
assert_eq!(
(row.reference.as_str(), row.product.as_str(), row.tenant.as_str(), row.step.as_str(), row.seq, row.at_unix_ms, row.open),
("gunnar/team/sub/2026-09-04", "gunnar", "team/sub", "Declined", 0, 1_700, false),
"every field is the row's own"
);
let second = next_within(&mut stream, "the second appended row").await;
assert_eq!(second.seq, 2);
let row = ledger_of(&second);
assert_eq!((row.reference.as_str(), row.product.as_str(), row.tenant.as_str()), ("holger/acme/2026-09-04", "holger", "acme"));
assert_eq!(state.bus.stats().published, 2);
assert_eq!(state.bus.stats().dropped, 0, "nobody lagged here");
}
/// (r) An entitlement change reaches a watcher with the states on both sides of
/// the move, and `pushed` says whether the product actually took it.
///
/// The `from`/`to` pair is the point: a feed that only carried the new state
/// could not tell a console that a tenant just LOST its plan, which is the event
/// an operator most needs.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_entitlement_change_carries_both_sides_of_the_move() {
let state = state_with_bus(16);
let mut stream = state.bus.subscribe(EventFilter::default());
state.report_change(monetize_server::state::StateChange {
product: "gunnar".into(),
tenant: "team/sub".into(),
from: Some(State::Paid),
to: State::Grace,
source: "ladder".into(),
at_unix_ms: 5,
pushed: false,
});
let event = next_within(&mut stream, "the entitlement change").await;
assert_eq!(event.kind, EventKind::EntitlementChanged as i32);
let Some(event::Payload::EntitlementChanged(e)) = event.payload.as_ref() else { panic!("wrong payload") };
assert_eq!(
(e.product.as_str(), e.tenant.as_str(), e.from_state(), e.had_previous, e.to_state(), e.source.as_str(), e.pushed),
("gunnar", "team/sub", pb::State::Paid, true, pb::State::Grace, "ladder", false),
"both sides of the move, the source, and whether the product took it"
);
}
/// (r) The same event, over a real gRPC stream, through the bearer.
///
/// The in-process assertions above prove the bus; this proves the SERVICE — that
/// `Events.Watch` is registered on the router, survives the interceptor, and
/// puts the prologue on the wire before anything else. A bus that worked with no
/// service wired to it would pass every other test in this file.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn events_watch_streams_the_prologue_then_real_rows_over_grpc() {
let state = state_with_bus(16);
let server = TestServer::spawn(state.clone(), BEARER).await;
let channel = Channel::from_shared(server.endpoint()).unwrap().connect().await.unwrap();
let mut req = Request::new(pb::WatchRequest { products: vec!["gunnar".into()], ..Default::default() });
req.metadata_mut().insert("authorization", MetadataValue::try_from(format!("Bearer {BEARER}")).unwrap());
let mut stream = EventsClient::new(channel).watch(req).await.expect("watch opens").into_inner();
// The prologue is the FIRST thing on every stream, ahead of any event.
let opened = frame(&mut stream, "the prologue").await;
assert_eq!(opened.kind, EventKind::StreamOpened as i32, "the epoch must arrive before any sequence number");
let Some(event::Payload::StreamOpened(p)) = opened.payload.as_ref() else { panic!("wrong payload") };
assert_eq!(p.epoch, state.bus.epoch(), "the stream reports this run's epoch");
assert!(!p.epoch.is_empty());
assert_eq!((p.last_seq, p.replaying, p.from_seq), (0, 0, 0), "a fresh bus, no cursor, nothing replayed");
// Only now, so the row cannot have been buffered before the subscription.
state.ledger.append(2_500, declined("gunnar/team/sub/2026-09-04")).unwrap();
// Filtered out by `products: [gunnar]` — and its absence below is what makes
// the filter assertion mean something.
state.ledger.append(2_600, declined("holger/acme/2026-09-04")).unwrap();
state.ledger.append(2_700, declined("gunnar/team/other/2026-09-04")).unwrap();
let first = frame(&mut stream, "the first gunnar row").await;
let row = ledger_of(&first);
assert_eq!((row.reference.as_str(), row.step.as_str(), row.at_unix_ms), ("gunnar/team/sub/2026-09-04", "Declined", 2_500));
let second = frame(&mut stream, "the second gunnar row").await;
let row = ledger_of(&second);
assert_eq!(
(row.reference.as_str(), row.tenant.as_str()),
("gunnar/team/other/2026-09-04", "team/other"),
"the holger row was filtered out, so the NEXT frame is the second gunnar row — not the holger one"
);
// The stream FIRST, then the server. `serve_with_incoming_shutdown` waits
// for its in-flight connections, and a `Watch` connection is in flight until
// the client lets go of it — a shutdown ordered the other way round hangs
// forever, which is a property of a server-streaming RPC and not a bug.
drop(stream);
server.stop().await;
}
// ---------------------------------------------------------------------------
// (b) THE WRITER IS NOT HOSTAGE TO A READER.
// ---------------------------------------------------------------------------
/// (r) **A subscriber that stops reading does not stall the ledger.**
///
/// A ring of four, a subscriber that is attached and never polled, and fifty
/// rows written through `Ledger::append` — the same call every payment makes.
/// The assertion is on the LEDGER: all fifty rows are committed and readable.
/// A bus that waited for its slowest reader would hang here and the timeout
/// would fire; that is the failure this test exists to catch, and it is the
/// difference between a live feed and an outage.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_stalled_subscriber_does_not_stop_the_ledger_being_written() {
const CAPACITY: usize = 4;
const ROWS: usize = 50;
let state = state_with_bus(CAPACITY);
// Attached, and deliberately never polled: a browser that opened the feed
// and then went to sleep.
let mut asleep = state.bus.subscribe(EventFilter::default());
assert_eq!(state.bus.stats().subscribers, 1, "the subscriber really is attached");
let writer = state.clone();
let written = tokio::time::timeout(
Duration::from_secs(5),
tokio::task::spawn_blocking(move || {
for i in 0..ROWS {
writer.ledger.append(3_000 + i as u64, declined(&format!("gunnar/team/{i}/2026-09-04"))).unwrap();
}
writer.ledger.recent(0).unwrap().len()
}),
)
.await
.expect("the ledger must not wait on a subscriber that stopped reading")
.expect("the writer must not panic");
assert_eq!(written, ROWS, "every row is committed and readable, with a subscriber {CAPACITY} events behind the whole time");
assert_eq!(state.ledger.rows("gunnar/team/49/2026-09-04").unwrap().len(), 1, "including the last one");
// And the sleeper is told, rather than being quietly short: the ring held
// four, so forty-six were dropped and it hears about them.
let first = next_within(&mut asleep, "the gap a sleeping subscriber is owed").await;
assert_eq!(first.kind, EventKind::Gap as i32, "a lagging subscriber's first delivery is the gap, not a silent jump");
assert_eq!(gap_of(&first).dropped as usize, ROWS - CAPACITY, "the gap names exactly what the ring could not hold");
}
// ---------------------------------------------------------------------------
// (c) THE LOSS IS VISIBLE.
// ---------------------------------------------------------------------------
/// (r) **The gap fires because the ring genuinely overflowed**, and the
/// arithmetic closes: every published event is either delivered or counted as
/// dropped, with nothing invented.
///
/// Proven by publishing twelve events into a ring of four and reading nothing in
/// between — never by calling the gap constructor. A test that built an
/// `EventGap` by hand would pass against a bus that never emitted one.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn overflowing_the_ring_produces_a_gap_that_accounts_for_every_event() {
const CAPACITY: usize = 4;
const PUBLISHED: usize = 12;
let state = state_with_bus(CAPACITY);
let mut stream = state.bus.subscribe(EventFilter::default());
for i in 0..PUBLISHED {
state.ledger.append(4_000 + i as u64, declined(&format!("gunnar/team/{i}/2026-09-04"))).unwrap();
}
let first = next_within(&mut stream, "the gap marker after an overflow").await;
assert_eq!(first.kind, EventKind::Gap as i32, "the FIRST delivery to a lagging subscriber must be the gap marker");
let gap = gap_of(&first).clone();
assert!(gap.dropped > 0, "a gap that dropped nothing is not a gap");
assert_eq!(gap.epoch, state.bus.epoch(), "a gap is self-describing: the numbers either side are only comparable within one run");
assert_eq!(gap.last_delivered_seq, 0, "nothing had been delivered before the hole");
assert_eq!(gap.resume_seq, gap.dropped + 1, "the gap names the first event delivered after it");
let mut delivered = Vec::new();
while let Ok(Some(item)) = tokio::time::timeout(QUIET, stream.next()).await {
delivered.push(item.expect("no status").seq);
}
assert_eq!(
gap.dropped as usize + delivered.len(),
PUBLISHED,
"every published event is either delivered or counted as dropped; got dropped={} delivered={delivered:?}",
gap.dropped
);
assert_eq!(delivered.len(), CAPACITY, "a ring of {CAPACITY} retains exactly {CAPACITY}. That bound is what replaces unbounded memory");
assert_eq!(delivered, vec![9, 10, 11, 12], "and it retains the NEWEST, in order");
assert_eq!(state.bus.stats().dropped, gap.dropped, "the loss is counted where an operator can see it, not only on the wire");
}
/// (r) A subscriber that reconnects with a cursor gets what the ring still
/// holds, and is TOLD when the cursor is older than that.
///
/// Both directions in one test, because the honest half (a resume that works) is
/// worthless without the honest other half (a resume that cannot work saying so
/// instead of quietly starting at the tail).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_resume_replays_what_it_can_and_names_the_hole_when_it_cannot() {
let state = state_with_bus(4);
for i in 0..6 {
state.ledger.append(5_000 + i as u64, declined(&format!("gunnar/team/{i}/2026-09-04"))).unwrap();
}
// The ring holds seq 3..=6.
assert_eq!(state.bus.oldest_buffered_seq(), 3);
// A cursor INSIDE the buffer: no gap, and the replay starts one past it.
let mut inside = state.bus.subscribe_from(EventFilter::default(), 4);
let opened = next_within(&mut inside, "the prologue of a resumed stream").await;
let Some(event::Payload::StreamOpened(p)) = opened.payload.as_ref() else { panic!("wrong payload") };
assert_eq!((p.last_seq, p.oldest_buffered_seq, p.replaying, p.from_seq), (6, 3, 2, 4));
let mut seqs = Vec::new();
while let Ok(Some(item)) = tokio::time::timeout(QUIET, inside.next()).await {
let e = item.unwrap();
assert_ne!(e.kind, EventKind::Gap as i32, "a cursor inside the buffer is not a gap");
seqs.push(e.seq);
}
assert_eq!(seqs, vec![5, 6], "everything past the cursor, in order, and nothing before it");
// A cursor the ring has already overwritten: the gap comes FIRST, before any
// replayed event, and names how many fell out.
let mut outside = state.bus.subscribe_from(EventFilter::default(), 1);
assert_eq!(next_within(&mut outside, "the prologue").await.kind, EventKind::StreamOpened as i32, "the epoch still leads");
let gap = next_within(&mut outside, "the resume gap").await;
assert_eq!(gap.kind, EventKind::Gap as i32);
assert_eq!(
(gap_of(&gap).dropped, gap_of(&gap).last_delivered_seq, gap_of(&gap).resume_seq),
(1, 1, 3),
"seq 2 fell out of the ring; the cursor was 1 and the replay picks up at 3"
);
}