From 94c9c63958e4fbe0504a117b62062a6df4f57ba7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:19:01 -0700 Subject: [PATCH 1/5] quest(cross-relay-flap-drill): claim From 68c943abdaa1e0462788f36355f8a256a773094b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:57:42 -0700 Subject: [PATCH 2/5] test(relay): run the cluster burst drill across a flapping peer link bursts_cross_a_flapping_peer cuts the peer link halfway through the second burst, restores it once the edge withdraws the route, and requires every group to still arrive within the drill's 10s bound. Reads and FETCHes the flap failed are claimed again on the restored route; anything else is graded as strictly as the steady drill. moq-shaper gains Shaper::cut, an RAII outage that drops every datagram both ways. The new peer-link-never-redialed mutation proves the drill fails when the edge stops redialing a lost peer. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/cross-relay-flap-drill.md | 15 -- rs/moq-relay/tests/drills.rs | 254 ++++++++++++++---- rs/moq-shaper/README.md | 4 +- rs/moq-shaper/src/lib.rs | 71 ++++- test/drill/README.md | 14 +- .../mutations/peer-link-never-redialed.patch | 24 ++ 7 files changed, 306 insertions(+), 77 deletions(-) delete mode 100644 quest/m1/cross-relay-flap-drill.md create mode 100644 test/drill/mutations/peer-link-never-redialed.patch diff --git a/quest/m1/README.md b/quest/m1/README.md index 21d2307fea..134a0091f1 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -56,7 +56,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 -- [Flapping peer drill](/quest/m1/cross-relay-flap-drill.md) - the cluster burst drill survives a flapping peer link, the regression for 0.15.6 route-flap drops - [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-flap-drill.md b/quest/m1/cross-relay-flap-drill.md deleted file mode 100644 index 500cfbafe5..0000000000 --- a/quest/m1/cross-relay-flap-drill.md +++ /dev/null @@ -1,15 +0,0 @@ -# [S] Cluster drill with a flapping peer link - -## Goal - -The two-relay burst drill also runs with the peer link flapping mid-burst, -and every group still arrives or fails by name within its bound. This is the -regression for the route-flap drops the 0.15.6 tree showed in #4349, which -current `main` no longer reproduces in a mock. - -## Plan - -Extend `bursts_cross_a_cluster` (#4920, `rs/moq-relay/tests/drills.rs`) rather -than adding a harness: flap the shaped peer link during a burst, keep strict -grading, and add a drill mutation proving the flap is graded. A moq-transport -peer link variant is optional; add it only if it is cheap here. diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index a51354559c..8e95a60861 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -104,13 +104,18 @@ impl Path { /// [`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 { + match lane { + Lane::Loopback => Self { url: relay.url(), shaper: None, - }; - }; + }, + Lane::Impaired => Self::through(relay, profile, seed).await, + } + } + /// A shaper in front of `relay` in any lane, treating the path to `profile` + /// both ways, seeded by `seed`. + async fn through(relay: &RelayHost, profile: moq_shaper::Profile, seed: u64) -> Self { 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(), @@ -120,7 +125,7 @@ impl Path { }) .await .expect("start the shaper"); - println!("impaired: seed {seed}, profile {:?}", shaper.config().up); + println!("shaper: seed {seed}, profile {:?}", shaper.config().up); Self { url: format!("https://{}/drill", shaper.addr()).parse().unwrap(), @@ -134,8 +139,8 @@ impl Path { /// 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}"); + let stats = shaper.verify().unwrap_or_else(|err| panic!("shaper: {err:#}")); + println!("shaper: {stats}"); } } @@ -167,7 +172,7 @@ impl Drop for Path { fn drop(&mut self) { // A drill that failed before `verify` still reports what the path did. if let Some(shaper) = &self.shaper { - println!("impaired: {}", shaper.stats()); + println!("shaper: {}", shaper.stats()); } } } @@ -285,19 +290,25 @@ fn quic() -> moq_tokio::quic::Config { /// [`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(); + let mut config = dial(); config.url = Some(url.clone()); + config.init(quic).expect("client init") +} + +/// How a drill's clients, and a relay dialing a peer, connect: trusting the +/// generated certificate, over QUIC only, from an ephemeral loopback port. +fn dial() -> moq_tokio::connect::Config { + let mut config = moq_tokio::connect::Config::default(); 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. + // Fast enough to keep a relay bounce or a link flap inside the drill's + // budget, paced enough that the loop is still a backoff. config.backoff.initial = Duration::from_millis(50); config.backoff.max = Duration::from_millis(200); config.backoff.timeout = Duration::from_secs(5); - - config.init(quic).expect("client init") + config } /// Append one finished group carrying `payload`. @@ -800,8 +811,10 @@ enum Outcome { enum Step { /// The live subscription's next group. Group(Box>>), - /// A group read or FETCH finished. - Settled(u64, Outcome), + /// A group read or FETCH, made through the numbered subscription, finished. + Settled(u64, Outcome, u64), + /// A route to this path started (`true`) or ended (`false`). + Route(String, bool), } /// Read the one frame of group `sequence`, checking it is the payload written there. @@ -835,20 +848,84 @@ fn payload(sequence: u64) -> Vec { /// 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) { + cross_cluster(lane, PeerLink::Steady).await +} + +lanes!(bursts_cross_a_cluster); + +/// Drill: [`bursts_cross_a_cluster`], with the peer link flapping mid-burst. +/// +/// The link between the relays is cut halfway through writing the second burst +/// and stays down until the subscriber sees the edge withdraw the route, then +/// comes back and the edge redials. That is a route flap: the regression for the +/// groups 0.15.6 dropped across one (#4349). The subscriber re-subscribes once +/// the route returns, and every read or FETCH the flap failed is claimed again +/// through it, the way an app recovering a control channel would. +/// +/// The grading stays as strict: the publisher keeps every group, so each still +/// arrives within [`SETTLE`] of being written or fails the drill by name. Only a +/// failure the flap caused is retried; a read or FETCH on the dead route that +/// never resolves, or a withdrawal while the link was up, still fails it. +async fn bursts_cross_a_flapping_peer(lane: Lane) { + cross_cluster(lane, PeerLink::Flapping).await +} + +lanes!(bursts_cross_a_flapping_peer); + +/// Whether the cross-relay drill's peer link stays up or flaps mid-burst. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum PeerLink { + Steady, + Flapping, +} + +/// The burst a flapping peer link is cut in the middle of: after one that +/// crossed the link, and before the bursts that prove the restored route carries on. +const FLAP_BURST: u64 = 1; + +/// Where a flapping peer link stands. +enum Flap { + /// Not cut, or never will be. + Up, + /// Cut at `at`, with the route over it still announced. + Cut { + outage: moq_shaper::Outage, + at: tokio::time::Instant, + }, + /// The edge withdrew the route after the link was down for `cut`. The link + /// is back and the edge redials. + Withdrawn { cut: Duration }, + /// The route came back and the subscriber re-subscribed through it. + Restored { cut: Duration }, +} + +impl Flap { + /// Whether the route is down or going, so a read or FETCH through it may fail. + fn flapping(&self) -> bool { + matches!(self, Flap::Cut { .. } | Flap::Withdrawn { .. }) + } +} + +async fn cross_cluster(lane: Lane, link: PeerLink) { + // Both relays get the drills' QUIC tuning, so a cut peer link idles out + // within the drill's budget rather than after the 10s default. let mut config = relay_config(None); - config.quic = credit(lane, config.quic); + config.quic = credit(lane, quic()); let origin = RelayHost::start(config).await; let seed = seed(lane); - let peer = Path::shaped(lane, &origin, bursty(), seed).await; + let peer = match (link, lane) { + (PeerLink::Steady, _) => Path::shaped(lane, &origin, bursty(), seed).await, + // Cutting the link takes a shaper on it, so loopback gets one that forwards untouched. + (PeerLink::Flapping, Lane::Loopback) => Path::through(&origin, Default::default(), seed).await, + (PeerLink::Flapping, Lane::Impaired) => Path::through(&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(); 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.quic = credit(lane, quic()); + config.connect = dial(); config.cluster.connect = vec![moq_relay::cluster::Peer::new(peer_url.to_string())]; let edge = RelayHost::start(config).await; @@ -874,6 +951,7 @@ async fn bursts_cross_a_cluster(lane: Lane) { let subscriber = moq_tokio::origin::spawn(); let subscribed = subscriber.consume(); + let mut announced = subscribed.announced(); let url = subscribe_path.url.clone(); let subscribe_session = tokio::time::timeout( TIMEOUT, @@ -896,39 +974,54 @@ async fn bursts_cross_a_cluster(lane: Lane) { 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 mut flap = Flap::Up; + // Write the next burst, cutting a flapping link halfway through the flap burst. + let burst = |track: &mut moq_net::track::Producer, written: &mut Vec, flap: &mut Flap| { + let index = written.len() as u64 / BURST_GROUPS; + for i in 0..BURST_GROUPS { + if link == PeerLink::Flapping && index == FLAP_BURST && i == BURST_GROUPS / 2 { + let outage = peer.shaper.as_ref().expect("a flapping link has a shaper").cut(); + *flap = Flap::Cut { + outage, + at: tokio::time::Instant::now(), + }; + } let sequence = written.len() as u64; write_group(track, &payload(sequence)); written.push(tokio::time::Instant::now()); } }; - burst(&mut track, &mut written); + burst(&mut track, &mut written, &mut flap); // 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. + // FETCHed in turn. Each read and FETCH is tagged with the subscription it went + // through, counted from zero, so one the flap took down can be told apart. 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 flapped = 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 mut generation = 0u64; + let mut subscription_up = true; + let fetch = + |pending: &mut tokio::task::JoinSet<_>, track: &moq_net::track::Consumer, sequence: u64, generation: 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, generation) }); - let outcome = match fetched.await { - Ok(Ok(())) => Outcome::Fetched, - Ok(Err(err)) => Outcome::Failed(err), - Err(_) => Outcome::Unanswered, - }; - (sequence, outcome) - }); - }; + }; // 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. @@ -936,44 +1029,89 @@ async fn bursts_cross_a_cluster(lane: Lane) { while (outcomes.len() as u64) < total { let step = tokio::time::timeout(step_timeout, async { tokio::select! { - group = reader.groups.recv_group() => Step::Group(Box::new(group)), + group = reader.groups.recv_group(), if subscription_up => Step::Group(Box::new(group)), Some(settled) = pending.join_next() => { - let (sequence, outcome) = settled.expect("settle task panicked"); - Step::Settled(sequence, outcome) + let (sequence, outcome, generation) = settled.expect("settle task panicked"); + Step::Settled(sequence, outcome, generation) } + Some((update, active)) = next_update(&mut announced) => Step::Route(update.prefix.to_string(), active), } }) .await .unwrap_or_else(|_| { let missing: Vec = (0..total).filter(|sequence| !outcomes.contains_key(sequence)).collect(); + let link = match flap { + Flap::Cut { .. } => " with the peer link cut", + Flap::Withdrawn { .. } => " with the route withdrawn", + Flap::Up | Flap::Restored { .. } => "", + }; panic!( - "stalled: nothing arrived within {step_timeout:?}, {} of {total} groups missing: {missing:?}", + "stalled{link}: nothing arrived within {step_timeout:?}, {} of {total} groups missing: {missing:?}", missing.len() ) }); match step { - Step::Settled(sequence, Outcome::Failed(err)) if live.remove(&sequence) => { + // The flap took this read or FETCH down with the route: claim the group + // again through the restored one, now if it is back, or as a gap once it is. + Step::Settled(sequence, Outcome::Failed(err), tagged) if tagged < generation || flap.flapping() => { + live.remove(&sequence); + flapped.push((sequence, err)); + match flap.flapping() { + true => { + claimed.remove(&sequence); + } + false => fetch(&mut pending, &reader.track, sequence, generation), + } + } + Step::Settled(sequence, Outcome::Failed(err), _) if live.remove(&sequence) => { resets.push((sequence, err)); - fetch(&mut pending, &reader.track, sequence); + fetch(&mut pending, &reader.track, sequence, generation); } - Step::Settled(sequence, outcome) => { + Step::Settled(sequence, outcome, _) => { live.remove(&sequence); let late = written[sequence as usize].elapsed(); slowest = slowest.max((late, sequence)); outcomes.insert(sequence, outcome); } + Step::Route(path, active) => { + assert_eq!(path, "live", "announced a broadcast nobody published"); + flap = match (active, flap) { + // The edge noticed the cut and withdrew the route: restore the + // link, so the edge's redial gets through. + (false, Flap::Cut { outage, at }) => { + drop(outage); + Flap::Withdrawn { cut: at.elapsed() } + } + (false, _) => panic!("the route was withdrawn while the peer link was up"), + (true, Flap::Withdrawn { cut }) => { + reader = subscribe(&subscribed, "live").await; + generation += 1; + subscription_up = true; + Flap::Restored { cut } + } + // The first announcement, or an update to a standing route. + (true, flap) => flap, + }; + } Step::Group(group) => { - let group = (*group) - .unwrap_or_else(|err| panic!("the subscription aborted: {err}")) - .unwrap_or_else(|| panic!("the subscription finished")); + let group = match *group { + Ok(Some(group)) => group, + // The route went, and the subscription with it, until it returns. + Err(_) if flap.flapping() => { + subscription_up = false; + continue; + } + Err(err) => panic!("the subscription aborted: {err}"), + Ok(None) => 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); + fetch(&mut pending, &reader.track, gap, generation); } } if claimed.insert(sequence) { @@ -984,14 +1122,14 @@ async fn bursts_cross_a_cluster(lane: Lane) { Ok(Err(err)) => Outcome::Failed(err), Err(_) => Outcome::Stalled, }; - (sequence, outcome) + (sequence, outcome, generation) }); } // 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); + burst(&mut track, &mut written, &mut flap); } } } @@ -1013,6 +1151,16 @@ async fn bursts_cross_a_cluster(lane: Lane) { if let Lane::Impaired = lane { assert!(overflowed > 0, "no burst overran a bottleneck, so none was exercised"); } + match (link, flap) { + (PeerLink::Steady, Flap::Up) => {} + (PeerLink::Flapping, Flap::Restored { cut }) => println!( + "fault activated: the peer link was down {cut:?} before the edge withdrew the route, which came back; the flap failed {} reads and FETCHes, each claimed again: {flapped:?}", + flapped.len() + ), + (PeerLink::Flapping, Flap::Up) => panic!("the peer link was never cut"), + (PeerLink::Flapping, _) => panic!("the route never came back after the peer link flapped"), + (PeerLink::Steady, _) => unreachable!("a steady link is never cut"), + } let lost: Vec<_> = outcomes .iter() @@ -1041,8 +1189,6 @@ async fn bursts_cross_a_cluster(lane: Lane) { subscribe_path.verify(); } -lanes!(bursts_cross_a_cluster); - /// Negative control: with nothing publishing, the drills' delivery assertions /// have to fail. /// diff --git a/rs/moq-shaper/README.md b/rs/moq-shaper/README.md index 1ccad1cea2..9367721e4c 100644 --- a/rs/moq-shaper/README.md +++ b/rs/moq-shaper/README.md @@ -45,7 +45,9 @@ As a library, `Shaper::bind` takes a `Config`: where to listen and forward, the per direction. It also takes a `Setup`, which is a `Config` plus the opt-in options below, each off by default. `Shaper::verify` fails when the shaper stopped forwarding, when an impairment the profile configures never acted and the traffic makes that silence implausible, or when a direction carried -traffic but no datagram saw one of its phases, before the first step or after one. The relay's drills +traffic but no datagram saw one of its phases, before the first step or after one. `Shaper::cut` takes +the path down both ways until the `Outage` it returns drops, so a test can sever a link mid-run and +restore it when it chooses rather than on a timer. The relay's drills (`rs/moq-relay/tests/drills.rs`, described in `test/drill/README.md`) run every scenario through it. ## What a seed does and does not fix diff --git a/rs/moq-shaper/src/lib.rs b/rs/moq-shaper/src/lib.rs index 66e359f033..77490623c2 100644 --- a/rs/moq-shaper/src/lib.rs +++ b/rs/moq-shaper/src/lib.rs @@ -411,6 +411,8 @@ pub struct Shaper { tally: Arc<[Tally; 2]>, /// Why forwarding stopped, if it did. failed: Arc>, + /// How many [`Outage`]s are holding the path down. + cuts: Arc, task: tokio::task::JoinHandle<()>, } @@ -429,12 +431,14 @@ impl Shaper { let tally = Arc::new([Tally::default(), Tally::default()]); let failed = Arc::new(OnceLock::new()); + let cuts = Arc::new(AtomicU64::new(0)); let task = tokio::spawn({ let setup = setup.clone(); let tally = tally.clone(); let failed = failed.clone(); + let cuts = cuts.clone(); async move { - if let Err(err) = run(Arc::new(listen), setup, tally).await { + if let Err(err) = run(Arc::new(listen), setup, tally, cuts).await { let _ = failed.set(format!("{err:#}")); } } @@ -445,10 +449,23 @@ impl Shaper { setup, tally, failed, + cuts, task, }) } + /// Take the path down both ways until the returned [`Outage`] drops. + /// + /// Every datagram is dropped untreated meanwhile, the way a severed link + /// loses it, so it counts toward no impairment. Datagrams already treated + /// still leave on time; only what arrives during the outage is lost. + pub fn cut(&self) -> Outage { + self.cuts.fetch_add(1, Ordering::Relaxed); + Outage { + cuts: self.cuts.clone(), + } + } + /// The address clients send to. pub fn addr(&self) -> SocketAddr { self.addr @@ -529,6 +546,23 @@ impl Drop for Shaper { } } +/// A path taken down by [`Shaper::cut`]; dropping it restores the path. +#[must_use = "dropping the outage restores the path at once"] +pub struct Outage { + cuts: Arc, +} + +impl Drop for Outage { + fn drop(&mut self) { + self.cuts.fetch_sub(1, Ordering::Relaxed); + } +} + +/// Whether an [`Outage`] is holding the path down. +fn is_cut(cuts: &AtomicU64) -> bool { + cuts.load(Ordering::Relaxed) > 0 +} + const UP: usize = 0; const DOWN: usize = 1; @@ -568,7 +602,7 @@ fn bump(counter: &AtomicU64) { /// A flow is a socket of its own toward the target, so the target sees one /// address per client just as it would without the shaper in the way. Each /// flow takes a link of its own each way, unless the path is shared. -async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>) -> anyhow::Result<()> { +async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>, cuts: Arc) -> anyhow::Result<()> { let config = &setup.config; let mut flows = HashMap::::new(); let mut tasks = JoinSet::new(); @@ -591,6 +625,9 @@ async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>) -> an } }; + if is_cut(&cuts) { + continue; + } let now = Instant::now(); let number = flows.len() as u64; @@ -610,6 +647,7 @@ async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>) -> an listen.clone(), from, tally.clone(), + cuts.clone(), )); entry.insert(Flow { upstream, up }) @@ -676,12 +714,13 @@ async fn reply( listen: Arc, client: SocketAddr, tally: Arc<[Tally; 2]>, + cuts: Arc, ) -> anyhow::Result<()> { let mut buf = vec![0u8; u16::MAX as usize]; loop { let (size, from) = socket.recv_from(&mut buf).await.context("receive from the target")?; // Anything else reaching this ephemeral port is not part of the path. - if from != target { + if from != target || is_cut(&cuts) { continue; } let datagram = buf[..size].to_vec(); @@ -1012,6 +1051,32 @@ mod tests { assert_eq!(stats.down.packets, 50); } + #[tokio::test(start_paused = true)] + async fn a_cut_path_drops_both_ways_until_restored() { + // The delay holds the datagram already treated on its way up, so the + // echo comes back while the path is cut. + let delayed = Profile { + delay: Duration::from_millis(10), + ..Default::default() + }; + let (shaper, client) = setup(1, delayed, Profile::default()).await; + client.send(b"held").await.unwrap(); + while shaper.stats().up.packets == 0 { + tokio::time::sleep(Duration::from_millis(1)).await; + } + + let outage = shaper.cut(); + let got = round_trip(&client, 10).await; + assert!(got.is_empty(), "a cut path delivered {got:?}"); + let stats = shaper.stats(); + assert_eq!(stats.up.packets, 1, "a cut path treated what arrived up: {stats}"); + assert_eq!(stats.down.packets, 0, "a cut path treated the echo: {stats}"); + + drop(outage); + let got = round_trip(&client, 10).await; + assert_eq!(got, (0..10).collect::>(), "the restored path lost datagrams"); + } + #[tokio::test(start_paused = true)] async fn an_undelayed_datagram_waits_for_no_timer() { let (_shaper, client) = setup(1, Profile::default(), Profile::default()).await; diff --git a/test/drill/README.md b/test/drill/README.md index 71b7157c8e..3b4eec30bb 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -4,7 +4,7 @@ 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, a publisher coming back under a name it already used, and bursts of small groups crossing a two-relay -cluster. +cluster, with its peer link steady or flapping. The drills themselves are ordinary Rust integration tests in `rs/moq-relay/tests/drills.rs`, so `just check` already runs them whenever @@ -30,6 +30,7 @@ checks that what the fault owned was released. | `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 | +| `bursts_cross_a_flapping_peer` | The same bursts, with the peer link cut halfway through writing the second burst and restored once the edge withdraws the route | how long the link was down before the withdrawal, and the reads and FETCHes the flap failed | the route comes back and every group still arrives within 10s of being written; only a failure the flap caused is retried, on the restored route, so a hang on the dead one or a withdrawal with the link up still fails | | `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 @@ -43,7 +44,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. Four drills prove things by +The negative control is what keeps the rest honest. Five drills prove things by reading a frame; if the harness could report success without data moving, they would all pass for free. @@ -83,6 +84,12 @@ 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. +`bursts_cross_a_flapping_peer` cuts the peer link with `Shaper::cut`, which +drops every datagram both ways until it is restored, so its loopback lane puts a +shaper on that link too, one that forwards untouched. The cut lasts until the +subscriber sees the route withdrawn, which the relays' two-second idle timeout +bounds, rather than for a fixed time. + 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. @@ -147,6 +154,7 @@ and grades the same body. | `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` | +| `peer-link-never-redialed` | redialing a cluster peer after its link is lost | `bursts_cross_a_flapping_peer::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 @@ -169,7 +177,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) | +| bursts across a cluster, steady or flapping | `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/peer-link-never-redialed.patch b/test/drill/mutations/peer-link-never-redialed.patch new file mode 100644 index 0000000000..874e580a4b --- /dev/null +++ b/test/drill/mutations/peer-link-never-redialed.patch @@ -0,0 +1,24 @@ +# drill: bursts_cross_a_flapping_peer::loopback +# expect: stalled with the route withdrawn +# +# Removes redialing a cluster peer whose link was lost. The edge stops driving +# its connection to the origin once the session drops, so a link flap withdraws +# the route for good: the groups caught in the flap and every burst after it +# never arrive. A steady link never drops, so only the flap drill depends on it. +diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs +index 6e3e22cbc..7bc6d15cf 100644 +--- a/rs/moq-relay/src/cluster.rs ++++ b/rs/moq-relay/src/cluster.rs +@@ -1728,7 +1728,11 @@ impl Cluster { + moq_tokio::Status::Connected if connection.is_none() => { + connection = Some(self.nodes.connect_outbound(id, redacted.to_string())); + } +- moq_tokio::Status::Disconnected => connection = None, ++ // MUTATION: a lost peer link is never redialed. ++ moq_tokio::Status::Disconnected => { ++ drop(reconnect); ++ return std::future::pending().await; ++ } + moq_tokio::Status::Migrating => {} + _ => {} + } From 2b1211831ee90c81cbef709fe3031042eabc0a20 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 11:24:26 -0700 Subject: [PATCH 3/5] test(relay): cut the flap before its burst and redial the peer forever A burst is written without yielding, so the cut lands before any of it leaves; say so and cut there. The peer link retries without a deadline, so a slow impaired handshake can't pass for the never-redialed regression. Record the IETF peer FETCH 'old' bug as a quest. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 + quest/m1/ietf-peer-fetch-old.md | 19 +++++++++++++++++ rs/moq-relay/tests/drills.rs | 37 +++++++++++++++++++-------------- test/drill/README.md | 2 +- 4 files changed, 42 insertions(+), 17 deletions(-) create mode 100644 quest/m1/ietf-peer-fetch-old.md diff --git a/quest/m1/README.md b/quest/m1/README.md index a2df967c92..b3421e289b 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -54,6 +54,7 @@ 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 +- [IETF peer FETCH refused old](/quest/m1/ietf-peer-fetch-old.md) - a cross-relay FETCH over a moq-transport peer link returns groups the origin still holds - [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/ietf-peer-fetch-old.md b/quest/m1/ietf-peer-fetch-old.md new file mode 100644 index 0000000000..776d1d5a74 --- /dev/null +++ b/quest/m1/ietf-peer-fetch-old.md @@ -0,0 +1,19 @@ +# [S] IETF peer FETCH refused old + +## Goal + +A cross-relay FETCH over a `moq-transport` peer link returns every group the origin still holds, instead of refusing some as `old`. +The cluster burst drill then runs a third peer link variant over `moq-transport-19`, and its impaired lane passes. + +## Plan + +Repro: in `rs/moq-relay/tests/drills.rs::cross_cluster`, pin the edge's peer link with `config.connect.version = ["moq-transport-19"]`. +The steady impaired drill then failed in 5 of 5 runs with FETCHes refused `old` for groups the origin still held, e.g. `groups lost: [(5, Failed("old")), (37, Failed("old")), (41, Failed("old"))]`. +Seeds: 11977074354257116273, 6755536760138253279, 17984197590923068922, 10974424043511566981. +The loopback lane and the flapping drill passed. + +Find where the IETF FETCH path decides a group is `old`, fix it at the source, then add the IETF peer link as a `PeerLink` variant. + +## Related + +- [Cross-relay bursts re-run](/quest/m1/cross-relay-bursts.md) - the #4349 `Stream(Old)` stalls, not reproduced over a lite peer link diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 8e95a60861..62e4345e66 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -853,10 +853,11 @@ async fn bursts_cross_a_cluster(lane: Lane) { lanes!(bursts_cross_a_cluster); -/// Drill: [`bursts_cross_a_cluster`], with the peer link flapping mid-burst. +/// Drill: [`bursts_cross_a_cluster`], with the peer link flapping mid-recovery. /// -/// The link between the relays is cut halfway through writing the second burst -/// and stays down until the subscriber sees the edge withdraw the route, then +/// The link between the relays is cut as the second burst is written, once the +/// first has started arriving, so the first burst's gap FETCHes are in flight and +/// the second is stranded at the origin. It stays down until the subscriber sees the edge withdraw the route, then /// comes back and the edge redials. That is a route flap: the regression for the /// groups 0.15.6 dropped across one (#4349). The subscriber re-subscribes once /// the route returns, and every read or FETCH the flap failed is claimed again @@ -872,15 +873,15 @@ async fn bursts_cross_a_flapping_peer(lane: Lane) { lanes!(bursts_cross_a_flapping_peer); -/// Whether the cross-relay drill's peer link stays up or flaps mid-burst. +/// Whether the cross-relay drill's peer link stays up or flaps. #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum PeerLink { Steady, Flapping, } -/// The burst a flapping peer link is cut in the middle of: after one that -/// crossed the link, and before the bursts that prove the restored route carries on. +/// The burst a flapping peer link is cut just before: after one that started +/// crossing the link, and before the bursts that prove the restored route carries on. const FLAP_BURST: u64 = 1; /// Where a flapping peer link stands. @@ -926,6 +927,9 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { let mut config = relay_config(None); config.quic = credit(lane, quic()); config.connect = dial(); + // Retry forever: a give-up after a slow impaired handshake would look exactly + // like the never-redialed regression, and a stall still fails the drill. + config.connect.backoff.timeout = Duration::ZERO; config.cluster.connect = vec![moq_relay::cluster::Peer::new(peer_url.to_string())]; let edge = RelayHost::start(config).await; @@ -975,17 +979,18 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { let total = BURSTS * BURST_GROUPS; let mut written = Vec::new(); let mut flap = Flap::Up; - // Write the next burst, cutting a flapping link halfway through the flap burst. + // Write the next burst, cutting a flapping link just before the flap burst. + // A burst is written without yielding, so a cut partway through it would + // still land before any of it left the publisher. let burst = |track: &mut moq_net::track::Producer, written: &mut Vec, flap: &mut Flap| { - let index = written.len() as u64 / BURST_GROUPS; - for i in 0..BURST_GROUPS { - if link == PeerLink::Flapping && index == FLAP_BURST && i == BURST_GROUPS / 2 { - let outage = peer.shaper.as_ref().expect("a flapping link has a shaper").cut(); - *flap = Flap::Cut { - outage, - at: tokio::time::Instant::now(), - }; - } + if link == PeerLink::Flapping && written.len() as u64 == FLAP_BURST * BURST_GROUPS { + let outage = peer.shaper.as_ref().expect("a flapping link has a shaper").cut(); + *flap = Flap::Cut { + outage, + at: tokio::time::Instant::now(), + }; + } + for _ in 0..BURST_GROUPS { let sequence = written.len() as u64; write_group(track, &payload(sequence)); written.push(tokio::time::Instant::now()); diff --git a/test/drill/README.md b/test/drill/README.md index 3b4eec30bb..b12797167a 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -30,7 +30,7 @@ checks that what the fault owned was released. | `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 | -| `bursts_cross_a_flapping_peer` | The same bursts, with the peer link cut halfway through writing the second burst and restored once the edge withdraws the route | how long the link was down before the withdrawal, and the reads and FETCHes the flap failed | the route comes back and every group still arrives within 10s of being written; only a failure the flap caused is retried, on the restored route, so a hang on the dead one or a withdrawal with the link up still fails | +| `bursts_cross_a_flapping_peer` | The same bursts, with the peer link cut as the second burst is written and restored once the edge withdraws the route | how long the link was down before the withdrawal, and the reads and FETCHes the flap failed | the route comes back and every group still arrives within 10s of being written; only a failure the flap caused is retried, on the restored route, so a hang on the dead one or a withdrawal with the link up still fails | | `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 From d3ccafae4d6097cb5aee4a157c7e7ea6a951316e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 12:03:46 -0700 Subject: [PATCH 4/5] quest: point the cross-relay re-run at the IETF peer FETCH repro Co-Authored-By: Claude Opus 5.5 --- quest/m1/cross-relay-bursts.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/quest/m1/cross-relay-bursts.md b/quest/m1/cross-relay-bursts.md index 81ffedf2fa..09e3c89d69 100644 --- a/quest/m1/cross-relay-bursts.md +++ b/quest/m1/cross-relay-bursts.md @@ -32,6 +32,9 @@ used and against `main`: 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). + That drill uses a lite peer link; over a `moq-transport-19` peer link, a + related refusal does reproduce: FETCHes answered `old` for groups the origin + still holds ([IETF peer FETCH refused old](/quest/m1/ietf-peer-fetch-old.md)). - **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. From be76021c2ffacc241c886d3e063dea57d62e6c1c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 14:55:36 -0700 Subject: [PATCH 5/5] test(relay): keep the relays' default idle in the cluster drills The relays go back to the 10s idle timeout main uses, so the steady drill is set up exactly as before. The flap drill instead waits that long for the cut peer link to idle out: reads, FETCHes, and steps get the relay idle on top of SETTLE, and a group that arrives after the route returns is timed from the return. The setup timeouts this was meant to remove also happen on main (1 of 40 runs), with the relays on 10s, so they are tracked separately in the new impaired-handshake quest. That run also lost a group to a live Old reset whose FETCH was refused old, now recorded in cross-relay-bursts. Co-Authored-By: Claude Opus 5.5 --- quest/m1/cross-relay-bursts.md | 5 ++- quest/m1/test-flakes-2/README.md | 1 + quest/m1/test-flakes-2/impaired-handshake.md | 27 +++++++++++++ rs/moq-relay/tests/drills.rs | 42 ++++++++++++++------ test/drill/README.md | 8 ++-- 5 files changed, 66 insertions(+), 17 deletions(-) create mode 100644 quest/m1/test-flakes-2/impaired-handshake.md diff --git a/quest/m1/cross-relay-bursts.md b/quest/m1/cross-relay-bursts.md index 09e3c89d69..693da237ef 100644 --- a/quest/m1/cross-relay-bursts.md +++ b/quest/m1/cross-relay-bursts.md @@ -30,7 +30,10 @@ used and against `main`: `rs/moq-relay/tests/drills.rs::bursts_cross_a_cluster` (bursts of 30 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 + group arrives within about 3.5 s, and the occasional live `Old` reset is + usually recovered by FETCH. Not always: on `main` at `fe0113ff2`, 1 of 40 + runs lost group 89 to a live `Old` reset whose FETCH was also refused `old` + (seed 11025055814286623932). Several 30 s timers of that era have since been removed or shortened (#4606, #4741). That drill uses a lite peer link; over a `moq-transport-19` peer link, a related refusal does reproduce: FETCHes answered `old` for groups the origin diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 42a83e4cab..c51f824ece 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -34,3 +34,4 @@ Public API: none. Wire: none. - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed - [Import catalog finish](/quest/m1/test-flakes-2/import-catalog-finish.md) - `moq-cli`'s subprocess EOF catalog-finish test holds up under load with event-based fixture coordination - [Relay restart rebind](/quest/m1/test-flakes-2/relay-restart-rebind.md) - the crash drill restarts on its original UDP address under concurrent load +- [Impaired handshake](/quest/m1/test-flakes-2/impaired-handshake.md) - the impaired cluster drills' clients never time out while connecting diff --git a/quest/m1/test-flakes-2/impaired-handshake.md b/quest/m1/test-flakes-2/impaired-handshake.md new file mode 100644 index 0000000000..ef4bb96a72 --- /dev/null +++ b/quest/m1/test-flakes-2/impaired-handshake.md @@ -0,0 +1,27 @@ +# [S] Impaired drill clients connect reliably + +## Goal + +The impaired cluster drills (`bursts_cross_a_cluster::impaired`, +`bursts_cross_a_flapping_peer::impaired` in `rs/moq-relay/tests/drills.rs`) +never fail while connecting their publisher or subscriber. + +## Plan + +Measured 2026-10-07 by running the built drill binary in a loop (four tests +in parallel): + +- `main` at `fe0113ff2`: 1 of 40 runs failed `publisher connect failed: + Noq(Connection("timed out"))`, seed 3495212678159036442. +- #4972, with both relays back on the default 10s idle timeout: 3 of 40 runs, + `subscriber connect failed: Noq(Connection("timed out"))` twice and once + `failed to exchange h3 settings: ... reset by peer`. + +So the relays' idle timeout is not the cause. The likely cause is the drills' +2s client idle timeout (`quic()`) against a handshake over 5% loss and a +100 kbit/s bottleneck: noq arms the idle timer at `max(idle, 3 * PTO)` with +the base PTO, while a backed-off PTO can leave the client silent longer than +that. Confirm with a qlog of a failing seed, then fix it at the cause, without +raising a timeout or retrying the connect. + +Public API: none. Wire: none. diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 62e4345e66..f27d3f11f9 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -864,9 +864,10 @@ lanes!(bursts_cross_a_cluster); /// through it, the way an app recovering a control channel would. /// /// The grading stays as strict: the publisher keeps every group, so each still -/// arrives within [`SETTLE`] of being written or fails the drill by name. Only a -/// failure the flap caused is retried; a read or FETCH on the dead route that -/// never resolves, or a withdrawal while the link was up, still fails it. +/// arrives within [`SETTLE`] of being written, or of the route coming back when it +/// arrives after that, or fails the drill by name. Only a failure the flap caused +/// is retried; a read or FETCH on the dead route that outlasts the relays' idle +/// timeout, or a withdrawal while the link was up, still fails it. async fn bursts_cross_a_flapping_peer(lane: Lane) { cross_cluster(lane, PeerLink::Flapping).await } @@ -908,10 +909,13 @@ impl Flap { } async fn cross_cluster(lane: Lane, link: PeerLink) { - // Both relays get the drills' QUIC tuning, so a cut peer link idles out - // within the drill's budget rather than after the 10s default. + // The relays keep the default idle timeout. The drills' 2s one would also + // bound each relay's side of an impaired handshake, which can sit silent + // that long between the client's backed-off retransmits. So a cut peer link + // idles out only after `relay_idle`, which the flap's waits allow for. let mut config = relay_config(None); - config.quic = credit(lane, quic()); + config.quic = credit(lane, config.quic); + let relay_idle = config.quic.idle_timeout; let origin = RelayHost::start(config).await; let seed = seed(lane); let peer = match (link, lane) { @@ -925,7 +929,7 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { let mut peer_url = peer.url.clone(); peer_url.set_path("/"); let mut config = relay_config(None); - config.quic = credit(lane, quic()); + config.quic = credit(lane, config.quic); config.connect = dial(); // Retry forever: a give-up after a slow impaired handshake would look exactly // like the never-redialed regression, and a stall still fails the drill. @@ -1011,11 +1015,21 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { let mut slowest = (Duration::ZERO, 0); let mut generation = 0u64; let mut subscription_up = true; + // When the route was restored after a flap. A group that settles after it is + // timed from it, since the publisher could not reach the subscriber before. + let mut restored_at = None; + // How long one read or FETCH may take. A flap leaves those on the dead route + // waiting until the edge's peer session idles out, so they get that much more; + // one that outlasts even that hung on the dead route and still fails the drill. + let wait = match link { + PeerLink::Steady => SETTLE, + PeerLink::Flapping => SETTLE + relay_idle, + }; let fetch = |pending: &mut tokio::task::JoinSet<_>, track: &moq_net::track::Consumer, sequence: u64, generation: u64| { let fetching = track.fetch_group(sequence, None); pending.spawn(async move { - let fetched = tokio::time::timeout(SETTLE, async { + let fetched = tokio::time::timeout(wait, async { let group = fetching.await.map_err(|err| err.to_string())?; read_payload(group, sequence).await }); @@ -1028,9 +1042,9 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { }); }; - // 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); + // A margin past each settle task's own timeout, so a group that times out is + // reported by name rather than as a generic stall. + let step_timeout = wait + Duration::from_secs(1); while (outcomes.len() as u64) < total { let step = tokio::time::timeout(step_timeout, async { tokio::select! { @@ -1075,7 +1089,8 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { } Step::Settled(sequence, outcome, _) => { live.remove(&sequence); - let late = written[sequence as usize].elapsed(); + let since = written[sequence as usize].max(restored_at.unwrap_or(written[sequence as usize])); + let late = since.elapsed(); slowest = slowest.max((late, sequence)); outcomes.insert(sequence, outcome); } @@ -1090,6 +1105,7 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { } (false, _) => panic!("the route was withdrawn while the peer link was up"), (true, Flap::Withdrawn { cut }) => { + restored_at = Some(tokio::time::Instant::now()); reader = subscribe(&subscribed, "live").await; generation += 1; subscription_up = true; @@ -1122,7 +1138,7 @@ async fn cross_cluster(lane: Lane, link: PeerLink) { if claimed.insert(sequence) { live.insert(sequence); pending.spawn(async move { - let outcome = match tokio::time::timeout(SETTLE, read_payload(group, sequence)).await { + let outcome = match tokio::time::timeout(wait, read_payload(group, sequence)).await { Ok(Ok(())) => Outcome::Live, Ok(Err(err)) => Outcome::Failed(err), Err(_) => Outcome::Stalled, diff --git a/test/drill/README.md b/test/drill/README.md index b12797167a..bb3461e6f7 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -30,7 +30,7 @@ checks that what the fault owned was released. | `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 | -| `bursts_cross_a_flapping_peer` | The same bursts, with the peer link cut as the second burst is written and restored once the edge withdraws the route | how long the link was down before the withdrawal, and the reads and FETCHes the flap failed | the route comes back and every group still arrives within 10s of being written; only a failure the flap caused is retried, on the restored route, so a hang on the dead one or a withdrawal with the link up still fails | +| `bursts_cross_a_flapping_peer` | The same bursts, with the peer link cut as the second burst is written and restored once the edge withdraws the route | how long the link was down before the withdrawal, and the reads and FETCHes the flap failed | the route comes back and every group still arrives within 10s of being written, or of the route's return when it arrives after it; only a failure the flap caused is retried, on the restored route, so a hang on the dead one past the relays' idle timeout or a withdrawal with the link up still fails | | `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 @@ -87,8 +87,10 @@ loopback lane keeps the defaults. `bursts_cross_a_flapping_peer` cuts the peer link with `Shaper::cut`, which drops every datagram both ways until it is restored, so its loopback lane puts a shaper on that link too, one that forwards untouched. The cut lasts until the -subscriber sees the route withdrawn, which the relays' two-second idle timeout -bounds, rather than for a fixed time. +subscriber sees the route withdrawn, which the relays' default 10s idle timeout +bounds, rather than for a fixed time. The relays keep that default because a +short one also bounds their side of an impaired handshake, which can go quiet +for seconds between a client's backed-off retransmits. 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