From 700b6e5c5cf304cd7487feca9c20d734862c85e6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 23:48:35 -0700 Subject: [PATCH 1/5] quest: claim cross-relay-drill Co-Authored-By: Claude Opus 5.5 From 15a9afbecbc4af276054da671a28e0f601aba5d7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 00:21:20 -0700 Subject: [PATCH 2/5] test(relay): drill bursts across a two-relay cluster on impaired links Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 6 +- rs/moq-relay/tests/drills.rs | 350 +++++++++++++++++++++++++++--- 2 files changed, 328 insertions(+), 28 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index ea801d25c8..55006c5f77 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -4445,8 +4445,6 @@ impl kio::Task for FetchServeRun { match &mut self.state { FetchRunState::Open { request } => { - tracing::info!(broadcast = %self.serve.subscriber.log_path(&self.serve.path), track = %self.serve.name, group = self.group, "fetch started"); - // A peer that sent GOAWAY told us to stop opening streams on this session. if self.serve.subscriber.going_away.is_set() { request.take().expect("request pending").reject(Error::GoingAway); @@ -4468,6 +4466,10 @@ impl kio::Task for FetchServeRun { } }; + // Only once the stream opens: the open parks on stream credit and + // re-polls this state, so logging before it repeats per poll. + tracing::info!(broadcast = %self.serve.subscriber.log_path(&self.serve.path), track = %self.serve.name, group = self.group, "fetch started"); + let request = request.take().expect("request pending"); // A peer that predates lite-06 addresses whole groups only, so ask for diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 6ba7488364..237f38e728 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -96,6 +96,11 @@ struct Path { impl Path { async fn start(lane: Lane, relay: &RelayHost) -> Self { + Self::shaped(lane, relay, impairment()).await + } + + /// [`Self::start`], treating the impaired lane to `profile` both ways. + async fn shaped(lane: Lane, relay: &RelayHost, profile: moq_shaper::Profile) -> Self { let Lane::Impaired = lane else { return Self { url: relay.url(), @@ -113,8 +118,8 @@ impl Path { bind: "127.0.0.1:0".parse().unwrap(), target: format!("127.0.0.1:{}", relay.port).parse().unwrap(), seed, - up: impairment(), - down: impairment(), + up: profile.clone(), + down: profile, }) .await .expect("start the shaper"); @@ -160,20 +165,11 @@ struct RelayHost { } impl RelayHost { - /// Bind a relay and wait for it to be ready to accept. - async fn start(requested_port: Option) -> Self { + /// Bind a relay running `config` and wait for it to be ready to accept. + async fn start(config: Config) -> Self { // Process-global; every drill in this binary races to be first. let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); - let mut config = Config::default(); - config.listen.bind = Some( - format!("127.0.0.1:{}", requested_port.unwrap_or_default()) - .parse() - .unwrap(), - ); - config.listen.tls.generate = vec!["localhost".into()]; - config.auth.public = vec![moq_auth::Pattern::all()]; - let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(2) .enable_all() @@ -199,9 +195,6 @@ impl RelayHost { .expect("relay task vanished during startup") .expect("relay failed to load"); let port = addr.expect("relay did not bind UDP").port(); - if let Some(requested_port) = requested_port { - assert_eq!(port, requested_port, "relay bound the wrong port"); - } Self { port, @@ -229,6 +222,15 @@ impl RelayHost { } } +/// A public relay on loopback with a generated certificate, on `port` or an ephemeral one. +fn relay_config(port: Option) -> Config { + let mut config = Config::default(); + config.listen.bind = Some(format!("127.0.0.1:{}", port.unwrap_or_default()).parse().unwrap()); + config.listen.tls.generate = vec!["localhost".into()]; + config.auth.public = vec![moq_auth::Pattern::all()]; + config +} + impl Drop for RelayHost { fn drop(&mut self) { // A drill that ends (or panics) with the relay still up must not leave @@ -245,12 +247,11 @@ impl Drop for RelayHost { /// path, and a silent fallback would grade a different transport than the one /// under test. fn client(url: &url::Url) -> moq_tokio::Client { - let mut config = moq_tokio::connect::Config::default(); - config.url = Some(url.clone()); - config.bind = Some("127.0.0.1:0".parse().expect("parse client bind")); - config.tls.insecure = Some(true); - config.websocket.enabled = Some(false); + client_over(url, quic()) +} +/// The drills' QUIC tuning for a client. +fn quic() -> moq_tokio::quic::Config { // A killed relay sends no CONNECTION_CLOSE, so the idle timeout is the only // thing that ever tells a client it is gone. The 30s default would put that // discovery past every budget here. The keep-alive is an eighth of it, so a @@ -259,6 +260,16 @@ fn client(url: &url::Url) -> moq_tokio::Client { let mut quic = moq_tokio::quic::Config::default(); quic.idle_timeout = Duration::from_secs(2); quic.keep_alive = Duration::from_millis(250); + quic +} + +/// [`client`], with `quic` as its transport tuning. +fn client_over(url: &url::Url, quic: moq_tokio::quic::Config) -> moq_tokio::Client { + let mut config = moq_tokio::connect::Config::default(); + config.url = Some(url.clone()); + config.bind = Some("127.0.0.1:0".parse().expect("parse client bind")); + config.tls.insecure = Some(true); + config.websocket.enabled = Some(false); // Fast enough to keep a relay bounce inside the drill's budget, paced enough // that the loop is still a backoff. @@ -370,7 +381,7 @@ async fn subscribe(origin: &moq_net::origin::Consumer, path: &str) -> Reader { /// keeps being served. The rejoin below is the half of that behavior worth /// grading here. async fn cancel_under_backpressure_releases_the_reader(lane: Lane) { - let relay = RelayHost::start(None).await; + let relay = RelayHost::start(relay_config(None)).await; let path = Path::start(lane, &relay).await; let url = path.url.clone(); @@ -492,7 +503,7 @@ lanes!(cancel_under_backpressure_releases_the_reader); /// restore service: the dead session's broadcast closes with it, and a fresh /// subscribe through the same origin resumes once the relay returns. async fn relay_killed_mid_group_aborts_then_resumes(lane: Lane) { - let mut relay = RelayHost::start(None).await; + let mut relay = RelayHost::start(relay_config(None)).await; let port = relay.port; let path = Path::start(lane, &relay).await; let url = path.url.clone(); @@ -544,7 +555,8 @@ async fn relay_killed_mid_group_aborts_then_resumes(lane: Lane) { expect_status(&mut subscribe_loop, moq_tokio::Status::Disconnected, "subscriber").await; expect_status(&mut publish_loop, moq_tokio::Status::Disconnected, "publisher").await; - let relay = RelayHost::start(Some(port)).await; + let relay = RelayHost::start(relay_config(Some(port))).await; + assert_eq!(relay.port, port, "relay bound the wrong port"); expect_status(&mut subscribe_loop, moq_tokio::Status::Connected, "subscriber").await; expect_status(&mut publish_loop, moq_tokio::Status::Connected, "publisher").await; @@ -597,7 +609,7 @@ async fn expect_status(reconnect: &mut moq_tokio::Connection, want: moq_tokio::S /// subscriber that re-consumes the same name after a republish gets what the /// new publisher is sending, never the previous one's cache. async fn interrupted_publisher_republishes_new_content(lane: Lane) { - let relay = RelayHost::start(None).await; + let relay = RelayHost::start(relay_config(None)).await; let path = Path::start(lane, &relay).await; let url = path.url.clone(); @@ -706,6 +718,292 @@ async fn expect_announce(announced: &mut moq_net::announce::Consumer, path: &str } } +/// How many bursts the cross-relay drill sends, and how many one-frame groups +/// each holds: a control channel firing messages back to back. +const BURSTS: u64 = 4; +const BURST_GROUPS: u64 = 30; + +/// Each cross-relay group's payload size, about what a small control message is. +const PAYLOAD: usize = 256; + +/// How long a cross-relay group may take, from being written, to arrive. Several +/// times the slowest group seen on the impaired path, and a third of the 30s +/// stalls of #4349. +const SETTLE: Duration = Duration::from_secs(10); + +/// The cross-relay drill's link: the drills' impairment behind a bottleneck a +/// burst overruns, so its tail waits in the queue or is dropped from it. +fn bursty() -> moq_shaper::Profile { + moq_shaper::Profile { + rate: Some(moq_shaper::Rate { + bits_per_second: 1_000_000, + burst: 1500, + queue: Duration::from_millis(50), + }), + ..impairment() + } +} + +/// QUIC credit tighter than a burst on the impaired lane: fewer concurrent +/// streams than a burst has groups (each group is a stream, each FETCH another), +/// and receive windows of a few groups, so senders wait on MAX_STREAMS and +/// MAX_DATA mid-burst, over a path that loses the frames granting them. +fn credit(lane: Lane, mut quic: moq_tokio::quic::Config) -> moq_tokio::quic::Config { + if let Lane::Impaired = lane { + quic.max_streams = Some(16); + quic.receive_window = Some(4 * PAYLOAD as u64); + quic.stream_receive_window = Some(2 * PAYLOAD as u64); + } + quic +} + +/// How a cross-relay group reached the subscriber, or why it never did. +#[derive(Debug)] +enum Outcome { + /// The live subscription delivered it. + Live, + /// A FETCH recovered it after the subscription skipped or lost it. + Fetched, + /// The FETCH failed, with this error. + Failed(String), + /// The FETCH got no answer within [`SETTLE`]. + Unanswered, + /// The live subscription opened it but its frame never came within [`SETTLE`]. + Stalled, +} + +/// One thing the cross-relay subscriber waits on. +enum Step { + /// The live subscription's next group. + Group(Box>>), + /// A group read or FETCH finished. + Settled(u64, Outcome), +} + +/// Read the one frame of group `sequence`, checking it is the payload written there. +async fn read_payload(mut group: moq_net::group::Consumer, sequence: u64) -> Result<(), String> { + match group.read_frame().await { + Ok(Some(frame)) if frame.payload == payload(sequence) => Ok(()), + Ok(Some(_)) => panic!("group {sequence} carried another group's payload"), + Ok(None) => Err("finished with no frame".to_string()), + Err(err) => Err(err.to_string()), + } +} + +/// The payload of group `sequence`: its number, padded to [`PAYLOAD`]. +fn payload(sequence: u64) -> Vec { + format!("{sequence:>PAYLOAD$}").into_bytes() +} + +/// Drill: bursts of one-frame groups cross a two-relay cluster. +/// +/// A publisher on an origin relay sends bursts of tens of single-frame groups. +/// A subscriber on an edge relay, which reaches the origin as a cluster peer, +/// reads them with the default subscription (only the newest group) and FETCHes +/// every group below it that the subscription skipped or reset, the way a +/// control channel that must apply every message recovers. In the impaired lane +/// every hop (publisher, peer link, subscriber) runs through a shaper that a +/// burst overruns, and every endpoint grants less QUIC credit than a burst needs. +/// +/// Nothing is evicted and no link goes down, so every group has to arrive within +/// [`SETTLE`] of being written; one that does not fails the drill by name, with +/// how it was lost: a FETCH that failed, a FETCH nobody answered, or a live group +/// that stalled. The newest group has to come over the live subscription, since +/// nothing newer can replace it. +async fn bursts_cross_a_cluster(lane: Lane) { + let mut config = relay_config(None); + config.quic = credit(lane, config.quic); + let origin = RelayHost::start(config).await; + let peer = Path::shaped(lane, &origin, bursty()).await; + + // The edge relay dials the origin as a cluster peer, through its own path. + let mut peer_url = peer.url.clone(); + peer_url.set_path("/"); + let mut config = relay_config(None); + config.quic = credit(lane, config.quic); + config.connect.bind = Some("127.0.0.1:0".parse().unwrap()); + config.connect.tls.insecure = Some(true); + config.connect.websocket.enabled = Some(false); + config.cluster.connect = vec![moq_relay::cluster::Peer::new(peer_url.to_string())]; + let edge = RelayHost::start(config).await; + + let publish_path = Path::shaped(lane, &origin, bursty()).await; + let subscribe_path = Path::shaped(lane, &edge, bursty()).await; + + let publisher = moq_tokio::origin::spawn(); + let broadcast = publisher.create_broadcast("live").expect("create broadcast"); + broadcast.announce(Default::default()).expect("announce broadcast"); + let mut track = broadcast.create_track(TRACK, None).expect("create track"); + let url = publish_path.url.clone(); + let publish_session = tokio::time::timeout( + TIMEOUT, + client_over(&url, credit(lane, quic())) + .with_publisher(&publisher) + .with_reconnect(false) + .connect(url) + .established(), + ) + .await + .expect("publisher connect timed out") + .expect("publisher connect failed"); + + let subscriber = moq_tokio::origin::spawn(); + let subscribed = subscriber.consume(); + let url = subscribe_path.url.clone(); + let subscribe_session = tokio::time::timeout( + TIMEOUT, + client_over(&url, credit(lane, quic())) + .with_subscriber(subscriber) + .with_reconnect(false) + .connect(url) + .established(), + ) + .await + .expect("subscriber connect timed out") + .expect("subscriber connect failed"); + + // Subscribed before the first burst, as an app watching a control channel is. + let mut reader = subscribe(&subscribed, "live").await; + + let total = BURSTS * BURST_GROUPS; + let mut written = Vec::new(); + let burst = |track: &mut moq_net::track::Producer, written: &mut Vec| { + for _ in 0..BURST_GROUPS { + let sequence = written.len() as u64; + write_group(track, &payload(sequence)); + written.push(tokio::time::Instant::now()); + } + }; + burst(&mut track, &mut written); + + // Each group is claimed once, by the subscription or by a FETCH, and settles + // in `pending`. A live group that fails is a gap like any other, so it is + // FETCHed in turn. + let mut claimed = std::collections::BTreeSet::new(); + let mut live = std::collections::BTreeSet::new(); + let mut outcomes = std::collections::BTreeMap::new(); + let mut pending = tokio::task::JoinSet::new(); + let mut resets = Vec::new(); + let mut slowest = (Duration::ZERO, 0); + let fetch = |pending: &mut tokio::task::JoinSet<_>, track: &moq_net::track::Consumer, sequence: u64| { + let fetching = track.fetch_group(sequence, None); + pending.spawn(async move { + let fetched = tokio::time::timeout(SETTLE, async { + let group = fetching.await.map_err(|err| err.to_string())?; + read_payload(group, sequence).await + }); + let outcome = match fetched.await { + Ok(Ok(())) => Outcome::Fetched, + Ok(Err(err)) => Outcome::Failed(err), + Err(_) => Outcome::Unanswered, + }; + (sequence, outcome) + }); + }; + + while (outcomes.len() as u64) < total { + let step = tokio::time::timeout(SETTLE, async { + tokio::select! { + group = reader.groups.recv_group() => Step::Group(Box::new(group)), + Some(settled) = pending.join_next() => { + let (sequence, outcome) = settled.expect("settle task panicked"); + Step::Settled(sequence, outcome) + } + } + }) + .await + .unwrap_or_else(|_| { + let missing: Vec = (0..total).filter(|sequence| !outcomes.contains_key(sequence)).collect(); + panic!( + "stalled: nothing arrived within {SETTLE:?}, {} of {total} groups missing: {missing:?}", + missing.len() + ) + }); + + match step { + Step::Settled(sequence, Outcome::Failed(err)) if live.remove(&sequence) => { + resets.push((sequence, err)); + fetch(&mut pending, &reader.track, sequence); + } + Step::Settled(sequence, outcome) => { + live.remove(&sequence); + let late = written[sequence as usize].elapsed(); + slowest = slowest.max((late, sequence)); + outcomes.insert(sequence, outcome); + } + Step::Group(group) => { + let group = (*group) + .unwrap_or_else(|err| panic!("the subscription aborted: {err}")) + .unwrap_or_else(|| panic!("the subscription finished")); + let sequence = group.sequence; + + // Every unclaimed group below this one is a gap the subscription + // skipped: FETCH it now rather than wait. + for gap in 0..sequence { + if claimed.insert(gap) { + fetch(&mut pending, &reader.track, gap); + } + } + if claimed.insert(sequence) { + live.insert(sequence); + pending.spawn(async move { + let outcome = match tokio::time::timeout(SETTLE, read_payload(group, sequence)).await { + Ok(Ok(())) => Outcome::Live, + Ok(Err(err)) => Outcome::Failed(err), + Err(_) => Outcome::Stalled, + }; + (sequence, outcome) + }); + } + + // The next burst leaves once this one starts arriving, so recovery + // overlaps fresh data the way a busy control channel's does. + if sequence + BURST_GROUPS >= written.len() as u64 && (written.len() as u64) < total { + burst(&mut track, &mut written); + } + } + } + } + + let fetched = outcomes + .values() + .filter(|outcome| matches!(outcome, Outcome::Fetched)) + .count(); + println!( + "fault activated: {fetched} of {total} groups recovered by FETCH, live groups reset {resets:?}, slowest group {} after {:?}", + slowest.1, slowest.0 + ); + assert!(fetched > 0, "every group came live, so no FETCH recovery was exercised"); + + let lost: Vec<_> = outcomes + .iter() + .filter(|(_, outcome)| !matches!(outcome, Outcome::Live | Outcome::Fetched)) + .collect(); + assert!(lost.is_empty(), "groups lost: {lost:?}"); + assert!( + slowest.0 <= SETTLE, + "group {} took {:?} to arrive, past {SETTLE:?}", + slowest.1, + slowest.0 + ); + assert!( + matches!(outcomes.get(&(total - 1)), Some(Outcome::Live)), + "the newest group came by {:?}, not the live subscription", + outcomes.get(&(total - 1)) + ); + + drop(reader); + drop(subscribe_session); + drop(publish_session); + drop(edge); + drop(origin); + peer.verify(); + publish_path.verify(); + subscribe_path.verify(); +} + +lanes!(bursts_cross_a_cluster); + /// Negative control: with nothing publishing, the drills' delivery assertions /// have to fail. /// @@ -713,7 +1011,7 @@ async fn expect_announce(announced: &mut moq_net::announce::Consumer, path: &str /// harness with no publisher and requires that no announcement and no broadcast /// ever arrive, so "the frame showed up" cannot be a harness artifact. async fn no_publisher_never_delivers(lane: Lane) { - let relay = RelayHost::start(None).await; + let relay = RelayHost::start(relay_config(None)).await; let path = Path::start(lane, &relay).await; let url = path.url.clone(); From 6ca1afef853459feb1351c2cfd6516cc9e87319a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 00:28:45 -0700 Subject: [PATCH 3/5] test(relay): grade the cluster drill's FETCH recovery and finish the quest Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/cross-relay-bursts.md | 14 +++++--- quest/m1/cross-relay-drill.md | 37 --------------------- rs/moq-relay/tests/drills.rs | 20 ++++++----- test/drill/README.md | 18 ++++++++-- test/drill/mutations/fetch-never-sent.patch | 24 +++++++++++++ 6 files changed, 60 insertions(+), 54 deletions(-) delete mode 100644 quest/m1/cross-relay-drill.md create mode 100644 test/drill/mutations/fetch-never-sent.patch diff --git a/quest/m1/README.md b/quest/m1/README.md index 5cd69305a1..287d3a021f 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -58,7 +58,6 @@ blocks. The quests that gated m0 lines moved under them. - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges in either order and replaces FETCH; relays fill misses by range, including over moq-transport - [Live media time](/quest/m1/subscribe-live-time.md) - re-scoped against `set_live`: a lite-07 SUBSCRIBE_OK carries the publisher's current media time only if a reader still needs it - [Cross-relay bursts re-run](/quest/m1/cross-relay-bursts.md) - condition: the #4349 reporter re-runs their A/B/C comparison against current cdn.moq.pro -- [Two-relay drill on impaired links](/quest/m1/cross-relay-drill.md) - a bursty small-group track crosses two clustered relays over lossy, delayed, flow-limited QUIC without unanswered FETCHes or `Old` stalls - [Untimed lite-07](/quest/m1/lite-untimed.md) - lite-07 carries an untimed track in both languages; lite-05/06 write send time - [Finalize moq-lite-07](/quest/m1/lite07-finalize.md) - when the maintainer cuts it, lite-07 negotiates as `moq-lite-07` and the next release ships it - [Late lower groups](/quest/m1/lite-late-lower-group.md) - a moq-lite subscriber with a floor receives a group created below the first served one, as moq-transport does diff --git a/quest/m1/cross-relay-bursts.md b/quest/m1/cross-relay-bursts.md index 34741261dc..66fba74b59 100644 --- a/quest/m1/cross-relay-bursts.md +++ b/quest/m1/cross-relay-bursts.md @@ -26,10 +26,15 @@ used and against `main`: route-flap drops (about 70 of 560 groups) reproduce only on 0.15.6 and are gone on moq-net 0.3.8 and later, most likely by #4387. - **Not reproduced.** Unanswered FETCHes and the 30 s `Stream(Old)` stalls - showed up on no build. The mock models neither loss nor flow control, so - [Two-relay drill on impaired links](/quest/m1/cross-relay-drill.md) covers - that gap. Several 30 s timers of that era have since been removed or - shortened (#4606, #4741). + showed up on no build, nor in the real-QUIC drill + `rs/moq-relay/tests/drills.rs::bursts_cross_a_cluster` (bursts of 30 + across two clustered relays, every hop with 5% loss, delay, a 1 Mbit/s + bottleneck, and 16-stream QUIC credit): every group arrives within about + 3 s, and the occasional live `Old` reset is recovered by FETCH. Several + 30 s timers of that era have since been removed or shortened (#4606, #4741). +- **Queueing.** FETCHes beyond a session's stream credit wait for it, one + cross-relay round trip per slot, so with tight credit a gap's FETCH can + outlast a 2 s deadline without being lost. - **Expected.** Across relays a burst arrives newest-first, as the lite draft specifies; fetching on every gap at once multiplies FETCHes. @@ -39,5 +44,4 @@ used and against `main`: ## Related -- [Two-relay drill on impaired links](/quest/m1/cross-relay-drill.md) - looks for the unreproduced FETCH and `Old` stalls without waiting on the reporter - [Routes and announces](/quest/m1/cluster-routing/routes.md) - owns the stale and flapping announcements from the same report diff --git a/quest/m1/cross-relay-drill.md b/quest/m1/cross-relay-drill.md deleted file mode 100644 index 249336d9b5..0000000000 --- a/quest/m1/cross-relay-drill.md +++ /dev/null @@ -1,37 +0,0 @@ -# [M] Two-relay drill on impaired links - -## Goal - -A real-QUIC test runs a bursty small-group track (one frame per group, bursts -of tens of groups) from a publisher on one relay to a subscriber on a -clustered second relay, with the links impaired by loss, delay, and -flow-control pressure. It shows every group either arriving or failing by -name within a bounded time: no FETCH left unanswered, and no track stalled -by relay-hop `Old` resets. A failure it finds gets fixed at the cause, or a -quest of its own if large. - -The mock two-relay harness behind -[#4349](https://github.com/moq-dev/moq/issues/4349) cannot produce those -resets, so the issue's unanswered FETCHes and 30 s `Stream(Old)` stalls are -still unexplained. - -## Plan - -`rs/moq-relay/tests/drills.rs` already runs single-relay drills over real -QUIC in a loopback and a `moq_shaper` impaired lane, and the cluster tests -(`cluster_idle.rs`, `goaway_cluster.rs`) start peered relays in-process. -Open for the planner: - -- A new drill in that file's lanes or a separate two-relay test, and which - hop or hops (client and peer links) get shaped. -- How to apply flow-control pressure: tight QUIC windows, a rate limit below - the burst rate, or both. -- The subscriber's recovery model: fetch-on-gap with a deadline, as the - reporter's app does, versus waiting and reordering locally. -- Whether a burst plus a flapping peer link joins as a regression for the - 0.15.6 losses fixed in moq-net 0.3.8. - -## Related - -- [Cross-relay bursts re-run](/quest/m1/cross-relay-bursts.md) - waits on the #4349 reporter's re-run against current cdn.moq.pro -- [Routes and announces](/quest/m1/cluster-routing/routes.md) - the cluster routing a flapping peer link exercises diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 237f38e728..a2ee31a894 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -1,8 +1,9 @@ //! Transport and lifecycle failure drills against a real relay. //! //! Delivery working on a healthy loopback says nothing about cancellation, a -//! relay dying mid-group, or a publisher coming back. Each drill here disrupts a -//! live session over real QUIC and asserts three things: +//! relay dying mid-group, a publisher coming back, or bursts crossing a relay +//! cluster. Each drill here disrupts a live session over real QUIC and asserts +//! three things: //! //! 1. the fault actually activated, recorded as a value the test prints; //! 2. the failure surfaced as a terminal result, never a clean finish or a hang; @@ -11,8 +12,9 @@ //! Each drill runs in two lanes, generated by [`lanes!`] so a scenario cannot //! drift between them: `loopback`, and `impaired`, where every client reaches //! the relay through a seeded [`moq_shaper`] that delays, jitters, drops, -//! reorders, and rate-limits the path both ways. A drill that only passes on -//! loopback is a finding, not a flake. +//! reorders, and rate-limits the path both ways (and, across a cluster, the +//! peer link between relays too). A drill that only passes on loopback is a +//! finding, not a flake. //! //! `no_publisher_never_delivers` is the negative control: the same harness with //! nothing publishing must report failure, so a drill that passes because no @@ -965,15 +967,17 @@ async fn bursts_cross_a_cluster(lane: Lane) { } } - let fetched = outcomes + // Fault activation: the subscription skipped or reset groups, so the FETCH + // recovery below had something to recover. + let gaps = outcomes .values() - .filter(|outcome| matches!(outcome, Outcome::Fetched)) + .filter(|outcome| !matches!(outcome, Outcome::Live)) .count(); println!( - "fault activated: {fetched} of {total} groups recovered by FETCH, live groups reset {resets:?}, slowest group {} after {:?}", + "fault activated: {gaps} of {total} groups missed the live subscription (reset: {resets:?}), slowest group {} after {:?}", slowest.1, slowest.0 ); - assert!(fetched > 0, "every group came live, so no FETCH recovery was exercised"); + assert!(gaps > 0, "every group came live, so no FETCH recovery was exercised"); let lost: Vec<_> = outcomes .iter() diff --git a/test/drill/README.md b/test/drill/README.md index bf3600f75d..08c38acf75 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -2,8 +2,9 @@ Media flowing on a healthy loopback proves delivery and nothing else. These drills disrupt a live session through a real relay over real QUIC and check what -happens next: cancellation, a relay dying mid-group, and a publisher coming back -under a name it already used. +happens next: cancellation, a relay dying mid-group, a publisher coming back +under a name it already used, and bursts of small groups crossing a two-relay +cluster. The drills themselves are ordinary Rust integration tests in `rs/moq-relay/tests/drills.rs`, so `just check` already runs them whenever @@ -28,6 +29,7 @@ checks that what the fault owned was released. | `cancel_under_backpressure_releases_the_reader` | A subscriber stops reading while the publisher runs ahead, then cancels | the number of unread groups queued behind the reader at the moment it cancels | everything the cancelled session was feeding closes rather than parking, and the relay still serves a subscriber that joins afterwards | | `relay_killed_mid_group_aborts_then_resumes` | The relay's runtime is dropped mid-group: every task and socket at once, no `CONNECTION_CLOSE` | the error the interrupted group aborted with, and both reconnect loops reporting `Disconnected` | the interrupted group fails rather than reporting a clean finish, and after the relay returns the same handles resume delivery | | `interrupted_publisher_republishes_new_content` | A publisher vanishes with no finish and no unannounce, then a new one publishes the same name | the withdrawal of the dead publisher's broadcast | the name stops being announced, and the republished name serves the new publisher's content rather than the dead one's cache | +| `bursts_cross_a_cluster` | Bursts of one-frame groups cross from an origin relay to a clustered edge relay; the subscriber reads newest-only and FETCHes every group it misses | how many groups missed the live subscription, which live groups were reset, and the slowest group's latency | every group arrives, live or by FETCH, within 10s of being written; a lost group fails by name (FETCH failed, FETCH unanswered, live group stalled), and the newest group comes live | | `no_publisher_never_delivers` | none: the negative control | - | with nothing publishing, nothing is announced and no broadcast resolves | A moq publisher is never blocked by a slow reader: `write_frame` is synchronous @@ -41,7 +43,7 @@ reader leaves, and is covered by moq-net's origin tests. Waiting out a linger would make this the slowest test in the workspace; the rejoin the drill does assert is the half of that behavior worth grading here. -The negative control is what keeps the rest honest. Three drills prove things by +The negative control is what keeps the rest honest. Four drills prove things by reading a frame; if the harness could report success without data moving, they would all pass for free. @@ -71,6 +73,14 @@ forwards each datagram, both ways, after applying the same profile: | reorder | 2% skip the delay and overtake whatever is in flight | | rate | 10 Mbit/s token bucket, 1500 byte burst, 100ms queue then tail drop | +`bursts_cross_a_cluster` shapes all three hops, the publisher's, the peer link +between the relays, and the subscriber's, each with that profile behind a +1 Mbit/s bottleneck with a 50ms queue, which a burst overruns. Its impaired lane +also grants less QUIC credit than a burst needs at every endpoint: 16 concurrent +streams (each group is a stream, each FETCH another) and receive windows of a +few groups, so senders wait on `MAX_STREAMS` and `MAX_DATA` that the path can +lose. Its loopback lane keeps the defaults. + It is a datagram relay, not an HTTP or TCP proxy, neither of which can impair QUIC. It needs no capabilities, works the same on macOS and Linux, and touches nothing on the host; QUIC is indifferent to the extra hop. @@ -133,6 +143,7 @@ and grades the same body. | `subscriber-leaks-broadcasts` | releasing the broadcasts a subscribing session fed when that session ends | `cancel_under_backpressure_releases_the_reader::loopback` | | `reconnect-stops-after-session-loss` | redialing after an established session is lost | `relay_killed_mid_group_aborts_then_resumes::loopback` | | `relay-withdraws-lost-publisher` | withdrawing a publisher's announcements when its session is lost | `interrupted_publisher_republishes_new_content::loopback` | +| `fetch-never-sent` | sending a FETCH for an uncached group, at the subscriber and at the edge relay | `bursts_cross_a_cluster::loopback` | A mutated tree that fails to compile is a failure of the proof, not a pass: a compile error shows the patch touched something, not that the drill was @@ -155,6 +166,7 @@ drill fails, these are where the mechanism is pinned. | cancel under backpressure | `kio::loom::{last_consumer_wakes_unused, consumer_churn_resolves_unused, first_consumer_wakes_used}`, `moq-net tests/loom.rs::{subscriber_wakes_parked_demand, concurrent_tracks_drain_a_shared_pool}` | `lite` (`SUBSCRIBE` / `SUBSCRIBE_UPDATE` encodings) | | relay killed mid-group | `kio::loom::{racing_last_producer_drops_still_close, queue_close_wakes_a_parked_pop, write_wakes_a_parked_consumer}`, `moq-net tests/loom.rs::publisher_drop_resolves_a_parked_subscriber` | `lite` and `ietf` (a truncated stream must fail to decode, never decode into something else), regression `ietf/subscribe-absolute-filter` | | interrupted publisher republishes | `kio::loom::{write_wakes_a_parked_consumer, weak_upgrade_never_resurrects_a_closed_channel}`, `moq-net tests/loom.rs::back_to_back_groups_arrive_in_order` | `path` (the broadcast name is a path: normalization, prefixes, and the wire round-trip) | +| bursts across a cluster | `moq-net tests/loom.rs::{a_fetch_never_joins_a_withdrawn_attempt, back_to_back_groups_arrive_in_order}` | `lite` (`FETCH` encodings) | Run them with `just rs loom` and `just rs fuzz `. The committed fuzz findings under `rs/moq-net/fuzz/regressions/` replay on stable as part of diff --git a/test/drill/mutations/fetch-never-sent.patch b/test/drill/mutations/fetch-never-sent.patch new file mode 100644 index 0000000000..22a5a13f4a --- /dev/null +++ b/test/drill/mutations/fetch-never-sent.patch @@ -0,0 +1,24 @@ +# drill: bursts_cross_a_cluster::loopback +# expect: Unanswered +# +# Removes sending a FETCH. A fetch of an uncached group parks before it opens +# its stream, so the subscriber's request for a skipped group, and the edge +# relay's request upstream for it, never reach anything that could answer. The +# live subscription still delivers each burst's newest group, so only the FETCH +# recovery of the groups it skipped is removed. +diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs +index 55006c5f7..93f2f9677 100644 +--- a/rs/moq-net/src/lite/subscriber.rs ++++ b/rs/moq-net/src/lite/subscriber.rs +@@ -4445,6 +4445,11 @@ impl kio::Task for FetchServeRun { + + match &mut self.state { + FetchRunState::Open { request } => { ++ // MUTATION: never send the FETCH, so nothing ever answers it. ++ if request.is_some() { ++ return Poll::Pending; ++ } ++ + // A peer that sent GOAWAY told us to stop opening streams on this session. + if self.serve.subscriber.going_away.is_set() { + request.take().expect("request pending").reject(Error::GoingAway); From c482cb9d956b4f372e1d31d9edfc0d9cdb8e986e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 01:33:31 -0700 Subject: [PATCH 4/5] test(relay): make the cluster drill's bottleneck overflow and replayable The 1 KiB receive window capped each sender near 160 kbit/s, so the 1 Mbit/s bottleneck never overflowed. Drop it to 100 kbit/s and fail the impaired lane unless some burst overflowed a queue. Seed the three shaped paths off one printed base so MOQ_SHAPER_SEED replays the run, and give the step timeout a margin past each group's own so a lost group fails by name. Co-Authored-By: Claude Opus 5.5 --- quest/m1/cross-relay-bursts.md | 6 +-- rs/moq-relay/tests/drills.rs | 74 ++++++++++++++++++++++------------ test/drill/README.md | 19 +++++---- 3 files changed, 62 insertions(+), 37 deletions(-) diff --git a/quest/m1/cross-relay-bursts.md b/quest/m1/cross-relay-bursts.md index 66fba74b59..81ffedf2fa 100644 --- a/quest/m1/cross-relay-bursts.md +++ b/quest/m1/cross-relay-bursts.md @@ -28,9 +28,9 @@ used and against `main`: - **Not reproduced.** Unanswered FETCHes and the 30 s `Stream(Old)` stalls showed up on no build, nor in the real-QUIC drill `rs/moq-relay/tests/drills.rs::bursts_cross_a_cluster` (bursts of 30 - across two clustered relays, every hop with 5% loss, delay, a 1 Mbit/s - bottleneck, and 16-stream QUIC credit): every group arrives within about - 3 s, and the occasional live `Old` reset is recovered by FETCH. Several + across two clustered relays, every hop with 5% loss, delay, a 100 kbit/s + bottleneck that each burst overflows, and 16-stream QUIC credit): every + group arrives within about 3.5 s, and the occasional live `Old` reset is recovered by FETCH. Several 30 s timers of that era have since been removed or shortened (#4606, #4741). - **Queueing.** FETCHes beyond a session's stream credit wait for it, one cross-relay round trip per slot, so with tight credit a gap's FETCH can diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index a2ee31a894..3d66ef96e0 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -98,11 +98,12 @@ struct Path { impl Path { async fn start(lane: Lane, relay: &RelayHost) -> Self { - Self::shaped(lane, relay, impairment()).await + Self::shaped(lane, relay, impairment(), seed(lane)).await } - /// [`Self::start`], treating the impaired lane to `profile` both ways. - async fn shaped(lane: Lane, relay: &RelayHost, profile: moq_shaper::Profile) -> Self { + /// [`Self::start`], treating the impaired lane to `profile` both ways with + /// the shaper seeded by `seed`. + async fn shaped(lane: Lane, relay: &RelayHost, profile: moq_shaper::Profile, seed: u64) -> Self { let Lane::Impaired = lane else { return Self { url: relay.url(), @@ -110,12 +111,6 @@ impl Path { }; }; - // A fresh seed each run explores more of the decision space; the one - // printed below replays a failure's decisions. - let seed = match std::env::var("MOQ_SHAPER_SEED") { - Ok(seed) => seed.parse().expect("MOQ_SHAPER_SEED is not a u64"), - Err(_) => rand::random(), - }; let shaper = moq_shaper::Shaper::bind(moq_shaper::Config { bind: "127.0.0.1:0".parse().unwrap(), target: format!("127.0.0.1:{}", relay.port).parse().unwrap(), @@ -125,7 +120,7 @@ impl Path { }) .await .expect("start the shaper"); - println!("impaired: MOQ_SHAPER_SEED={seed}, profile {:?}", shaper.config().up); + println!("impaired: seed {seed}, profile {:?}", shaper.config().up); Self { url: format!("https://{}/drill", shaper.addr()).parse().unwrap(), @@ -136,15 +131,30 @@ impl Path { /// Fail the drill unless the impairment it ran under actually acted. /// /// A profile that silently did nothing would turn this lane into a second - /// loopback run that passes for free. - fn verify(mut self) { - if let Some(shaper) = self.shaper.take() { - let stats = shaper.verify().unwrap_or_else(|err| panic!("impaired: {err:#}")); - println!("impaired: {stats}"); - } + /// loopback run that passes for free. Returns what the shaper did, if any. + fn verify(mut self) -> Option { + let shaper = self.shaper.take()?; + let stats = shaper.verify().unwrap_or_else(|err| panic!("impaired: {err:#}")); + println!("impaired: {stats}"); + Some(stats) } } +/// A drill's base shaper seed, printed in the impaired lane. A fresh seed each +/// run explores more of the decision space; `MOQ_SHAPER_SEED` set to the +/// printed one replays a failure's decisions. A drill with several paths seeds +/// each one off it. +fn seed(lane: Lane) -> u64 { + let seed = match std::env::var("MOQ_SHAPER_SEED") { + Ok(seed) => seed.parse().expect("MOQ_SHAPER_SEED is not a u64"), + Err(_) => rand::random(), + }; + if let Lane::Impaired = lane { + println!("impaired: MOQ_SHAPER_SEED={seed}"); + } + seed +} + impl Drop for Path { fn drop(&mut self) { // A drill that failed before `verify` still reports what the path did. @@ -734,11 +744,14 @@ const PAYLOAD: usize = 256; const SETTLE: Duration = Duration::from_secs(10); /// The cross-relay drill's link: the drills' impairment behind a bottleneck a -/// burst overruns, so its tail waits in the queue or is dropped from it. +/// burst overruns, so its tail waits in the queue or is dropped from it. It is +/// slower than [`credit`]'s receive window lets a sender go (about 1 KiB per +/// round trip, some 160 kbit/s), or the window would bind first and the queue +/// would never fill. fn bursty() -> moq_shaper::Profile { moq_shaper::Profile { rate: Some(moq_shaper::Rate { - bits_per_second: 1_000_000, + bits_per_second: 100_000, burst: 1500, queue: Duration::from_millis(50), }), @@ -816,7 +829,8 @@ async fn bursts_cross_a_cluster(lane: Lane) { let mut config = relay_config(None); config.quic = credit(lane, config.quic); let origin = RelayHost::start(config).await; - let peer = Path::shaped(lane, &origin, bursty()).await; + let seed = seed(lane); + let peer = Path::shaped(lane, &origin, bursty(), seed).await; // The edge relay dials the origin as a cluster peer, through its own path. let mut peer_url = peer.url.clone(); @@ -829,8 +843,8 @@ async fn bursts_cross_a_cluster(lane: Lane) { config.cluster.connect = vec![moq_relay::cluster::Peer::new(peer_url.to_string())]; let edge = RelayHost::start(config).await; - let publish_path = Path::shaped(lane, &origin, bursty()).await; - let subscribe_path = Path::shaped(lane, &edge, bursty()).await; + let publish_path = Path::shaped(lane, &origin, bursty(), seed.wrapping_add(1)).await; + let subscribe_path = Path::shaped(lane, &edge, bursty(), seed.wrapping_add(2)).await; let publisher = moq_tokio::origin::spawn(); let broadcast = publisher.create_broadcast("live").expect("create broadcast"); @@ -903,8 +917,11 @@ async fn bursts_cross_a_cluster(lane: Lane) { }); }; + // A margin past each settle task's own [`SETTLE`] timeout, so a group that + // times out is reported by name rather than as a generic stall. + let step_timeout = SETTLE + Duration::from_secs(1); while (outcomes.len() as u64) < total { - let step = tokio::time::timeout(SETTLE, async { + let step = tokio::time::timeout(step_timeout, async { tokio::select! { group = reader.groups.recv_group() => Step::Group(Box::new(group)), Some(settled) = pending.join_next() => { @@ -917,7 +934,7 @@ async fn bursts_cross_a_cluster(lane: Lane) { .unwrap_or_else(|_| { let missing: Vec = (0..total).filter(|sequence| !outcomes.contains_key(sequence)).collect(); panic!( - "stalled: nothing arrived within {SETTLE:?}, {} of {total} groups missing: {missing:?}", + "stalled: nothing arrived within {step_timeout:?}, {} of {total} groups missing: {missing:?}", missing.len() ) }); @@ -1001,9 +1018,14 @@ async fn bursts_cross_a_cluster(lane: Lane) { drop(publish_session); drop(edge); drop(origin); - peer.verify(); - publish_path.verify(); - subscribe_path.verify(); + let overflowed: u64 = [peer, publish_path, subscribe_path] + .into_iter() + .filter_map(Path::verify) + .map(|stats| stats.up.overflowed + stats.down.overflowed) + .sum(); + if let Lane::Impaired = lane { + assert!(overflowed > 0, "no burst overran a bottleneck, so none was exercised"); + } } lanes!(bursts_cross_a_cluster); diff --git a/test/drill/README.md b/test/drill/README.md index 08c38acf75..71b7157c8e 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -75,11 +75,13 @@ forwards each datagram, both ways, after applying the same profile: `bursts_cross_a_cluster` shapes all three hops, the publisher's, the peer link between the relays, and the subscriber's, each with that profile behind a -1 Mbit/s bottleneck with a 50ms queue, which a burst overruns. Its impaired lane -also grants less QUIC credit than a burst needs at every endpoint: 16 concurrent -streams (each group is a stream, each FETCH another) and receive windows of a -few groups, so senders wait on `MAX_STREAMS` and `MAX_DATA` that the path can -lose. Its loopback lane keeps the defaults. +100 kbit/s bottleneck with a 50ms queue, which a burst overruns. Its impaired +lane also grants less QUIC credit than a burst needs at every endpoint: 16 +concurrent streams (each group is a stream, each FETCH another) and receive +windows of a few groups, so senders wait on `MAX_STREAMS` and `MAX_DATA` that +the path can lose. The bottleneck sits below the rate those windows allow, so +both bind, and the drill fails unless some burst overflowed a queue. Its +loopback lane keeps the defaults. It is a datagram relay, not an HTTP or TCP proxy, neither of which can impair QUIC. It needs no capabilities, works the same on macOS and Linux, and touches @@ -88,9 +90,10 @@ The shaper runs on the test's runtime rather than the relay's, so killing the relay leaves the path up, the way a network outlives the server behind it. Every decision the shaper makes comes from one seed. Each run picks a fresh one -and prints it, with the profile, as `impaired: MOQ_SHAPER_SEED=...`; setting that -variable replays the same decisions. Kernel scheduling still varies delivery -timing, so the seed makes the decisions reproducible, not the clock. Each client +and prints it as `impaired: MOQ_SHAPER_SEED=...`; setting that variable replays +the same decisions. A drill with several paths seeds each one off that base. +Kernel scheduling still varies delivery timing, so the seed makes the decisions +reproducible, not the clock. Each client draws from its own stream, numbered in the order clients first send, so concurrent clients that race to connect can swap streams between runs. From aaba76b3289e665804736d3b2c62097adc3542bd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 02:40:13 -0700 Subject: [PATCH 5/5] test(relay): grade only the cluster drill's burst overflow Snapshot each shaper's overflow once connected and subscribed, so a handshake's padded Initial flight cannot satisfy the bottleneck check. The 50ms queue stays: the 1 KiB window keeps burst datagrams small, so it both queues and drops them, where 100ms and longer absorbed every burst. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/tests/drills.rs | 47 +++++++++++++++++++++++------------- 1 file changed, 30 insertions(+), 17 deletions(-) diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 3d66ef96e0..36532c96bf 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -131,12 +131,20 @@ impl Path { /// Fail the drill unless the impairment it ran under actually acted. /// /// A profile that silently did nothing would turn this lane into a second - /// loopback run that passes for free. Returns what the shaper did, if any. - fn verify(mut self) -> Option { - let shaper = self.shaper.take()?; - let stats = shaper.verify().unwrap_or_else(|err| panic!("impaired: {err:#}")); - println!("impaired: {stats}"); - Some(stats) + /// loopback run that passes for free. + fn verify(mut self) { + if let Some(shaper) = self.shaper.take() { + let stats = shaper.verify().unwrap_or_else(|err| panic!("impaired: {err:#}")); + println!("impaired: {stats}"); + } + } + + /// Datagrams the shaper has dropped from a full rate-limit queue so far, both ways. + fn overflowed(&self) -> u64 { + self.shaper.as_ref().map_or(0, |shaper| { + let stats = shaper.stats(); + stats.up.overflowed + stats.down.overflowed + }) } } @@ -747,7 +755,8 @@ const SETTLE: Duration = Duration::from_secs(10); /// burst overruns, so its tail waits in the queue or is dropped from it. It is /// slower than [`credit`]'s receive window lets a sender go (about 1 KiB per /// round trip, some 160 kbit/s), or the window would bind first and the queue -/// would never fill. +/// would never fill. The window keeps a burst's datagrams small, so the 50ms +/// queue holds one or two of them before it drops the rest. fn bursty() -> moq_shaper::Profile { moq_shaper::Profile { rate: Some(moq_shaper::Rate { @@ -881,6 +890,10 @@ async fn bursts_cross_a_cluster(lane: Lane) { // Subscribed before the first burst, as an app watching a control channel is. let mut reader = subscribe(&subscribed, "live").await; + // What connecting and subscribing overflowed, so only the bursts are graded. + let paths = [&peer, &publish_path, &subscribe_path]; + let setup_overflowed: u64 = paths.iter().map(|path| path.overflowed()).sum(); + let total = BURSTS * BURST_GROUPS; let mut written = Vec::new(); let burst = |track: &mut moq_net::track::Producer, written: &mut Vec| { @@ -985,16 +998,21 @@ async fn bursts_cross_a_cluster(lane: Lane) { } // Fault activation: the subscription skipped or reset groups, so the FETCH - // recovery below had something to recover. + // recovery below had something to recover, and in the impaired lane the + // bursts overflowed a bottleneck. let gaps = outcomes .values() .filter(|outcome| !matches!(outcome, Outcome::Live)) .count(); + let overflowed = paths.iter().map(|path| path.overflowed()).sum::() - setup_overflowed; println!( - "fault activated: {gaps} of {total} groups missed the live subscription (reset: {resets:?}), slowest group {} after {:?}", + "fault activated: {gaps} of {total} groups missed the live subscription (reset: {resets:?}), {overflowed} datagrams overflowed a bottleneck during the bursts, slowest group {} after {:?}", slowest.1, slowest.0 ); assert!(gaps > 0, "every group came live, so no FETCH recovery was exercised"); + if let Lane::Impaired = lane { + assert!(overflowed > 0, "no burst overran a bottleneck, so none was exercised"); + } let lost: Vec<_> = outcomes .iter() @@ -1018,14 +1036,9 @@ async fn bursts_cross_a_cluster(lane: Lane) { drop(publish_session); drop(edge); drop(origin); - let overflowed: u64 = [peer, publish_path, subscribe_path] - .into_iter() - .filter_map(Path::verify) - .map(|stats| stats.up.overflowed + stats.down.overflowed) - .sum(); - if let Lane::Impaired = lane { - assert!(overflowed > 0, "no burst overran a bottleneck, so none was exercised"); - } + peer.verify(); + publish_path.verify(); + subscribe_path.verify(); } lanes!(bursts_cross_a_cluster);