fix(mux): redundant TS failover survives a killed importer behind bufferbloat

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();
+	}
+});

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<Milli>;
 	#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,15 +197,20 @@ 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,
 				};
 
 				// 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);

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

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.
 

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

rs/moq-cli/tests/integration/export.rs

@@ -848,9 +848,19 @@ fn redundant_feed() -> (tokio::sync::mpsc::UnboundedSender<ChildStdin>, 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<ChildStdin>, 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)