From b6a1f96f56c82094f62f57c04861868ff7ff9c20 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:24:33 -0700 Subject: [PATCH 1/8] quest: claim redundant-failover-drift From 1eb62967bcd1233f17d59dc47208855ae56aadf7 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:43:34 -0700 Subject: [PATCH 2/8] fix(json): a relay's reset of a lost snapshot group is not fatal A relay resets a superseded group with the Old stream code, which reaches the subscriber as Error::Stream(StreamError::Old). The snapshot and window consumers only matched the local Error::Old, so a TS export behind a slow link died with "json: old" when the relay reset its catalog group. Co-Authored-By: Claude Opus 5.5 --- rs/moq-flate/src/snapshot/consumer.rs | 11 ++++- rs/moq-flate/src/snapshot/mod.rs | 59 ++++++++++++++------------- rs/moq-json/src/snapshot/consumer.rs | 11 ++++- rs/moq-json/src/snapshot/mod.rs | 57 ++++++++++++++------------ rs/moq-json/src/window/consumer.rs | 9 +++- rs/moq-json/src/window/mod.rs | 4 ++ 6 files changed, 91 insertions(+), 60 deletions(-) diff --git a/rs/moq-flate/src/snapshot/consumer.rs b/rs/moq-flate/src/snapshot/consumer.rs index 0fea756618d..5884f0999c5 100644 --- a/rs/moq-flate/src/snapshot/consumer.rs +++ b/rs/moq-flate/src/snapshot/consumer.rs @@ -99,9 +99,16 @@ impl Consumer { err @ (moq_net::Error::Old | moq_net::Error::Evicted | moq_net::Error::Lagged - | moq_net::Error::GroupTooLarge), + | moq_net::Error::GroupTooLarge + | moq_net::Error::Stream( + moq_net::StreamError::Old + | moq_net::StreamError::Evicted + | moq_net::StreamError::TooFarBehind + | moq_net::StreamError::GroupTooLarge, + )), )) => { - // A later group starts a fresh snapshot and compression window. + // Lost here or by a relay, whose reset arrives as its stream code. A later group + // starts a fresh snapshot and compression window. tracing::warn!(track = self.track.name(), group = group.sequence, error = ?err, "snapshot group lost; waiting for a newer one"); self.group = None; diff --git a/rs/moq-flate/src/snapshot/mod.rs b/rs/moq-flate/src/snapshot/mod.rs index 91887e02d3c..7ab9134bcbc 100644 --- a/rs/moq-flate/src/snapshot/mod.rs +++ b/rs/moq-flate/src/snapshot/mod.rs @@ -152,34 +152,37 @@ mod test { /// replacement rather than ending the reader. #[test] fn a_lost_group_waits_for_its_replacement() { - let track = moq_net::broadcast::Info::new() - .produce() - .create_track("test", timed()) - .unwrap(); - let mut consumer = consume(track.subscribe(None), false); - let waiter = kio::Waiter::noop(); - - // Group 0 delivers a value, then stays open with the reader parked on its next frame. - let mut group = track.create_group(moq_net::group::Info { sequence: 0 }).unwrap(); - group - .write_frame(moq_net::Timestamp::now(), Bytes::from_static(b"one")) - .unwrap(); - assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == "one")); - assert!(consumer.poll_latest(&waiter).is_pending()); - - // The relay reclaims the group out from under the reader. - group.abort(moq_net::Error::Old).unwrap(); - assert!( - consumer.poll_latest(&waiter).is_pending(), - "a lost group must not end the reader" - ); - - // The replacement arrives and the reader picks up where the value now lives. - let mut group = track.create_group(moq_net::group::Info { sequence: 1 }).unwrap(); - group - .write_frame(moq_net::Timestamp::now(), Bytes::from_static(b"two")) - .unwrap(); - assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == "two")); + // The relay's reset arrives as its stream code. + for err in [moq_net::Error::Old, moq_net::Error::Stream(moq_net::StreamError::Old)] { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("test", timed()) + .unwrap(); + let mut consumer = consume(track.subscribe(None), false); + let waiter = kio::Waiter::noop(); + + // Group 0 delivers a value, then stays open with the reader parked on its next frame. + let mut group = track.create_group(moq_net::group::Info { sequence: 0 }).unwrap(); + group + .write_frame(moq_net::Timestamp::now(), Bytes::from_static(b"one")) + .unwrap(); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == "one")); + assert!(consumer.poll_latest(&waiter).is_pending()); + + // The relay reclaims the group out from under the reader. + group.abort(err).unwrap(); + assert!( + consumer.poll_latest(&waiter).is_pending(), + "a lost group must not end the reader" + ); + + // The replacement arrives and the reader picks up where the value now lives. + let mut group = track.create_group(moq_net::group::Info { sequence: 1 }).unwrap(); + group + .write_frame(moq_net::Timestamp::now(), Bytes::from_static(b"two")) + .unwrap(); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == "two")); + } } /// A stamped payload is written at its capture time, and the returned size is the encoded frame diff --git a/rs/moq-json/src/snapshot/consumer.rs b/rs/moq-json/src/snapshot/consumer.rs index e630536ce56..a0ba18f870a 100644 --- a/rs/moq-json/src/snapshot/consumer.rs +++ b/rs/moq-json/src/snapshot/consumer.rs @@ -128,9 +128,16 @@ impl Consumer { err @ (moq_net::Error::Old | moq_net::Error::Evicted | moq_net::Error::Lagged - | moq_net::Error::GroupTooLarge), + | moq_net::Error::GroupTooLarge + | moq_net::Error::Stream( + moq_net::StreamError::Old + | moq_net::StreamError::Evicted + | moq_net::StreamError::TooFarBehind + | moq_net::StreamError::GroupTooLarge, + )), )) => { - // A later group starts a fresh snapshot and compression window. + // Lost here or by a relay, whose reset arrives as its stream code. A later group + // starts a fresh snapshot and compression window. tracing::warn!(track = self.track.name(), group = group.sequence, error = ?err, "snapshot group lost; waiting for a newer one"); self.group = None; diff --git a/rs/moq-json/src/snapshot/mod.rs b/rs/moq-json/src/snapshot/mod.rs index 6bc03283839..c74616bb6b0 100644 --- a/rs/moq-json/src/snapshot/mod.rs +++ b/rs/moq-json/src/snapshot/mod.rs @@ -237,34 +237,37 @@ mod test { /// catalog group aged out of the relay cache underneath it. #[test] fn a_lost_group_waits_for_its_replacement() { - let track = moq_net::broadcast::Info::new() - .produce() - .create_track("test", timed()) - .unwrap(); - let mut consumer = Consumer::::new(track.subscribe(None), consumer::Config::default()); - let waiter = kio::Waiter::noop(); - - // Group 0 delivers a value, then stays open with the reader parked on its next frame. - let mut group = track.create_group(moq_net::group::Info { sequence: 0 }).unwrap(); - group - .write_frame(moq_net::Timestamp::now(), Bytes::from_static(br#"{"a":1}"#)) - .unwrap(); - assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == json!({ "a": 1 }))); - assert!(consumer.poll_latest(&waiter).is_pending()); - - // The relay reclaims the group out from under the reader. - group.abort(moq_net::Error::Old).unwrap(); - assert!( - consumer.poll_latest(&waiter).is_pending(), - "a lost group must not end the reader" - ); + // The relay's reset arrives as its stream code. + for err in [moq_net::Error::Old, moq_net::Error::Stream(moq_net::StreamError::Old)] { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("test", timed()) + .unwrap(); + let mut consumer = Consumer::::new(track.subscribe(None), consumer::Config::default()); + let waiter = kio::Waiter::noop(); - // The replacement arrives and the reader picks up where the value now lives. - let mut group = track.create_group(moq_net::group::Info { sequence: 1 }).unwrap(); - group - .write_frame(moq_net::Timestamp::now(), Bytes::from_static(br#"{"a":2}"#)) - .unwrap(); - assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == json!({ "a": 2 }))); + // Group 0 delivers a value, then stays open with the reader parked on its next frame. + let mut group = track.create_group(moq_net::group::Info { sequence: 0 }).unwrap(); + group + .write_frame(moq_net::Timestamp::now(), Bytes::from_static(br#"{"a":1}"#)) + .unwrap(); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == json!({ "a": 1 }))); + assert!(consumer.poll_latest(&waiter).is_pending()); + + // The relay reclaims the group out from under the reader. + group.abort(err).unwrap(); + assert!( + consumer.poll_latest(&waiter).is_pending(), + "a lost group must not end the reader" + ); + + // The replacement arrives and the reader picks up where the value now lives. + let mut group = track.create_group(moq_net::group::Info { sequence: 1 }).unwrap(); + group + .write_frame(moq_net::Timestamp::now(), Bytes::from_static(br#"{"a":2}"#)) + .unwrap(); + assert!(matches!(consumer.poll_latest(&waiter), Poll::Ready(Ok(Some(v))) if v.value == json!({ "a": 2 }))); + } } /// Nothing replaces a lost group once the track is finished, so the reader ends cleanly on the diff --git a/rs/moq-json/src/window/consumer.rs b/rs/moq-json/src/window/consumer.rs index 8d217777e96..2a67ee2edd6 100644 --- a/rs/moq-json/src/window/consumer.rs +++ b/rs/moq-json/src/window/consumer.rs @@ -73,8 +73,15 @@ impl Consumer { moq_net::Error::Old | moq_net::Error::Lagged | moq_net::Error::Evicted - | moq_net::Error::GroupTooLarge, + | moq_net::Error::GroupTooLarge + | moq_net::Error::Stream( + moq_net::StreamError::Old + | moq_net::StreamError::TooFarBehind + | moq_net::StreamError::Evicted + | moq_net::StreamError::GroupTooLarge, + ), )) => { + // Lost here or by a relay, whose reset arrives as its stream code. // This group is no longer complete, but the next one starts with a checkpoint // that can account for everything this reader missed. self.group = None; diff --git a/rs/moq-json/src/window/mod.rs b/rs/moq-json/src/window/mod.rs index 82c5fdc0eea..040987c8463 100644 --- a/rs/moq-json/src/window/mod.rs +++ b/rs/moq-json/src/window/mod.rs @@ -556,6 +556,10 @@ mod test { moq_net::Error::Lagged, moq_net::Error::Evicted, moq_net::Error::GroupTooLarge, + moq_net::Error::Stream(moq_net::StreamError::Old), + moq_net::Error::Stream(moq_net::StreamError::TooFarBehind), + moq_net::Error::Stream(moq_net::StreamError::Evicted), + moq_net::Error::Stream(moq_net::StreamError::GroupTooLarge), ] { let track = moq_net::broadcast::Info::new() .produce() From c6441ac36baf88af6ffd1d521ed39c1375f6720f Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:30:01 -0700 Subject: [PATCH 3/8] fix(mux): holes a stalled source left behind expire together A missing group's monotonic budget started when the read cursor reached it. Once a source stopped (a killed importer whose route outlives it until the QUIC idle timeout), the newest content stopped advancing, so a backlog with every other group missing drained one budget per hole. The TS export's jitter clock read that metered trickle as a source far slower than real time and failed a stitching export before it could switch back to the surviving importer. A hole is now late from when the first group past it arrived, in both the Rust and JS container consumers. The redundant failover test also prints the exporter's stderr when it exits early. Co-Authored-By: Claude Opus 5.5 --- js/hang/src/container/consumer.holes.test.ts | 41 +++++++++++++++ js/hang/src/container/consumer.ts | 23 +++++---- rs/moq-cli/tests/integration/export.rs | 5 +- rs/moq-mux/src/container/consumer.rs | 52 +++++++++++++++----- 4 files changed, 99 insertions(+), 22 deletions(-) diff --git a/js/hang/src/container/consumer.holes.test.ts b/js/hang/src/container/consumer.holes.test.ts index 5f17293e93d..8d184335c69 100644 --- a/js/hang/src/container/consumer.holes.test.ts +++ b/js/hang/src/container/consumer.holes.test.ts @@ -102,3 +102,44 @@ test("a filled hole cancels its timer and the next hole gets a fresh budget", as restore(); } }); + +test("the holes a stalled source left behind share one budget", async () => { + const clock = new Manual(); + const restore = clock.install(); + const track = new Track.Producer("data"); + const consumer = new Consumer(track.subscribe({ maxDelay: Milli(30_000) }), { + format: new Format("data"), + maxDelay: Milli(100), + }); + const results: Read[] = []; + const reading = (async () => { + for (;;) { + const read = await consumer.next(); + if (!read) return; + results.push(read); + } + })(); + const write = (sequence: number) => { + const group = new Group.Producer(sequence); + track.writeGroup(group); + group.writeFrame({ + payload: encodeFrame(new Uint8Array([1]), Micro(sequence * 10_000)), + timestamp: Time.Timestamp.fromMicros(sequence * 10_000), + }); + group.close(); + }; + try { + write(0); + await flush(); + for (const sequence of [2, 4, 6]) write(sequence); + await flush(); + expect(results).toHaveLength(1); + await clock.advanceAsync(Nano.fromMilli(Milli(100))); + expect(results.map((read) => read.kind)).toEqual(["frame", "gap", "frame", "gap", "frame", "gap", "frame"]); + } finally { + consumer.close(); + track.close(); + await reading; + restore(); + } +}); diff --git a/js/hang/src/container/consumer.ts b/js/hang/src/container/consumer.ts index 51770c432a4..66a879f5f4b 100644 --- a/js/hang/src/container/consumer.ts +++ b/js/hang/src/container/consumer.ts @@ -13,8 +13,8 @@ export interface ConsumerProps { * How stale a group may get before it is skipped, in milliseconds (default: 0). * * Measured as the span from the oldest buffered frame to the newest, so it bounds how long a - * late group is waited for. A missing sequence also expires after this much monotonic - * time blocked on it, even when the next group is a marker on a paused track. + * late group is waited for. A missing sequence also expires this much monotonic time after + * the first group past it arrived, even when the next group is a marker on a paused track. * The local half of the subscription's * `maxDelay`; both measure the same budget, one on the wire and one as frames are read. */ @@ -52,6 +52,7 @@ export type Read = interface Group { consumer: Moq.Group.Consumer; + arrived: Instant; // from then on, any sequence missing below it is late frames: ReceivedFrame[]; // decode order empty: boolean; // no wire frame was published; empty groups mean nothing media: boolean; // a decodable (non-marker) frame was buffered @@ -105,7 +106,8 @@ export class Consumer { #maxDelay: Getter; #groups: Group[] = []; #active?: number; // the active group sequence number - #hole?: { sequence: number; since: Instant }; + // When the missing sequence at the cursor became late, while one is. + #hole?: Instant; // Presentation end (max PTS + duration) of the group we most recently advanced past, so next()'s // promotion guard can tell a timeline-continuous next group from one sitting after a gap. // Maintained only via #recordPresented; see its comment for the invariant. @@ -193,6 +195,7 @@ export class Consumer { // may be is the subscription's own max delay, applied before it ever reaches here. const group: Group = { consumer, + arrived: now(), frames: [], empty: true, media: false, @@ -527,13 +530,13 @@ export class Consumer { // here matters: #checkMaxDelay would instead drop the head, the very group to play next. // After track termination no missing group can arrive, so drain across any remaining gap. // A missing sequence also expires on elapsed monotonic time, so timestamp-free markers - // behind a hole can end a paused track without more arrivals. + // behind a hole can end a paused track without more arrivals. It is late since a group + // past it arrived, not since the cursor got here: once the edge stops moving, a backlog + // with every other group missing would otherwise drain one budget per hole. if (this.#active !== undefined && this.#groups.length > 0) { const head = this.#groups[0]; if (head.consumer.sequence > this.#active) { - if (this.#hole?.sequence !== this.#active) { - this.#hole = { sequence: this.#active, since: now() }; - } + this.#hole = this.#groups.reduce((min, group) => (group.arrived < min ? group.arrived : min), head.arrived); const startup = this.#start === undefined; const contiguous = ptsContiguous(this.#presentedEnd, head.frames.at(0)?.timestamp); const empty = head.empty && head.consumer.done; @@ -544,7 +547,7 @@ export class Consumer { (this.#presentedEnd !== undefined && head.latest !== undefined && head.latest - this.#presentedEnd > maxDelay)); - const elapsed = Instant.durationSince(now(), this.#hole.since) >= Nano.fromMicro(maxDelay); + const elapsed = Instant.durationSince(now(), this.#hole) >= Nano.fromMicro(maxDelay); if (startup || empty || contiguous || skipHole || elapsed || ended !== undefined) { if (!contiguous && !empty) { this.#lost = true; @@ -640,9 +643,9 @@ export class Consumer { const abort = this.#signals.abort; if (abort.aborted) return undefined; let timer: Effect | undefined; - if (this.#hole) { + if (this.#hole !== undefined) { timer = new Effect(); - const since = this.#hole.since; + const since = this.#hole; timer.run((effect) => { const budget = Nano.fromMilli(effect.get(this.#maxDelay)); const elapsed = Instant.durationSince(now(), since); diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index 8e268e39436..1acb83a8d2a 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -912,7 +912,10 @@ async fn redundant_failover(bufferbloat: bool) { for export in &mut exports { tokio::time::timeout(TIMEOUT, async { loop { - assert!(export.child.try_wait().unwrap().is_none(), "stitching export exited"); + if export.child.try_wait().unwrap().is_some() { + let errors = (&mut export.stderr).await.expect("collect exporter stderr"); + panic!("stitching export exited: {errors}"); + } if packets(&export.output.lock().unwrap()) .filter_map(|packet| pat_version(packet)) .any(|v| v == version) diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index 150766f8345..18ed65b94de 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -43,8 +43,9 @@ pub(crate) enum Item { /// With a non-zero max delay, slow groups are tolerated up to that budget before being /// skipped. A missing sequence gets the same tolerance: there is no way to tell a stream /// that lost the delivery race from one the cache evicted. -/// Its budget also expires after that much monotonic time blocked on it, so a marker -/// behind a hole is delivered even when the track is paused. +/// Its budget also expires that much monotonic time after the first group past it arrived, so +/// a marker behind a hole is delivered even when the track is paused, and the holes a stalled +/// source left behind expire together rather than one budget each. /// /// Delivery starts at the lowest absolute start among the /// [`Subscription::ranges`](moq_net::track::Subscription::ranges) when every range names one, waiting for that group under the budget; without one it @@ -100,8 +101,7 @@ pub struct Consumer { // How far we may drift from the live edge before skipping a group. max_delay: std::time::Duration, - // Missing sequence and when it first blocked delivery; cleared when the cursor moves. - hole: Option<(u64, moq_clock::Instant)>, + // Wakes the reader once the hole at the cursor expires. hole_timer: Option>>, // The furthest presentation point delivered so far: max(timestamp + duration), endpoints @@ -188,7 +188,6 @@ impl Consumer { startup: start.is_none(), newest: false, max_delay, - hole: None, hole_timer: None, edge: None, start: None, @@ -371,7 +370,6 @@ impl Consumer { 'read: loop { if self.pending.front().is_none_or(|group| group.sequence <= self.current) { - self.hole = None; self.hole_timer = None; } @@ -516,10 +514,10 @@ impl Consumer { && front.sequence > self.current { let sequence = front.sequence; - if self.hole.is_none_or(|(sequence, _)| sequence != self.current) { - self.hole = Some((self.current, moq_clock::now())); - } - let since = self.hole.unwrap().1; + // Late since a group past it arrived, not since the cursor got here: once the edge + // stops moving, a backlog with every other group missing would otherwise drain one + // budget per hole, which reads downstream as a source far slower than real time. + let since = self.pending.iter().map(|group| group.arrived).min().expect("a front group"); let elapsed = since.elapsed() >= self.max_delay; let aged = next_group.is_some_and(|(_, start)| max_timestamp.saturating_sub(start) >= self.max_delay); let excluded = self @@ -617,7 +615,6 @@ impl Consumer { } fn note_group_edge(&mut self) { - self.hole = None; self.hole_timer = None; if let Some((_, ts)) = self.start { self.floor = Some(ts); @@ -724,6 +721,9 @@ enum Failure { struct GroupBuffer { group: moq_net::group::Consumer, + // When the group arrived: from then on, any sequence missing below it is late. + arrived: moq_clock::Instant, + // The current frame index within the group. index: usize, @@ -764,6 +764,7 @@ impl GroupBuffer { fn new(group: moq_net::group::Consumer) -> Self { Self { group, + arrived: moq_clock::now(), index: 0, empty: true, media: false, @@ -1310,6 +1311,35 @@ mod tests { track.finish().unwrap(); } + /// The holes a source left behind when it stopped expire together, a budget after the groups + /// past them arrived, rather than one budget each as the cursor reaches them. + #[tokio::test(start_paused = true)] + async fn holes_behind_a_stalled_source_share_one_budget() { + let budget = Duration::from_millis(100); + let mut track = track_producer("data", hang::container::track_info(hang::catalog::PRIORITY.audio)); + let subscriber = track.subscribe(moq_net::track::Subscription::default().with_max_delay(budget)); + let mut reader = Consumer::new(subscriber, Container::Legacy(crate::container::Kind::Data)); + write_group(&mut track, 0, &[ts(0)]); + assert!(matches!( + reader.read().await.unwrap(), + Some(Read::Frame { group: 0, .. }) + )); + for sequence in [2, 4, 6] { + write_group(&mut track, sequence, &[ts(sequence * 10_000)]); + } + assert!(reader.poll_read(&kio::Waiter::noop()).is_pending()); + tokio::time::advance(budget).await; + let mut read = Vec::new(); + while let Poll::Ready(Ok(Some(item))) = reader.poll_read(&kio::Waiter::noop()) { + read.push(match item { + Read::Frame { group, .. } => format!("frame {group}"), + Read::Gap { .. } => "gap".to_string(), + Read::End { .. } => "end".to_string(), + }); + } + assert_eq!(read, ["gap", "frame 2", "gap", "frame 4", "gap", "frame 6"]); + } + /// A malformed data marker group found while looking ahead fails the reader, rather than /// reading as a gap when a latency skip passes over it. #[tokio::test(start_paused = true)] From e4e24f61baaa30dc31b339aee45b7af7b57e3813 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:32:47 -0700 Subject: [PATCH 4/8] test(cli): give redundant importers a 2s QUIC idle timeout A killed importer sends no close, so the relay keeps its route until the idle timeout. A short one, as a redundant ingest would run, keeps the outage the failover tests wait out to seconds. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/tests/integration/export.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index 1acb83a8d2a..82e8cbdba20 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -834,9 +834,12 @@ fn redundant_feed() -> (tokio::sync::mpsc::UnboundedSender, JoinHand (add, task) } -/// Start an importer of the live source on lite-07, with `flags` for `import ts`. +/// Start an importer of the live source on lite-07, with `flags` for `import ts`. A killed +/// importer sends no close, so the relay keeps its route until the QUIC idle timeout: a short +/// one, as a redundant ingest would run, keeps that outage to seconds. fn redundant_import(relay: &Relay, feed: &tokio::sync::mpsc::UnboundedSender, flags: &[&str]) -> Child { - let mut child = moq(relay, &["--connect-version", LITE_07]) + let global = ["--connect-version", LITE_07, "--quic-idle-timeout", "2s", "--quic-keep-alive", "500ms"]; + let mut child = moq(relay, &global) .args(["import", "--max-age", "10s", "ts"]) .args(flags) .stdin(Stdio::piped()) From 6d206abf9112955a6dbb2edb13429657a5033bac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 20 Oct 2026 16:42:53 -0700 Subject: [PATCH 5/8] style(hang): format the hole budget anchor Co-Authored-By: Claude Opus 5.5 --- js/hang/src/container/consumer.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/js/hang/src/container/consumer.ts b/js/hang/src/container/consumer.ts index 66a879f5f4b..7457ca5710b 100644 --- a/js/hang/src/container/consumer.ts +++ b/js/hang/src/container/consumer.ts @@ -536,7 +536,10 @@ export class Consumer { if (this.#active !== undefined && this.#groups.length > 0) { const head = this.#groups[0]; if (head.consumer.sequence > this.#active) { - this.#hole = this.#groups.reduce((min, group) => (group.arrived < min ? group.arrived : min), head.arrived); + this.#hole = this.#groups.reduce( + (min, group) => (group.arrived < min ? group.arrived : min), + head.arrived, + ); const startup = this.#start === undefined; const contiguous = ptsContiguous(this.#presentedEnd, head.frames.at(0)?.timestamp); const empty = head.empty && head.consumer.done; From f94b997f184564bfd081d15b099f882dd40f000e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 20 Oct 2026 16:48:29 -0700 Subject: [PATCH 6/8] quest: finish redundant-failover-drift Co-Authored-By: Claude Opus 5.5 --- quest/m1/audit-test-cost.md | 5 ++- quest/m1/resume-outage.md | 1 - quest/m1/test-flakes-2/README.md | 1 - .../test-flakes-2/redundant-failover-drift.md | 41 ------------------- 4 files changed, 3 insertions(+), 45 deletions(-) delete mode 100644 quest/m1/test-flakes-2/redundant-failover-drift.md diff --git a/quest/m1/audit-test-cost.md b/quest/m1/audit-test-cost.md index b4a5f697e0b..6e1391e889d 100644 --- a/quest/m1/audit-test-cost.md +++ b/quest/m1/audit-test-cost.md @@ -32,8 +32,9 @@ the stats session graph's recursive Send proof (send-depth and fuzz), bench/ffi-gen's stale wasm-bindgen lock, moq-uring's fixed 1 ms calibration slack, the Nightly JS benchmark stall, the CI selection test's feature count, and the stats producer benchmark lint. It planned -[redundant failover drift](/quest/m1/test-flakes-2/redundant-failover-drift.md) -and [macOS QUIC connects](/quest/m1/test-flakes-2/macos-quic-connect.md). +redundant failover drift, fixed by +[moq-dev/moq#6113](https://github.com/moq-dev/moq/pull/6113), and +[macOS QUIC connects](/quest/m1/test-flakes-2/macos-quic-connect.md). Baselines from queue run [39371990135](https://github.com/moq-dev/moq/actions/runs/39371990135): diff --git a/quest/m1/resume-outage.md b/quest/m1/resume-outage.md index e63430d778e..31ea4001424 100644 --- a/quest/m1/resume-outage.md +++ b/quest/m1/resume-outage.md @@ -65,4 +65,3 @@ Decided with the maintainer, 2026-10-20: - [Resumed in-flight groups](/quest/m1/resume-in-flight.md) - which replaced copies stay while groups below them are in flight - [A widened max age re-offers cached groups](/quest/m1/resolved-demand.md) - the same max-age bound on a warm relay -- [Redundant failover drift](/quest/m1/test-flakes-2/redundant-failover-drift.md) - the same idle-timeout outage hitting the TS jitter clock diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index bcf8a0dad5c..98e955f19e7 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -40,7 +40,6 @@ Public API: none. Wire: none. ## Required -- [Redundant failover drift](/quest/m1/test-flakes-2/redundant-failover-drift.md) - a stitching TS export switches back to the surviving importer behind bufferbloat instead of failing on a stalled clock - [macOS QUIC connects](/quest/m1/test-flakes-2/macos-quic-connect.md) - moq-tokio's loopback QUIC tests stop timing out on the macOS Platform job ## Related diff --git a/quest/m1/test-flakes-2/redundant-failover-drift.md b/quest/m1/test-flakes-2/redundant-failover-drift.md deleted file mode 100644 index 4701cd08ee6..00000000000 --- a/quest/m1/test-flakes-2/redundant-failover-drift.md +++ /dev/null @@ -1,41 +0,0 @@ -# [S] Redundant TS failover survives a killed importer behind bufferbloat - -## Goal - -moq-cli's `export::redundant_importers_switch_through_bufferbloat_on_lite_07` -(`rs/moq-cli/tests/integration/export.rs`) passes reliably, fixed at its -cause. On current main it failed 3 of 10 isolated local runs and 2 of 3 -loaded `just check` runs with "stitching export exited". It is also the -slowest test in the queue suite (37 s, past nextest's 30 s SLOW mark). - -## Plan - -Found by the 2026-10-19 test cost audit, with the exporter's stderr captured -on a failing run. After the test kills the newest importer, the exporter -fails instead of switching back to the survivor: - -> export failed with the broadcast still up, so not lingering ... the -> source's clock runs -6129221.2 ppm off ours, past the 30 ppm the output's may follow - -A killed importer sends no close, so the relay keeps its route until the QUIC -idle timeout (10 s by default). The viewer meanwhile gets no media; behind the -160 kbit/s shaper the backlog adds to that stall. The jitter clock in -`rs/moq-mux/src/jitter.rs` charges the stall to source drift and gives up -once it has cost half the delay. The unshaped sibling shows the same stall as -an 11 s output gap at its second switch, but passes. - -Decide which side is wrong before changing the test, recommended first: - -1. The jitter clock should not read a stalled source as drift. A stall is - missing media, which the follower can wait out or switch away from. -2. The follower should switch to the live survivor once the stalled route's - media is stale, without waiting for the relay to drop the route. -3. Both are right, and the export legitimately fails behind a link below the - media rate. Then the test's scenario needs another shape. - -Giving the importers `--quic-idle-timeout 2s --quic-keep-alive 500ms` -passed 5 of 5 runs and cut the three redundant-importer tests from 15, 29 and -37 s to 4, 16 and 30 s. That only shortens the stall, so land it after the -cause is fixed, not instead of it. - -Public API: none. Wire: none. From e7a4765ef65d700d79123b63a65b860c5328b223 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 20 Oct 2026 16:53:51 -0700 Subject: [PATCH 7/8] style: rustfmt the failover fix Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/tests/integration/export.rs | 9 ++++++++- rs/moq-mux/src/container/consumer.rs | 7 ++++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index 82e8cbdba20..c2d004c1446 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -838,7 +838,14 @@ fn redundant_feed() -> (tokio::sync::mpsc::UnboundedSender, JoinHand /// importer sends no close, so the relay keeps its route until the QUIC idle timeout: a short /// one, as a redundant ingest would run, keeps that outage to seconds. fn redundant_import(relay: &Relay, feed: &tokio::sync::mpsc::UnboundedSender, flags: &[&str]) -> Child { - let global = ["--connect-version", LITE_07, "--quic-idle-timeout", "2s", "--quic-keep-alive", "500ms"]; + let global = [ + "--connect-version", + LITE_07, + "--quic-idle-timeout", + "2s", + "--quic-keep-alive", + "500ms", + ]; let mut child = moq(relay, &global) .args(["import", "--max-age", "10s", "ts"]) .args(flags) diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index 18ed65b94de..4b7bbdfa6f7 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -517,7 +517,12 @@ impl Consumer { // Late since a group past it arrived, not since the cursor got here: once the edge // stops moving, a backlog with every other group missing would otherwise drain one // budget per hole, which reads downstream as a source far slower than real time. - let since = self.pending.iter().map(|group| group.arrived).min().expect("a front group"); + let since = self + .pending + .iter() + .map(|group| group.arrived) + .min() + .expect("a front group"); let elapsed = since.elapsed() >= self.max_delay; let aged = next_group.is_some_and(|(_, start)| max_timestamp.saturating_sub(start) >= self.max_delay); let excluded = self From f14d324a167d1fe410c790dcb950c6834b8b4994 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 21:33:05 -0700 Subject: [PATCH 8/8] perf(mux): keep a hole's lateness on insert instead of rescanning the backlog Each pending group now carries the earliest arrival among it and the groups past it. Arrivals only move forward, so an insert copies it from the next group and nothing below changes; a hole reads it from the head in O(1) rather than scanning every pending group per hole, which a sparse backlog from a peer made quadratic. Co-Authored-By: Claude Opus 5.5 --- js/hang/src/container/consumer.holes.test.ts | 44 +++++++++++++++++ js/hang/src/container/consumer.ts | 19 ++++---- rs/moq-mux/src/container/consumer.rs | 50 ++++++++++++++++---- 3 files changed, 95 insertions(+), 18 deletions(-) diff --git a/js/hang/src/container/consumer.holes.test.ts b/js/hang/src/container/consumer.holes.test.ts index 8d184335c69..407356a2fd4 100644 --- a/js/hang/src/container/consumer.holes.test.ts +++ b/js/hang/src/container/consumer.holes.test.ts @@ -143,3 +143,47 @@ test("the holes a stalled source left behind share one budget", async () => { restore(); } }); + +test("a hole is late from the earliest arrival past it", async () => { + const clock = new Manual(); + const restore = clock.install(); + const track = new Track.Producer("data"); + const consumer = new Consumer(track.subscribe({ maxDelay: Milli(30_000) }), { + format: new Format("data"), + maxDelay: Milli(100), + }); + const results: Read[] = []; + const reading = (async () => { + for (;;) { + const read = await consumer.next(); + if (!read) return; + results.push(read); + } + })(); + const write = (sequence: number) => { + const group = new Group.Producer(sequence); + track.writeGroup(group); + group.writeFrame({ + payload: encodeFrame(new Uint8Array([1]), Micro(sequence * 10_000)), + timestamp: Time.Timestamp.fromMicros(sequence * 10_000), + }); + group.close(); + }; + try { + write(0); + await flush(); + write(4); + await flush(); + await clock.advanceAsync(Nano.fromMilli(Milli(50))); + write(2); + await flush(); + expect(results).toHaveLength(1); + await clock.advanceAsync(Nano.fromMilli(Milli(50))); + expect(results.map((read) => read.kind)).toEqual(["frame", "gap", "frame", "gap", "frame"]); + } finally { + consumer.close(); + track.close(); + await reading; + restore(); + } +}); diff --git a/js/hang/src/container/consumer.ts b/js/hang/src/container/consumer.ts index 7457ca5710b..d2495c9c868 100644 --- a/js/hang/src/container/consumer.ts +++ b/js/hang/src/container/consumer.ts @@ -52,7 +52,9 @@ export type Read = interface Group { consumer: Moq.Group.Consumer; - arrived: Instant; // from then on, any sequence missing below it is late + // From then on, any sequence missing below this group is late: the earliest arrival among it + // and the groups past it, kept on insert so a hole reads it without scanning the backlog. + late: Instant; frames: ReceivedFrame[]; // decode order empty: boolean; // no wire frame was published; empty groups mean nothing media: boolean; // a decodable (non-marker) frame was buffered @@ -195,7 +197,7 @@ export class Consumer { // may be is the subscription's own max delay, applied before it ever reaches here. const group: Group = { consumer, - arrived: now(), + late: now(), frames: [], empty: true, media: false, @@ -203,8 +205,12 @@ export class Consumer { // Insert into #groups based on the group sequence number (ascending). // This is used to cancel old groups. - this.#groups.push(group); - this.#groups.sort((a, b) => a.consumer.sequence - b.consumer.sequence); + let index = this.#groups.findIndex((other) => other.consumer.sequence > consumer.sequence); + if (index === -1) index = this.#groups.length; + // Arrivals only move forward, so the groups below this one keep their earliest arrival. + const next = this.#groups.at(index); + if (next && next.late < group.late) group.late = next.late; + this.#groups.splice(index, 0, group); // Start buffering frames from this group this.#signals.spawn(this.#runGroup.bind(this, group)); @@ -536,10 +542,7 @@ export class Consumer { if (this.#active !== undefined && this.#groups.length > 0) { const head = this.#groups[0]; if (head.consumer.sequence > this.#active) { - this.#hole = this.#groups.reduce( - (min, group) => (group.arrived < min ? group.arrived : min), - head.arrived, - ); + this.#hole = head.late; const startup = this.#start === undefined; const contiguous = ptsContiguous(this.#presentedEnd, head.frames.at(0)?.timestamp); const empty = head.empty && head.consumer.done; diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index 993aad53d2b..3f85b3e8da0 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -525,12 +525,7 @@ impl Consumer { // Late since a group past it arrived, not since the cursor got here: once the edge // stops moving, a backlog with every other group missing would otherwise drain one // budget per hole, which reads downstream as a source far slower than real time. - let since = self - .pending - .iter() - .map(|group| group.arrived) - .min() - .expect("a front group"); + let since = front.late; let elapsed = since.elapsed() >= self.max_delay; let aged = next_group.is_some_and(|(_, start)| max_timestamp.saturating_sub(start) >= self.max_delay); let excluded = self @@ -655,7 +650,7 @@ impl Consumer { return Poll::Ready(()); }; - let reader = GroupBuffer::new(group); + let mut reader = GroupBuffer::new(group); let sequence = reader.group.sequence; if sequence < self.current { @@ -666,6 +661,10 @@ impl Consumer { let idx = self .pending .partition_point(|g| g.group.sequence < reader.group.sequence); + // Arrivals only move forward, so the groups below this one keep their earliest arrival. + if let Some(next) = self.pending.get(idx) { + reader.late = reader.late.min(next.late); + } self.pending.insert(idx, reader); } } @@ -734,8 +733,9 @@ enum Failure { struct GroupBuffer { group: moq_net::group::Consumer, - // When the group arrived: from then on, any sequence missing below it is late. - arrived: moq_clock::Instant, + // From then on, any sequence missing below this group is late: the earliest arrival among it + // and the groups past it, kept on insert so a hole reads it without scanning the backlog. + late: moq_clock::Instant, // The current frame index within the group. index: usize, @@ -777,7 +777,7 @@ impl GroupBuffer { fn new(group: moq_net::group::Consumer) -> Self { Self { group, - arrived: moq_clock::now(), + late: moq_clock::now(), index: 0, empty: true, media: false, @@ -1341,6 +1341,36 @@ mod tests { assert_eq!(read, ["gap", "frame 2", "gap", "frame 4", "gap", "frame 6"]); } + /// A group that arrives late below an earlier one inherits that earlier arrival: the holes + /// around it were already observable then. + #[tokio::test(start_paused = true)] + async fn a_hole_is_late_from_the_earliest_arrival_past_it() { + let budget = Duration::from_millis(100); + let mut track = track_producer("data", hang::container::track_info(hang::catalog::PRIORITY.audio)); + let subscriber = track.subscribe(moq_net::track::Subscription::default().with_max_delay(budget)); + let mut reader = Consumer::new(subscriber, Container::Legacy(crate::container::Kind::Data)); + write_group(&mut track, 0, &[ts(0)]); + assert!(matches!( + reader.read().await.unwrap(), + Some(Read::Frame { group: 0, .. }) + )); + write_group(&mut track, 4, &[ts(40_000)]); + assert!(reader.poll_read(&kio::Waiter::noop()).is_pending()); + tokio::time::advance(budget / 2).await; + write_group(&mut track, 2, &[ts(20_000)]); + assert!(reader.poll_read(&kio::Waiter::noop()).is_pending()); + tokio::time::advance(budget / 2).await; + let mut read = Vec::new(); + while let Poll::Ready(Ok(Some(item))) = reader.poll_read(&kio::Waiter::noop()) { + read.push(match item { + Read::Frame { group, .. } => format!("frame {group}"), + Read::Gap { .. } => "gap".to_string(), + Read::End { .. } => "end".to_string(), + }); + } + assert_eq!(read, ["gap", "frame 2", "gap", "frame 4"]); + } + /// A malformed data marker group found while looking ahead fails the reader, rather than /// reading as a gap when a latency skip passes over it. #[tokio::test(start_paused = true)]