From f87bf7fa4c596e90ed6d5e30e1ee4e0ce17d5592 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 18:21:22 -0700 Subject: [PATCH 1/4] quest: claim same-epoch replica resume coverage Co-Authored-By: GPT-6 From 93f9319f408a42b7bf95183c6ca12d5846dc42d5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 18:34:02 -0700 Subject: [PATCH 2/4] test(net): cover lagging replicas on epoch wires Exercise IETF 17 and 22 beside lite-07 and mirror recovery from one original source in JS. Assert the complete resumed suffix and successor group exactly once and in order. Co-Authored-By: GPT-6 --- js/net/src/resume.test.ts | 136 +++++++++++++++++++ quest/m1/README.md | 1 - quest/m1/replica-resume.md | 34 ----- rs/moq-net/tests/integration/route_change.rs | 44 ++++-- 4 files changed, 166 insertions(+), 49 deletions(-) delete mode 100644 quest/m1/replica-resume.md diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index fe7ac399f6..e9dd56bd4c 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -220,6 +220,142 @@ test("a resumed fork shares an on-demand track request", async () => { expect(await extra).toBeUndefined(); }); +test("a lagging same-epoch replica resumes from one publisher's bytes exactly once", async () => { + const { clock, restore } = manual(); + const publisher = new Producer(); + const original = publisher.createBroadcast(path); + original.announce({ epoch: EPOCH }); + const source = original.createTrack("seq"); + const origin = new Producer(); + + // Both replicas forward this one source. Only the incumbent pauses mid-group. + function replica(lagging: boolean) { + const broadcast = new BroadcastProducer(); + const output = broadcast.createTrack("seq"); + const input = source.subscribe({ maxDelay: Milli(60_000) }); + const release = Promise.withResolvers(); + let stopped = false; + const done = (async () => { + try { + for (;;) { + const group = await input.recvGroup(); + if (!group) return; + const copy = new GroupProducer(group.sequence); + output.writeGroup(copy); + try { + for (let index = 0; ; index++) { + const frame = await group.readFrame(); + if (!frame) break; + if (lagging && group.sequence === 1 && index === 2) await release.promise; + copy.writeFrame(frame); + } + copy.close(); + } finally { + group.close(); + } + } + } catch (err) { + if (stopped) return; + output.close(err instanceof Error ? err : new Error(String(err))); + broadcast.close(); + return err; + } + })(); + return { + broadcast, + done, + stop(err?: Error) { + stopped = true; + input.close(); + release.resolve(); + output.close(err); + broadcast.close(); + }, + }; + } + + const a = replica(true); + const b = replica(false); + const disposeA = serve(origin, a.broadcast, 5n); + const request = origin.request(path); + const standby = b.broadcast.consume(); + const warm = standby.track("seq").subscribe({ maxDelay: Milli(60_000) }); + let sub: track.Subscriber | undefined; + let disposeB: (() => void) | undefined; + try { + await clock.advanceAsync(Nano(0)); + const active = request.active.peek(); + if (!active) throw new Error("expected the incumbent route"); + sub = active.track("seq").subscribe({ maxDelay: Milli(60_000) }); + const first = source.appendGroup(); + for (let frame = 0; frame < 4; frame++) first.writeFrame({ payload: enc.encode(`0.${frame}`) }); + first.close(); + const seen: string[] = []; + const initial = await sub.recvGroup(); + if (!initial) throw new Error("expected group 0"); + expect(initial.sequence).toBe(0); + for (let frame = 0; frame < 4; frame++) seen.push(dec.decode((await initial.readFrame())?.payload)); + expect(await initial.readFrame()).toBeUndefined(); + initial.close(); + + const open = source.appendGroup(); + for (let frame = 0; frame < 2; frame++) open.writeFrame({ payload: enc.encode(`1.${frame}`) }); + const interrupted = await sub.recvGroup(); + if (!interrupted) throw new Error("expected group 1"); + expect(interrupted.sequence).toBe(1); + for (let frame = 0; frame < 2; frame++) seen.push(dec.decode((await interrupted.readFrame())?.payload)); + for (let frame = 2; frame < 4; frame++) open.writeFrame({ payload: enc.encode(`1.${frame}`) }); + open.close(); + const last = source.appendGroup(); + for (let frame = 0; frame < 4; frame++) last.writeFrame({ payload: enc.encode(`2.${frame}`) }); + last.close(); + + // The standby holds the complete original while the incumbent cannot pass 1.1. + for (let sequence = 0; sequence < 3; sequence++) { + const group = await warm.recvGroup(); + if (!group) throw new Error("expected a warmed group"); + expect(group.sequence).toBe(sequence); + for (let frame = 0; frame < 4; frame++) { + expect(dec.decode((await group.readFrame())?.payload)).toBe(`${sequence}.${frame}`); + } + expect(await group.readFrame()).toBeUndefined(); + group.close(); + } + disposeB = serve(origin, b.broadcast, 9n); + await clock.advanceAsync(Nano(0)); + expect(request.active.peek()).toBe(active); + disposeA(); + a.stop(new StreamError(StreamCode.SessionClosed)); + await clock.advanceAsync(Nano(0)); + for (let frame = 2; frame < 4; frame++) seen.push(dec.decode((await interrupted.readFrame())?.payload)); + expect(await interrupted.readFrame()).toBeUndefined(); + interrupted.close(); + const successor = await sub.recvGroup(); + if (!successor) throw new Error("expected group 2"); + expect(successor.sequence).toBe(2); + for (let frame = 0; frame < 4; frame++) seen.push(dec.decode((await successor.readFrame())?.payload)); + expect(await successor.readFrame()).toBeUndefined(); + successor.close(); + expect(seen).toEqual(Array.from({ length: 12 }, (_, index) => `${Math.floor(index / 4)}.${index % 4}`)); + expect(sub.tryRecvGroup()).toBeUndefined(); + } finally { + sub?.close(); + warm.close(); + standby.close(); + request.close(); + disposeA(); + disposeB?.(); + a.stop(); + b.stop(); + original.close(); + publisher.close(); + origin.close(); + const outcomes = await Promise.all([a.done, b.done]); + restore(); + expect(outcomes).toEqual([undefined, undefined]); + } +}); + test("a subscription reads on across a same-epoch route swap, each group once", async () => { const { a, b, active, request, swap, close } = await setup(); const ta = copy(a); diff --git a/quest/m1/README.md b/quest/m1/README.md index bbf5a6796d..109552fe92 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -104,7 +104,6 @@ lite-07 stream type here. - [A widened max age re-offers cached groups](/quest/m1/resolved-demand.md) - a warm relay serves a reader with a wider max age the cached groups a cold relay would, with no wire change - [Drop set](/quest/m1/drop-set.md) - each cursor judges missing groups against its own budget, so a narrow reader drops promptly and a widening re-offers groups already reported dropped on lite-07 - [Resumed close](/quest/m1/resume-closed.md) - a resumed JS track keeps its declared end across copies, and its `closed` and `largest()` settle on a clean end -- [Replica resume](/quest/m1/replica-resume.md) - a lagging same-epoch replica resumes with every frame exactly once and in order, on lite-07 and on moq-transport - [Resumed datagrams](/quest/m1/resume-datagrams.md) - a resumed track skips only datagrams it already delivered, not every lower sequence - [Recovery verdicts](/quest/m1/recovery-verdicts.md) - group recovery refetches only on errors a new copy can fix, so a repeating error ends the group instead of looping - [Rust resume holds a declared end](/quest/m1/resume-rust-end.md) - a resumed Rust track keeps a declared end across copies, dedupes flapping copies, and judges retention inside the reader's window diff --git a/quest/m1/replica-resume.md b/quest/m1/replica-resume.md deleted file mode 100644 index c6fed73973..0000000000 --- a/quest/m1/replica-resume.md +++ /dev/null @@ -1,34 +0,0 @@ -# [S] A lagging same-epoch replica resumes on every epoch wire - -## Goal - -One publisher announced through two relays under its epoch, one of them -lagging: when the route a subscriber reads dies mid-group, the subscriber -resumes on the other route and sees every frame exactly once, in order. This -holds on lite-07 and on moq-transport with the MoQ Epoch extension (draft -17+), since epochs ride moq-transport as of -[#5932](https://github.com/moq-dev/moq/pull/5932). - -## Plan - -Decided with the maintainer on 2026-10-20. Today -`lite_resumes_after_a_lagging_route_dies` -(`rs/moq-net/tests/integration/route_change.rs`) covers the lagging case on -lite-07 only, and the route-flap tests cover moq-transport 22 without a lag. - -- Run the lagging scenario on moq-transport 17 and 22 as well. Lite carries - the resume point upstream; find what moq-transport carries instead, and fix - a duplicate or gap it shows at its cause rather than weakening the - assertion. -- A shared epoch here means true replicas of one publisher's bytes, never two - independent publishers. -- Mirror the case in `js/net` if its resume tests can drive a lagging route. - -Mocked time, wired into CI with the other integration tests. - -Public API: none. Wire: none. - -## Related - -- [Claim answer epochs](https://github.com/moq-dev/moq/pull/5678) - changed how a claim answers, so this contract is rechecked after it -- [Resume through an outage](https://github.com/moq-dev/moq/pull/6183) - landed the same-epoch resume after an unclosed route dies diff --git a/rs/moq-net/tests/integration/route_change.rs b/rs/moq-net/tests/integration/route_change.rs index 2d8f607f9c..ecac88f05c 100644 --- a/rs/moq-net/tests/integration/route_change.rs +++ b/rs/moq-net/tests/integration/route_change.rs @@ -520,7 +520,7 @@ pinned_tests! { /// partway through group 1: `A` has handed on (1, 0) and (1, 1) while `P` has already /// moved on to group 2. `B` (warmed by another reader on `W`) holds everything, so `R` /// resumes there from (1, 2). -async fn lagging_route_dies(version: Version) -> mpsc::UnboundedReceiver { +async fn lagging_route_dies(version: Version) { async fn lagged(version: Version, from: &origin::Producer, to: &origin::Producer, lag: Duration) -> MockPair { let mut options = MockConnectOptions::new(version); options.server_publish = Some(from.consume()); @@ -580,27 +580,43 @@ async fn lagging_route_dies(version: Version) -> mpsc::UnboundedReceiver().await - }); - rx + for (sequence, frame) in (2..FRAMES) + .map(|frame| (1, frame)) + .chain((0..FRAMES).map(|frame| (2, frame))) + { + let timestamp = Timestamp::from_millis(1_000 + sequence * 100 + frame).unwrap(); + assert_eq!( + next_timed(&mut rx).await, + (sequence, Some(timestamp), payload(sequence, frame)), + "{version}" + ); + } + settle().await; + assert!(rx.try_recv().is_err(), "{version}: trailing delivery"); + let _keep = (broadcast, track, group, p_a, p_b, b_w, b_r, warm, a_r, p, a, b, r, w); } -/// Lite carries the resume point upstream, so the rest of the group and the next one -/// arrive in order. #[moq_clock::test] async fn lite_resumes_after_a_lagging_route_dies() { - let version: Version = "moq-lite-07-wip".parse().unwrap(); - let mut rx = lagging_route_dies(version).await; - assert_eq!(next(&mut rx).await, (1, payload(1, 2))); - assert_eq!(next(&mut rx).await, (1, payload(1, 3))); - assert_eq!(next(&mut rx).await, (2, payload(2, 0))); + lagging_route_dies("moq-lite-07-wip".parse().unwrap()).await; +} + +#[moq_clock::test] +async fn ietf_17_resumes_after_a_lagging_route_dies() { + lagging_route_dies("moq-transport-17".parse().unwrap()).await; +} + +#[moq_clock::test] +async fn ietf_22_resumes_after_a_lagging_route_dies() { + lagging_route_dies("moq-transport-22".parse().unwrap()).await; } /// `P` publishes through the relay `O`, which `M1` and `M2` both pull from, and `E` From 886623212b5e1ee4b6aaa3644c1ef2cd606b74e1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 23:19:11 -0700 Subject: [PATCH 3/4] test(net): require the resumed subscription to stay live An empty channel or queue also matched a subscription that ended after catch-up. The source is still open, so assert the subscriber is open rather than merely idle. Co-Authored-By: Claude Opus 5.5 --- js/net/src/resume.test.ts | 3 +++ rs/moq-net/tests/integration/route_change.rs | 5 ++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index 9142945e59..a9a93e93e7 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -361,7 +361,10 @@ test("a lagging same-epoch replica resumes from one publisher's bytes exactly on expect(await successor.readFrame()).toBeUndefined(); successor.close(); expect(seen).toEqual(Array.from({ length: 12 }, (_, index) => `${Math.floor(index / 4)}.${index % 4}`)); + await clock.advanceAsync(Nano(0)); expect(sub.tryRecvGroup()).toBeUndefined(); + // The source is still open, so the resumed subscription must stay live. + expect(sub.closed.peek()).toBeUndefined(); } finally { sub?.close(); warm.close(); diff --git a/rs/moq-net/tests/integration/route_change.rs b/rs/moq-net/tests/integration/route_change.rs index ecac88f05c..e0c3605c4d 100644 --- a/rs/moq-net/tests/integration/route_change.rs +++ b/rs/moq-net/tests/integration/route_change.rs @@ -600,7 +600,10 @@ async fn lagging_route_dies(version: Version) { ); } settle().await; - assert!(rx.try_recv().is_err(), "{version}: trailing delivery"); + assert!( + matches!(rx.try_recv(), Err(mpsc::TryRecvError::Empty)), + "{version}: trailing delivery, or the subscription ended" + ); let _keep = (broadcast, track, group, p_a, p_b, b_w, b_r, warm, a_r, p, a, b, r, w); } From 86b607f7f6d25827ec718e5921e25cabcd7ae88c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 14:07:22 -0700 Subject: [PATCH 4/4] test(net): await advance in the lagging replica test after the clock rename Co-Authored-By: Claude Opus 5.5 --- js/net/src/resume.test.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index 1ba78e6b3e..834b56c178 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -310,7 +310,7 @@ test("a lagging same-epoch replica resumes from one publisher's bytes exactly on let sub: track.Subscriber | undefined; let disposeB: (() => void) | undefined; try { - await clock.advanceAsync(Nano(0)); + await clock.advance(Nano(0)); const active = request.active.peek(); if (!active) throw new Error("expected the incumbent route"); sub = active.track("seq").subscribe({ maxDelay: Milli(60_000) }); @@ -349,11 +349,11 @@ test("a lagging same-epoch replica resumes from one publisher's bytes exactly on group.close(); } disposeB = serve(origin, b.broadcast, 9n); - await clock.advanceAsync(Nano(0)); + await clock.advance(Nano(0)); expect(request.active.peek()).toBe(active); disposeA(); a.stop(new StreamError(StreamCode.SessionClosed)); - await clock.advanceAsync(Nano(0)); + await clock.advance(Nano(0)); for (let frame = 2; frame < 4; frame++) seen.push(dec.decode((await interrupted.readFrame())?.payload)); expect(await interrupted.readFrame()).toBeUndefined(); interrupted.close(); @@ -364,7 +364,7 @@ test("a lagging same-epoch replica resumes from one publisher's bytes exactly on expect(await successor.readFrame()).toBeUndefined(); successor.close(); expect(seen).toEqual(Array.from({ length: 12 }, (_, index) => `${Math.floor(index / 4)}.${index % 4}`)); - await clock.advanceAsync(Nano(0)); + await clock.advance(Nano(0)); expect(sub.tryRecvGroup()).toBeUndefined(); // The source is still open, so the resumed subscription must stay live. expect(sub.closed.peek()).toBeUndefined();