diff --git a/js/hang/src/container/consumer.holes.test.ts b/js/hang/src/container/consumer.holes.test.ts index 5f17293e93d..407356a2fd4 100644 --- a/js/hang/src/container/consumer.holes.test.ts +++ b/js/hang/src/container/consumer.holes.test.ts @@ -102,3 +102,88 @@ 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(); + } +}); + +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 51770c432a4..d2495c9c868 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,9 @@ export type Read = interface Group { consumer: Moq.Group.Consumer; + // 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 @@ -105,7 +108,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 +197,7 @@ export class Consumer { // may be is the subscription's own max delay, applied before it ever reaches here. const group: Group = { consumer, + late: now(), frames: [], empty: true, media: false, @@ -200,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)); @@ -527,13 +536,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 = head.late; const startup = this.#start === undefined; const contiguous = ptsContiguous(this.#presentedEnd, head.frames.at(0)?.timestamp); const empty = head.empty && head.consumer.done; @@ -544,7 +553,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 +649,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/quest/m0/README.md b/quest/m0/README.md index d27fb92f2ba..cae1b4f9705 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -30,11 +30,6 @@ viewers). #4741 can merge to main, but no release ships until first-party publishers mint epochs. Backport patches cut from `release` don't carry #4741 and aren't gated (decided 2026-10-08). -Flakes (2026-10-20, maintainer): the redundant failover drift moved here from -the m1 [load round](/quest/m1/test-flakes-2/README.md). It fails loaded -`just check` runs, the local gate every merge waits on, and may hide a TS -export bug. - Hygiene (decided 2026-10-19): the audits sit at the end of this list. Each recurs once per major release: its PR adds the next run to the following milestone, gated from m2 onward on the [release cut](/quest/m2/release/cut.md). @@ -43,7 +38,6 @@ Each lands its fixes in its own PR and plans only what needs a decision. ## Required - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - every first-party publisher that can restart mints a fresh route epoch, the newest wins a path, and only routes with the same epoch resume a subscription -- [Redundant failover drift](/quest/m0/redundant-failover-drift.md) - a stitching TS export switches back to the surviving importer behind bufferbloat instead of failing on a stalled clock - [web-transport-proto 0.6.3](/quest/m0/web-transport-proto-0.6.3.md) - `main` and `release` lock the upstream release that bounds decoded CONNECT headers, so a peer can't make a relay hold megabytes per session before auth - [qmux 0.5.3](/quest/m0/qmux-0.5.3.md) - crates.io publishes qmux 0.5.3 with the builder session config the release backport of #5609 needs - [Release shutdown backport](/quest/m0/ffi-shutdown-backport.md) - `release` publishes a non-breaking moq-ffi with an async `drain()` and qmux sessions on the QUIC stream limit diff --git a/quest/m0/redundant-failover-drift.md b/quest/m0/redundant-failover-drift.md deleted file mode 100644 index 060a2967303..00000000000 --- a/quest/m0/redundant-failover-drift.md +++ /dev/null @@ -1,43 +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. Moved from the m1 -[load round](/quest/m1/test-flakes-2/README.md) to m0 on 2026-10-20 -(maintainer), since it fails loaded `just check` runs. 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. diff --git a/quest/m1/audit-compile-time.md b/quest/m1/audit-compile-time.md index 48384dc8233..0b6a9475a87 100644 --- a/quest/m1/audit-compile-time.md +++ b/quest/m1/audit-compile-time.md @@ -88,7 +88,8 @@ selected 29 Rust packages (moq-net and every dependent, including C, FFI, GStreamer and WASM). Phases: rustdoc 1m41s (678 hits, 76 misses, 34 bypassed), WASM clippy 18.8 s, io-uring clippy 33.5 s, test build 5m40s (524 hits, 83 misses, 5 bypassed). Nextest started 7,890 tests and stopped after 663 on the -known [failover flake](/quest/m0/redundant-failover-drift.md). +known failover flake, fixed by +[moq-dev/moq#6113](https://github.com/moq-dev/moq/pull/6113). Cargo rebuilds every dependent when an upstream crate's source changes at all, even a comment, so the test build is the bulk of the compile cost. diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 342aff67ec7..b270c59b001 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -29,9 +29,8 @@ Rules every child keeps: This README's own work, after the children: run `just check --all` several times on a loaded machine, as the first round did. -The [redundant failover drift](/quest/m0/redundant-failover-drift.md) moved -to m0 on 2026-10-20 (maintainer), since it may hide a TS export bug and fails -loaded `just check` runs. +The redundant failover drift moved to m0 on 2026-10-20 (maintainer) and is +fixed by [moq-dev/moq#6113](https://github.com/moq-dev/moq/pull/6113). Added 2026-10-20 (maintainer): three more from that day's merge batch, the draft-16 cancel race first since it fails on a mocked clock and so may hide diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index a428706054c..96e22b0a86f 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -848,9 +848,19 @@ 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()) @@ -928,7 +938,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 b1d47f5c464..3f85b3e8da0 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -48,8 +48,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 @@ -105,8 +106,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 @@ -193,7 +193,6 @@ impl Consumer { startup: start.is_none(), newest: false, max_delay, - hole: None, hole_timer: None, edge: None, start: None, @@ -377,7 +376,6 @@ impl Consumer { loop { if self.pending.front().is_none_or(|group| group.sequence <= self.current) { - self.hole = None; self.hole_timer = None; } @@ -524,10 +522,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 = 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 @@ -625,7 +623,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); @@ -653,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 { @@ -664,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); } } @@ -732,6 +733,10 @@ enum Failure { struct GroupBuffer { group: moq_net::group::Consumer, + // 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, @@ -772,6 +777,7 @@ impl GroupBuffer { fn new(group: moq_net::group::Consumer) -> Self { Self { group, + late: moq_clock::now(), index: 0, empty: true, media: false, @@ -1306,6 +1312,65 @@ 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 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)]