diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index 90461ec4f87..dcec3a61e7c 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -328,6 +328,145 @@ 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.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) }); + 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.advance(Nano(0)); + expect(request.active.peek()).toBe(active); + disposeA(); + a.stop(new StreamError(StreamCode.SessionClosed)); + 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(); + 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}`)); + 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(); + } 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 81f1f845876..6fae6a8afad 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -125,7 +125,6 @@ tracks, so per-group stats stop costing cluster announces. - [JS release preflight](/quest/m1/js-publish-preflight.md) - `release.ts` refuses a package whose `@moq/*` dependency is not on npm, so no dependent publishes uninstallable - [Stream sockets disable Nagle](/quest/m1/tcp-nodelay.md) - every qmux TCP, TLS, and WebSocket socket, accepted or dialed, sets `TCP_NODELAY` - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide -- [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 - [Rust resume checks compression](/quest/m1/resume-compress.md) - a same-epoch route copy with a different `compress` is refused instead of corrupting framing - [Rust resume backs single-route groups](/quest/m1/resume-handed-out.md) - a group first read from one route keeps the later copy that holds it, with a passthrough benchmark - [WebGPU on Safari](/quest/m1/webgpu-safari.md) - the WebGPU renderer is verified on Safari 26 for macOS and iOS diff --git a/quest/m1/replica-resume.md b/quest/m1/replica-resume.md deleted file mode 100644 index c6fed73973b..00000000000 --- 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 2d8f607f9c2..e0c3605c4d5 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,46 @@ 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!( + 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); } -/// 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`