crates/gunnar-grpc/tests/event_bus.rs
//! The event bus, including the path that actually matters: what happens
//! to a subscriber that cannot keep up.
//!
//! The happy path, "two subscribers both got the event", is the easy half
//! and is tested first. The half that decides whether this server can be left
//! running is the lag path, and it is exercised by genuinely overflowing the
//! ring, not by asserting that a gap message can be constructed.
use std::time::Duration;
use gunnar_grpc::{EventBus, EventFilter};
use gunnar_proto::{event, EventGap, EventKind, RepoDeleted, RepoPath};
use tokio_stream::StreamExt;
/// Long enough that a missing event is a real absence, short enough that the
/// suite stays fast.
const QUIET: Duration = Duration::from_millis(200);
fn deleted(namespace: &str, repo: &str) -> event::Payload {
event::Payload::RepoDeleted(RepoDeleted {
repo: Some(RepoPath::new(namespace, repo)),
principal_id: String::new(),
})
}
fn gap_of(event: &gunnar_proto::Event) -> &EventGap {
match event.payload.as_ref() {
Some(event::Payload::Gap(gap)) => gap,
other => panic!("expected a gap marker, got {other:?}"),
}
}
#[tokio::test]
async fn every_subscriber_sees_every_event_in_order() {
let bus = EventBus::new(16);
let mut one = bus.subscribe(EventFilter::default());
let mut two = bus.subscribe(EventFilter::default());
assert_eq!(bus.stats().subscribers, 2);
for repo in ["a", "b", "c"] {
bus.publish(deleted("nordisk", repo));
}
for stream in [&mut one, &mut two] {
let mut seen = Vec::new();
for _ in 0..3 {
let event = stream
.next()
.await
.expect("stream open")
.expect("no status");
assert_eq!(event.kind, EventKind::RepoDeleted as i32);
seen.push((event.seq, event.repo_path().unwrap().repo.clone()));
}
assert_eq!(
seen,
vec![
(1, "a".to_string()),
(2, "b".to_string()),
(3, "c".to_string())
],
"a broadcast subscriber must see every event, in publication order"
);
}
assert_eq!(bus.stats().published, 3);
assert_eq!(bus.stats().dropped, 0, "nobody lagged here");
}
#[tokio::test]
async fn a_subscriber_that_falls_behind_gets_a_gap_marker() {
// THE test. A ring of 4, twelve events published, and nothing polled in
// between: the subscriber must be TOLD it lost events, and told how many.
const CAPACITY: usize = 4;
const PUBLISHED: u64 = 12;
let bus = EventBus::new(CAPACITY);
let mut stream = bus.subscribe(EventFilter::default());
for i in 0..PUBLISHED {
bus.publish(deleted("nordisk", &format!("repo-{i}")));
}
let first = stream
.next()
.await
.expect("stream open")
.expect("no status");
assert_eq!(
first.kind,
EventKind::Gap as i32,
"a lagging subscriber's FIRST delivery must be the gap marker, not a silent jump in \
sequence numbers"
);
// Cloned, not copied: `EventGap` carries the stream epoch now, so it is no
// longer a bag of integers.
let gap = gap_of(&first).clone();
assert!(gap.dropped > 0, "a gap that dropped nothing is not a gap");
// Drain what is left and assert the arithmetic closes: dropped + delivered
// accounts for every published event, with nothing invented.
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 as usize,
"every published event must be either delivered or counted as dropped; got \
dropped={} delivered={:?}",
gap.dropped,
delivered
);
assert_eq!(
delivered.len(),
CAPACITY,
"a ring of {CAPACITY} retains exactly {CAPACITY} events. This is the bound that \
replaces unbounded memory"
);
assert_eq!(
gap.resume_seq, delivered[0],
"the gap must name the sequence the stream actually resumes at"
);
assert_eq!(
gap.resume_seq,
gap.dropped + 1,
"sequences are contiguous, so the resume point is exactly past the dropped run"
);
assert_eq!(
gap.last_delivered_seq, 0,
"this subscriber had received nothing before the gap"
);
assert!(
!delivered.contains(&1),
"the oldest events are the ones dropped; seq 1 must be gone"
);
assert_eq!(
bus.stats().dropped,
gap.dropped,
"Admin.Stats must be able to show an operator that the buffer is too small"
);
assert_eq!(bus.stats().published, PUBLISHED);
}
#[tokio::test]
async fn memory_is_bounded_by_the_ring_no_matter_how_far_behind() {
// 10 000 events into a ring of 8 with nobody reading. If the design had
// any per-subscriber queue, this is where it would grow. The observable
// consequence of the bound: at most capacity + 1 deliveries survive.
const CAPACITY: usize = 8;
const PUBLISHED: u64 = 10_000;
let bus = EventBus::new(CAPACITY);
let mut stream = bus.subscribe(EventFilter::default());
for i in 0..PUBLISHED {
bus.publish(deleted("nordisk", &format!("repo-{i}")));
}
let mut items = Vec::new();
while let Ok(Some(item)) = tokio::time::timeout(QUIET, stream.next()).await {
items.push(item.expect("no status"));
}
assert_eq!(
items.len(),
CAPACITY + 1,
"one gap marker plus at most a ringful; anything more means events were queued \
somewhere they should not have been"
);
let gap = gap_of(&items[0]);
assert_eq!(gap.dropped, PUBLISHED - CAPACITY as u64);
assert_eq!(items.last().unwrap().seq, PUBLISHED, "the newest survives");
}
#[tokio::test]
async fn one_slow_subscriber_does_not_cost_a_fast_one_anything() {
// The reason the bus is a broadcast ring and not a fan-out of queues: an
// asleep laptop must not degrade the dwarves panel next to it.
const CAPACITY: usize = 4;
let bus = EventBus::new(CAPACITY);
let mut fast = bus.subscribe(EventFilter::default());
let mut slow = bus.subscribe(EventFilter::default());
for i in 0..CAPACITY as u64 {
bus.publish(deleted("nordisk", &format!("repo-{i}")));
// The fast subscriber keeps up, one event at a time.
let event = fast.next().await.expect("open").expect("no status");
assert_eq!(event.seq, i + 1);
assert_eq!(event.kind, EventKind::RepoDeleted as i32);
}
// Now overflow. Only the slow one should suffer.
for i in 0..CAPACITY as u64 {
bus.publish(deleted("nordisk", &format!("late-{i}")));
}
let next_fast = fast.next().await.expect("open").expect("no status");
assert_eq!(
next_fast.kind,
EventKind::RepoDeleted as i32,
"the fast subscriber never lagged, so it must not receive a gap"
);
assert_eq!(next_fast.seq, CAPACITY as u64 + 1);
let first_slow = slow.next().await.expect("open").expect("no status");
assert_eq!(
first_slow.kind,
EventKind::Gap as i32,
"the slow subscriber is the one that lost events"
);
}
#[tokio::test]
async fn a_filtered_subscriber_sees_only_its_own_namespace() {
let bus = EventBus::new(16);
let mut watching_alpha = bus.subscribe(EventFilter {
namespaces: vec!["alpha".into()],
..Default::default()
});
bus.publish(deleted("beta", "not-mine"));
bus.publish(deleted("alpha", "mine"));
bus.publish(deleted("beta", "also-not-mine"));
let event = watching_alpha
.next()
.await
.expect("open")
.expect("no status");
assert_eq!(event.repo_path().unwrap().namespace, "alpha");
assert_eq!(event.repo_path().unwrap().repo, "mine");
// …and nothing else arrives. Without this half the filter could be a
// no-op and the test above would still pass.
assert!(
tokio::time::timeout(QUIET, watching_alpha.next())
.await
.is_err(),
"events from another namespace must not reach this subscriber"
);
}
#[tokio::test]
async fn a_filtered_subscriber_still_learns_that_it_lagged() {
// A gap that could be filtered away would let a subscriber believe it had
// seen everything. Filter to a namespace that no event uses, then overflow.
let bus = EventBus::new(4);
let mut narrow = bus.subscribe(EventFilter {
namespaces: vec!["never-published-to".into()],
kinds: vec![EventKind::KeyUsed],
..Default::default()
});
for i in 0..12 {
bus.publish(deleted("nordisk", &format!("repo-{i}")));
}
// Bounded: if the gap were ever swallowed this stream would simply never
// produce anything, and an unbounded `.await` would HANG instead of
// failing. A guard that hangs reports nothing; it must go red.
let first = tokio::time::timeout(QUIET, narrow.next())
.await
.expect("a gap must reach even a subscriber whose filter matches nothing")
.expect("open")
.expect("no status");
assert_eq!(
first.kind,
EventKind::Gap as i32,
"a gap must reach even a subscriber whose filter matches nothing"
);
assert!(gap_of(&first).dropped > 0);
}
#[tokio::test]
async fn attestation_object_ids_are_opt_in() {
use gunnar_proto::PushAttestation;
let bus = EventBus::new(8);
let mut lean = bus.subscribe(EventFilter::default());
let mut full = bus.subscribe(EventFilter {
include_attestation_object_ids: true,
..Default::default()
});
bus.publish(event::Payload::PushAttestation(PushAttestation {
repo: Some(RepoPath::new("nordisk", "gunnar")),
new_object_oids: vec!["a".repeat(64), "b".repeat(64)],
new_object_count: 2,
..Default::default()
}));
let lean_event = lean.next().await.expect("open").expect("no status");
let full_event = full.next().await.expect("open").expect("no status");
let unwrap_attestation = |event: gunnar_proto::Event| match event.payload {
Some(event::Payload::PushAttestation(att)) => att,
other => panic!("expected an attestation, got {other:?}"),
};
let lean_att = unwrap_attestation(lean_event);
let full_att = unwrap_attestation(full_event);
assert!(
lean_att.new_object_oids.is_empty(),
"a subscriber that did not ask for object ids must not be sent them"
);
assert_eq!(
lean_att.new_object_count, 2,
"…but the COUNT stays: 12 000 objects arriving is information, 12 000 oids are payload"
);
assert_eq!(full_att.new_object_oids.len(), 2);
// **The strip names itself.** Three different things now produce an empty
// `new_object_oids`: a push that introduced nothing, a store that never
// enumerated the set, and this — a subscriber that did not ask. The first
// two are told apart by `new_object_oids_unknown`, and a strip that left it
// empty would put a third meaning back on the same field and undo that.
assert!(
lean_att
.new_object_oids_unknown
.contains("include_attestation_object_ids"),
"the stripped event must say the SERVER removed the list, and name the flag that brings \
it back: {:?}",
lean_att.new_object_oids_unknown
);
assert!(
full_att.new_object_oids_unknown.is_empty(),
"nothing was stripped from this one, so it must carry no excuse: {:?}",
full_att.new_object_oids_unknown
);
// A record that ALREADY carried a reason keeps its own: "the store never
// enumerated" is the truer sentence than "the server stripped a list", and
// there was no list to strip.
bus.publish(event::Payload::PushAttestation(PushAttestation {
repo: Some(RepoPath::new("nordisk", "gunnar")),
new_object_count: 9,
new_object_oids_unknown: "the policy has no gate over objects, so the object set was \
not enumerated"
.into(),
..Default::default()
}));
let lean_att = unwrap_attestation(lean.next().await.expect("open").expect("no status"));
assert!(
lean_att
.new_object_oids_unknown
.contains("no gate over objects"),
"a reason the server already had must survive the strip: {:?}",
lean_att.new_object_oids_unknown
);
assert_eq!(lean_att.new_object_count, 9);
}
#[tokio::test]
async fn publishing_with_nobody_watching_is_not_an_error() {
// receive-pack publishes on every push. A push must never fail because no
// UI happened to be connected.
let bus = EventBus::new(4);
assert_eq!(bus.stats().subscribers, 0);
let event = bus.publish(deleted("nordisk", "gunnar"));
assert_eq!(event.seq, 1);
assert_eq!(bus.stats().published, 1);
assert_eq!(bus.stats().dropped, 0);
}
// ───────────────────────────────────────────────────────────────────────────
// resuming: the cursor, the gap it earns, and the epoch that says whether the
// cursor meant anything at all
//
// A resume feature is exactly the kind that passes without ever being able to
// fail — "the stream produced something" is satisfied by a server that ignored
// the cursor completely. So the tests below name the events they expect by
// their CONTENT (the object id each push carried, and its kind) rather than by
// how many arrived, and the ordering assertions are on positions, not on
// membership: a gap that arrives after the replay it was meant to precede is a
// gap that told a UI nothing.
// ───────────────────────────────────────────────────────────────────────────
/// A ref update whose `new_oid` is unique per `n`, so an event can be
/// identified by what it says rather than by where it sat in a queue.
fn updated(n: u64) -> event::Payload {
event::Payload::RefUpdated(gunnar_proto::RefUpdated {
repo: Some(RepoPath::new("nordisk", "gunnar")),
ref_name: "refs/heads/main".to_owned(),
old_oid: format!("{n:040x}"),
new_oid: format!("{:040x}", n + 1),
..Default::default()
})
}
fn oid_of(event: &gunnar_proto::Event) -> String {
match event.payload.as_ref() {
Some(event::Payload::RefUpdated(u)) => u.new_oid.clone(),
other => panic!("expected a RefUpdated, got {other:?}"),
}
}
fn opened_of(event: &gunnar_proto::Event) -> &gunnar_proto::StreamOpened {
match event.payload.as_ref() {
Some(event::Payload::StreamOpened(opened)) => opened,
other => panic!("expected the StreamOpened prologue, got {other:?}"),
}
}
fn kinds_of(events: &[gunnar_proto::Event]) -> Vec<i32> {
events.iter().map(|e| e.kind).collect()
}
/// Drain a stream until it goes quiet. Bounded: a stream that never produces
/// what a test waits for must FAIL, not hang.
async fn drain(stream: &mut gunnar_grpc::EventStream) -> Vec<gunnar_proto::Event> {
let mut out = Vec::new();
while let Ok(Some(item)) = tokio::time::timeout(QUIET, stream.next()).await {
out.push(item.expect("no status"));
}
out
}
#[tokio::test]
async fn a_resume_inside_the_buffer_delivers_exactly_what_followed_the_cursor() {
// Eight events into a ring of eight, then a resume from seq 3. What must
// come back is 4, 5, 6, 7, 8 — named by their object ids, in order, with
// nothing before them and nothing repeated.
let bus = EventBus::new(8);
for n in 1..=8 {
bus.publish(updated(n));
}
let mut stream = bus.subscribe_from(EventFilter::default(), 3);
let items = drain(&mut stream).await;
assert_eq!(
items[0].kind,
EventKind::StreamOpened as i32,
"the prologue must be the FIRST message: a client cannot interpret a sequence number \
before it knows which run assigned it. Got {:?}",
kinds_of(&items)
);
let opened = opened_of(&items[0]);
assert_eq!(
opened.from_seq, 3,
"the server must echo the cursor it acted on"
);
assert_eq!(opened.replaying, 5);
assert_eq!(opened.oldest_buffered_seq, 1);
assert_eq!(opened.last_seq, 8);
assert!(
items[1..]
.iter()
.all(|e| e.kind == EventKind::RefUpdated as i32),
"seq 4 was still in the ring, so nothing was lost and no gap may appear: {:?}",
kinds_of(&items)
);
let replayed: Vec<(u64, String)> = items[1..].iter().map(|e| (e.seq, oid_of(e))).collect();
assert_eq!(
replayed,
vec![
(4, format!("{:040x}", 5)),
(5, format!("{:040x}", 6)),
(6, format!("{:040x}", 7)),
(7, format!("{:040x}", 8)),
(8, format!("{:040x}", 9)),
],
"a resume must deliver the events AFTER the cursor, in order, and nothing else. A \
server that ignored `from_seq` and replayed the whole ring would start at seq 1; one \
that ignored the buffer would deliver nothing at all"
);
}
#[tokio::test]
async fn a_replay_joins_up_with_the_live_tail() {
// No event may fall into the seam between "what was buffered when I
// subscribed" and "what is published next", and none may cross it twice.
let bus = EventBus::new(8);
bus.publish(updated(1));
bus.publish(updated(2));
let mut stream = bus.subscribe_from(EventFilter::default(), 1);
bus.publish(updated(3));
let items = drain(&mut stream).await;
assert_eq!(
items[1..].iter().map(|e| e.seq).collect::<Vec<_>>(),
vec![2, 3],
"seq 2 was replayed and seq 3 arrived live; a seam that dropped or duplicated one \
shows here"
);
assert_eq!(oid_of(&items[2]), format!("{:040x}", 4));
}
#[tokio::test]
async fn a_resume_from_an_evicted_sequence_gets_the_gap_first_and_it_counts_right() {
// A ring of 4, twelve events published, a cursor at seq 1. Seqs 2..=8 are
// GONE — the ring holds 9..=12 — so seven were lost.
const CAPACITY: usize = 4;
let bus = EventBus::new(CAPACITY);
for n in 1..=12 {
bus.publish(updated(n));
}
let mut stream = bus.subscribe_from(EventFilter::default(), 1);
let items = drain(&mut stream).await;
assert_eq!(
items[0].kind,
EventKind::StreamOpened as i32,
"the prologue comes first: {:?}",
kinds_of(&items)
);
assert_eq!(
items[1].kind,
EventKind::Gap as i32,
"the gap must be the message IMMEDIATELY after the prologue, ahead of every replayed \
event. A UI told about the hole only after it has already rendered the events past it \
has already drawn a lie. Got: {:?}",
kinds_of(&items)
);
let gap = items[1].clone();
let gap = gap_of(&gap);
assert_eq!(
gap.dropped, 7,
"seqs 2..=8 fell out of a ring of {CAPACITY} before this subscriber asked for them"
);
assert_eq!(
gap.last_delivered_seq, 1,
"the gap names the cursor it answers"
);
assert_eq!(
gap.resume_seq, 9,
"…and the sequence the replay actually starts at"
);
assert!(
!gap.epoch.is_empty(),
"a gap must carry the epoch, or the numbers either side of it are not comparable"
);
let replayed: Vec<u64> = items[2..].iter().map(|e| e.seq).collect();
assert_eq!(
replayed,
vec![9, 10, 11, 12],
"everything the ring still held, and only that"
);
assert_eq!(
gap.dropped as usize + replayed.len() + 1,
12,
"dropped + replayed + the cursor's own event accounts for all twelve; a `dropped` \
invented rather than measured would not close"
);
assert!(
items[2..].iter().all(|e| e.kind != EventKind::Gap as i32),
"exactly one gap, and it is not trailing the replay"
);
}
#[tokio::test]
async fn a_cursor_the_buffer_can_still_reach_earns_no_gap() {
// The other direction. Without this, "always emit a gap" would pass the
// test above and the gap would carry no information at all.
let bus = EventBus::new(4);
for n in 1..=4 {
bus.publish(updated(n));
}
let mut stream = bus.subscribe_from(EventFilter::default(), 1);
let items = drain(&mut stream).await;
assert!(
items.iter().all(|e| e.kind != EventKind::Gap as i32),
"seq 2 was still in the ring, so nothing was lost and nothing may be claimed lost: {:?}",
kinds_of(&items)
);
assert_eq!(
items[1..].iter().map(|e| e.seq).collect::<Vec<_>>(),
vec![2, 3, 4]
);
}
#[tokio::test]
async fn no_cursor_still_starts_at_the_live_tail() {
// The case that must not have changed. Everything published BEFORE the
// subscription stays invisible: `Watch` is a notification channel, not a
// log, and `from_seq: 0` is how every existing client asks for that.
let bus = EventBus::new(16);
bus.publish(updated(1));
bus.publish(updated(2));
let mut stream = bus.subscribe_from(EventFilter::default(), 0);
bus.publish(updated(3));
let items = drain(&mut stream).await;
let opened = opened_of(&items[0]);
assert_eq!(opened.from_seq, 0);
assert_eq!(opened.replaying, 0, "no cursor means no replay");
assert_eq!(
opened.last_seq, 2,
"…but the prologue still says where the tail was"
);
assert_eq!(
items[1..].iter().map(|e| e.seq).collect::<Vec<_>>(),
vec![3],
"only the event published after the subscription. Seq 1 or 2 appearing here would mean \
`from_seq: 0` had quietly become a replay"
);
assert!(
items.iter().all(|e| e.kind != EventKind::Gap as i32),
"a subscriber with no cursor never claimed to have seen anything, so nothing is missing"
);
}
#[tokio::test]
async fn an_in_process_subscription_has_no_prologue_at_all() {
// `subscribe` is the replicator's door, and it is
// unchanged: no prologue, no replay, first message is the next event.
let bus = EventBus::new(8);
let mut stream = bus.subscribe(EventFilter::default());
bus.publish(updated(1));
let first = stream.next().await.expect("open").expect("no status");
assert_eq!(
first.kind,
EventKind::RefUpdated as i32,
"an in-process subscriber must still see its first published event first"
);
assert_eq!(first.seq, 1);
}
#[tokio::test]
async fn a_resume_across_a_restart_is_told_the_epoch_changed() {
// THE restart test. A console holds a cursor at seq 3 from one run of the
// server. The server restarts: a new `EventBus`, sequences from 1 again.
// The cursor is now a number that means nothing, and NOTHING in the
// sequence space can say so — seq 3 of the old run and seq 3 of the new one
// are the same number. Only the epoch can.
let before = EventBus::new(8);
for n in 1..=3 {
before.publish(updated(n));
}
let mut old_stream = before.subscribe_from(EventFilter::default(), 0);
let old = drain(&mut old_stream).await;
let old_opened = opened_of(&old[0]).clone();
assert_eq!(old_opened.last_seq, 3);
// ── the restart ──
drop(old_stream);
let after = EventBus::new(8);
after.publish(updated(100));
let mut resumed = after.subscribe_from(EventFilter::default(), 3);
let items = drain(&mut resumed).await;
let new_opened = opened_of(&items[0]);
assert_ne!(
new_opened.epoch, old_opened.epoch,
"a restarted server MUST report a different epoch. With the epoch unchanged a console \
carries its cursor across the restart, sees seq 1 where it expected seq 4, and either \
invents a gap or silently believes it missed nothing — which is the whole defect this \
field exists to close"
);
assert!(
!new_opened.epoch.is_empty(),
"an empty epoch compares equal to another empty epoch, which is worse than no epoch"
);
assert_eq!(
new_opened.last_seq, 1,
"the new run's sequence really did restart, which is what makes the old cursor unusable"
);
assert_eq!(
new_opened.from_seq, 3,
"the cursor is echoed even though it belongs to another run — the server cannot know \
that, and does not pretend to"
);
assert_eq!(
new_opened.replaying, 0,
"nothing in the new run's buffer is past seq 3, so nothing is replayed"
);
assert!(
items.iter().all(|e| e.kind != EventKind::Gap as i32),
"and no gap is invented from a cursor the server cannot place: a `dropped` count here \
would be fiction"
);
}
#[tokio::test]
async fn every_bus_mints_its_own_epoch() {
// The property the restart test rests on, asserted alone so a broken epoch
// fails by its own name rather than only inside a longer story.
let mut seen = std::collections::HashSet::new();
for _ in 0..64 {
let bus = EventBus::new(2);
assert!(!bus.epoch().is_empty(), "an epoch is never empty");
assert!(
seen.insert(bus.epoch().to_owned()),
"two buses shared the epoch {}. An epoch that does not change cannot report a \
restart, and a constant one is indistinguishable from no epoch at all",
bus.epoch()
);
}
}
#[tokio::test]
async fn every_stream_from_one_bus_reports_the_same_epoch() {
// The other half: the epoch identifies the RUN, not the subscription. If it
// changed per stream, every reconnect would look like a restart and the
// console would re-read the world for ever.
let bus = EventBus::new(4);
let first = drain(&mut bus.subscribe_from(EventFilter::default(), 0)).await;
let one = opened_of(&first[0]).epoch.clone();
bus.publish(updated(1));
let second = drain(&mut bus.subscribe_from(EventFilter::default(), 0)).await;
let two = opened_of(&second[0]).epoch.clone();
assert_eq!(one, two, "one bus, one epoch");
assert_eq!(one, bus.epoch(), "and it is the bus's own");
}
#[tokio::test]
async fn a_resume_gap_reaches_a_subscriber_whose_filter_matches_nothing() {
// Same rule as the lag gap: a subscriber must never be able to filter away
// the news that it lost events. Bounded, because a swallowed gap makes this
// stream produce nothing at all and an unbounded await would hang instead
// of failing.
let bus = EventBus::new(2);
for n in 1..=9 {
bus.publish(updated(n));
}
let mut narrow = bus.subscribe_from(
EventFilter {
namespaces: vec!["never-published-to".into()],
kinds: vec![EventKind::KeyUsed],
..Default::default()
},
1,
);
let items = drain(&mut narrow).await;
assert_eq!(items[0].kind, EventKind::StreamOpened as i32);
assert_eq!(
items[1].kind,
EventKind::Gap as i32,
"a resume gap must reach even a subscriber whose filter matches nothing: {:?}",
kinds_of(&items)
);
let gap = items[1].clone();
assert_eq!(gap_of(&gap).dropped, 6, "seqs 2..=7 are gone");
assert_eq!(
items.len(),
2,
"…and the replayed events themselves ARE filtered: this filter matches none of them"
);
}
#[tokio::test]
async fn a_resume_cannot_make_the_bus_hold_more_than_a_ringful() {
// The bound the whole design rests on. 10 000 events into a ring of 8, then
// a resume from the very beginning: the replay is capped by the ring, not
// by how far back the cursor points.
const CAPACITY: usize = 8;
let bus = EventBus::new(CAPACITY);
for n in 1..=10_000 {
bus.publish(updated(n));
}
let mut stream = bus.subscribe_from(EventFilter::default(), 1);
let items = drain(&mut stream).await;
assert_eq!(
items.len(),
CAPACITY + 2,
"the prologue, one gap, and at most a ringful. Anything more means a resume opened a \
second unbounded place for the server to grow"
);
let gap = items[1].clone();
assert_eq!(gap_of(&gap).dropped, 10_000 - CAPACITY as u64 - 1);
assert_eq!(items.last().unwrap().seq, 10_000, "the newest survives");
}
#[tokio::test]
async fn a_cursor_from_the_future_replays_nothing_and_claims_nothing() {
// A cursor past the newest event the bus ever assigned — what a stale
// cursor from a longer previous run looks like. It is not a gap: there is
// nothing between that number and now, and a `dropped` count would be
// invented.
let bus = EventBus::new(8);
bus.publish(updated(1));
let mut stream = bus.subscribe_from(EventFilter::default(), 900);
let items = drain(&mut stream).await;
assert_eq!(items.len(), 1, "the prologue and nothing else: {items:?}");
let opened = opened_of(&items[0]);
assert_eq!(opened.replaying, 0);
assert_eq!(opened.last_seq, 1);
assert_eq!(
opened.from_seq, 900,
"the cursor is reported back so a client can see for itself that the server is behind \
its own number"
);
}