From 498ccf2c2aa419426770dd2531b320cc458aeb97 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 04:13:22 -0800 Subject: [PATCH 1/3] chore(quest): claim quest/m1/join-tests-draft20.md Quest-Claim: 66c5a919dfd234e8 From 4e39364a0bfef722b9c35a0788f2e0e69c5e6923 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 04:15:04 -0800 Subject: [PATCH 2/3] test(net): cover draft 20 joins with the supported version matrix Co-Authored-By: GPT-6.1 Sol --- js/net/src/ietf/join.test.ts | 116 ++++++++++++++++++--------------- quest/m1/README.md | 1 - quest/m1/join-tests-draft20.md | 19 ------ 3 files changed, 65 insertions(+), 71 deletions(-) delete mode 100644 quest/m1/join-tests-draft20.md diff --git a/js/net/src/ietf/join.test.ts b/js/net/src/ietf/join.test.ts index 40da0f6e939..72b9b5f3547 100644 --- a/js/net/src/ietf/join.test.ts +++ b/js/net/src/ietf/join.test.ts @@ -19,10 +19,15 @@ import { RequestError } from "./request.ts"; import { NativeSession } from "./session.ts"; import { Subscribe, SubscribeOk } from "./subscribe.ts"; import { Subscriber } from "./subscriber.ts"; -import { ALPN, type IetfVersion, Version } from "./version.ts"; +import { ALPN, type IetfVersion, Version, versionName } from "./version.ts"; +const VERSIONS = [Version.DRAFT_18, Version.DRAFT_20, Version.DRAFT_22] as const; const VERSION = Version.DRAFT_18; -const DRAFT_22 = Version.DRAFT_22; +const ALPNS: Record = { + [Version.DRAFT_18]: ALPN.DRAFT_18, + [Version.DRAFT_20]: ALPN.DRAFT_20, + [Version.DRAFT_22]: ALPN.DRAFT_22, +}; const ALIAS = 9n; /** The group SUBSCRIBE_OK names as Largest in most cases below. */ const GROUP = 3; @@ -52,7 +57,7 @@ async function nextStream( /** A subscriber with one track, and the SUBSCRIBE it sent, still unanswered. */ async function subscribed(version: IetfVersion = VERSION) { - const pair = createMockTransportPair(version === VERSION ? ALPN.DRAFT_18 : ALPN.DRAFT_22); + const pair = createMockTransportPair(ALPNS[version]); const session = new NativeSession(pair.server, version, true); const subscriber = new Subscriber({ session }); const consumer = subscriber.consume(Path.from("room")).track("video").subscribe(); @@ -166,6 +171,7 @@ async function frames(group: { readFrame(): Promise<{ payload: Uint8Array } | un test("a draft-18 join subscribes at Largest Object and fetches Largest's group", async () => { const { subscribe, fetch } = await join(); expect(subscribe.filter).toEqual({ kind: "nextObject" }); + expect(subscribe.fill).toBeUndefined(); expect(fetch?.type).toEqual({ kind: "relativeJoining", subscription: subscribe.requestId, offset: 0n }); expect(fetch?.groupOrder).toBe("ascending"); expect(fetch?.requestId).not.toBe(subscribe.requestId); @@ -453,56 +459,65 @@ test("a joining FETCH spanning another group drops the head", async () => { expect(await group?.closed).toBeInstanceOf(Error); }); -test("a draft-22 join subscribes at Next Object with a StartGroup=1 fill, and sends no FETCH", async () => { - const { subscribe, answer, client } = await subscribed(DRAFT_22); - expect(subscribe.filter).toEqual({ kind: "nextObject" }); - expect(subscribe.fill).toEqual({ filter: { kind: "relative", groups: 1n }, rangeFilters: false }); - await answer({ groupId: BigInt(GROUP), objectId: 1n }); - expect(await nextStream(client, 200, DRAFT_22)).toBeUndefined(); -}); +// Drafts 20 and 22 fill the current group; only draft 18 sends a joining FETCH. +for (const version of VERSIONS) { + if (version === VERSION) continue; -test("a draft-22 fill and its tail stitch into one group", async () => { - const { subscriber, consumer, subscribe, answer } = await subscribed(DRAFT_22); - await answer({ groupId: BigInt(GROUP), objectId: 1n }); + test(`${versionName(version)}: a join subscribes at Next Object with a StartGroup=1 fill, and sends no FETCH`, async () => { + const { subscribe, answer, client } = await subscribed(version); + expect(subscribe.filter).toEqual({ kind: "nextObject" }); + expect(subscribe.fill).toEqual({ filter: { kind: "relative", groups: 1n }, rangeFilters: false }); + await answer({ groupId: BigInt(GROUP), objectId: 1n }); + expect(await nextStream(client, 200, version)).toBeUndefined(); + }); - const tail = subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"], DRAFT_22)); - // The fill names the SUBSCRIBE's own request. - await subscriber.handleFetch(await head(subscribe.requestId, ["h0", "h1"], GROUP, DRAFT_22)); - await tail; + test(`${versionName(version)}: a fill and its tail stitch into one group`, async () => { + const { subscriber, consumer, subscribe, answer } = await subscribed(version); + await answer({ groupId: BigInt(GROUP), objectId: 1n }); - const group = await consumer.ordered().nextGroup(); - expect(group?.sequence).toBe(GROUP); - expect(await frames(group)).toEqual(["h0", "h1", "t2"]); -}); + const tail = subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"], version)); + // The fill names the SUBSCRIBE's own request. + await subscriber.handleFetch(await head(subscribe.requestId, ["h0", "h1"], GROUP, version)); + await tail; -test("a draft-22 fill that overtakes SUBSCRIBE_OK waits for it", async () => { - const { subscriber, consumer, subscribe, answer } = await subscribed(DRAFT_22); - const fill = subscriber.handleFetch(await head(subscribe.requestId, ["h0", "h1"], GROUP, DRAFT_22)); - await answer({ groupId: BigInt(GROUP), objectId: 1n }); - await fill; - await subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"], DRAFT_22)); + const group = await consumer.ordered().nextGroup(); + expect(group?.sequence).toBe(GROUP); + expect(await frames(group)).toEqual(["h0", "h1", "t2"]); + }); - const group = await consumer.ordered().nextGroup(); - expect(await frames(group)).toEqual(["h0", "h1", "t2"]); -}); + test(`${versionName(version)}: a fill that overtakes SUBSCRIBE_OK waits for it`, async () => { + const { subscriber, consumer, subscribe, answer } = await subscribed(version); + const fill = subscriber.handleFetch(await head(subscribe.requestId, ["h0", "h1"], GROUP, version)); + await answer({ groupId: BigInt(GROUP), objectId: 1n }); + await fill; + await subscriber.handleGroup(...liveStream(GROUP, 2, ["t2"], version)); -test("a draft-22 fill counts toward PUBLISH_DONE's Stream Count", async () => { - const { subscriber, consumer, subscribe, peer, answer } = await subscribed(DRAFT_22); - await answer({ groupId: 0n, objectId: 0n }); - const started = now(); - await subscriber.handleFetch(await head(subscribe.requestId, ["catalog"], 0, DRAFT_22)); - await peer.writer.u53(PublishDone.id); - const done = new PublishDone({ statusCode: PublishDoneStatus.TRACK_ENDED, streamCount: 1n, reasonPhrase: "done" }); - await done.encode(peer.writer, DRAFT_22); + const group = await consumer.ordered().nextGroup(); + expect(await frames(group)).toEqual(["h0", "h1", "t2"]); + }); - // The fill was the one stream counted, so the head is the whole group and the track ends. - const group = await consumer.recvGroup(); - expect(group?.sequence).toBe(0); - expect(await frames(group)).toEqual(["catalog"]); - expect(await consumer.recvGroup()).toBeUndefined(); - expect(await consumer.closed).toBeNull(); - expect(Nano.toMilli(Instant.durationSince(now(), started))).toBeLessThan(GRACE); -}); + test(`${versionName(version)}: a fill counts toward PUBLISH_DONE's Stream Count`, async () => { + const { subscriber, consumer, subscribe, peer, answer } = await subscribed(version); + await answer({ groupId: 0n, objectId: 0n }); + const started = now(); + await subscriber.handleFetch(await head(subscribe.requestId, ["catalog"], 0, version)); + await peer.writer.u53(PublishDone.id); + const done = new PublishDone({ + statusCode: PublishDoneStatus.TRACK_ENDED, + streamCount: 1n, + reasonPhrase: "done", + }); + await done.encode(peer.writer, version); + + // The fill was the one stream counted, so the head is the whole group and the track ends. + const group = await consumer.recvGroup(); + expect(group?.sequence).toBe(0); + expect(await frames(group)).toEqual(["catalog"]); + expect(await consumer.recvGroup()).toBeUndefined(); + expect(await consumer.closed).toBeNull(); + expect(Nano.toMilli(Instant.durationSince(now(), started))).toBeLessThan(GRACE); + }); +} test("a clean end finishes a head nothing claimed", () => { const t = new Tail(); @@ -544,7 +559,7 @@ test("fetch objects decode as FetchFrame wrote them", async () => { // serves a fill split from the subscription at Largest, as a strict relay does. async function publishing(version: IetfVersion) { const origin = new OriginProducer(); - const pair = createMockTransportPair(version === VERSION ? ALPN.DRAFT_18 : ALPN.DRAFT_22); + const pair = createMockTransportPair(ALPNS[version]); const url = new URL("https://localhost:4443/test"); const [client, server] = await Promise.all([ Connection.connect({ url, transport: pair.client }), @@ -571,9 +586,8 @@ async function publishing(version: IetfVersion) { return { remote: wireOf(client).consume(Path.from("room")), group, sent }; } -for (const version of [VERSION, DRAFT_22]) { - const draft = version === VERSION ? "draft 18" : "draft 22"; - test(`${draft}: a late JS subscriber gets a JS publisher's current groups from object 0`, async () => { +for (const version of VERSIONS) { + test(`${versionName(version)}: a late JS subscriber gets a JS publisher's current groups from object 0`, async () => { const { remote, group, sent } = await publishing(version); const catalogGroup = await remote.track("catalog.json").subscribe().recvGroup(); expect(catalogGroup?.sequence).toBe(0); @@ -584,7 +598,7 @@ for (const version of [VERSION, DRAFT_22]) { group.writeFrame(textFrame("v2")); group.finish(); expect(await frames(videoGroup)).toEqual(["v0", "v1", "v2"]); - // A joining FETCH per track on draft 18; draft 22 asks for the head with a fill. + // Draft 18 sends a joining FETCH per track; drafts 20 and 22 request a fill. expect(sent.mock.calls).toHaveLength(version === VERSION ? 2 : 0); }); } diff --git a/quest/m1/README.md b/quest/m1/README.md index 38b62fe1570..10c46b93117 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -171,7 +171,6 @@ removed). - [Next Object on a cold Rust copy](/quest/m1/next-object-cold-copy.md) - a join after an upstream group report starts at the first frame received, never skipping - [Legacy lite close drain](/quest/m1/lite-legacy-close-drain.md) - a lite-01..06 graceful close never cuts a final group the subscriber has not read - [moqsrc max-delay](/quest/m1/gst-max-delay.md) - GStreamer's `moqsrc` takes a `max-delay` property instead of a hard-coded 1 s budget -- [JS join tests cover draft 20](/quest/m1/join-tests-draft20.md) - join.test.ts runs on every supported draft, not just 18 and 22 - [Poll-first async](/quest/m1/poll-first/README.md) - library crates compile no coroutines by default: async wrappers are generated from `poll_*` and gated behind an `async` feature - [Binding announcements always carry captures](/quest/m1/ffi-required-captures.md) - moq-ffi and its generated bindings make `captures` a required list, as #7266 always sets it - [HLS export follows a resolution change](/quest/m1/hls-reinit-same-codec.md) - a decoder-identity change under the same codec string starts a new init and segment tag diff --git a/quest/m1/join-tests-draft20.md b/quest/m1/join-tests-draft20.md deleted file mode 100644 index e5ae1a7f324..00000000000 --- a/quest/m1/join-tests-draft20.md +++ /dev/null @@ -1,19 +0,0 @@ -# [XS] JS join tests cover draft 20 - -## Goal - -`js/net/src/ietf/join.test.ts` runs its join cases on every moq-transport -draft @moq/net supports, including draft 20, instead of hard-coding drafts 18 -and 22. - -## Plan - -[moq-dev/moq#7190](https://github.com/moq-dev/moq/pull/7190) restored draft 20 -and added it to `fetch.test.ts`, but `join.test.ts` still names drafts 18 and -22. Iterate its cases over the supported versions, as `fetch.test.ts` does, -with each draft's own expectation: draft 18 joins with a relative joining -FETCH and draft 22 with a fill, so confirm which path draft 20 takes first. - -Decided 2026-10-30 (maintainer, /quest-plan of the 2026-10-30 /quest-complete follow-ups): ✅ new m1 XS quest / drop. - -Public API: none. Wire: none. From 338fa80916488de716c7663c51939dd383fb4544 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 07:01:32 -0800 Subject: [PATCH 3/3] test(net): buffer peer-close fixture streams before acceptance Collect incoming streams independently of the connection accept budget so the mock cannot discard finished groups before the fixture buffers them. Exercise every drain case with yielding accepts. Co-Authored-By: GPT-6.1 Sol --- js/net/src/lite/connection.test.ts | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 90d848a32c7..d4d343eaf55 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -7,7 +7,7 @@ import { SessionCode, SessionError } from "../error.ts"; import { createMockTransportPair, type MockTransport, textFrame } from "../mock.ts"; import { Producer } from "../origin.ts"; import * as Path from "../path.ts"; -import { Stream, Writer } from "../stream.ts"; +import { acceptBudget, Stream, Writer } from "../stream.ts"; import { Timescale } from "../time.ts"; import { wireOf } from "../wire.ts"; import { Connection, probeLevel } from "./connection.ts"; @@ -214,14 +214,15 @@ function holdFins(transport: MockTransport, streams: "uni" | "incoming-bidi") { }; } -// Hand `transport` each incoming unidirectional stream only once its FIN arrived, buffered outside -// the transport so a session close cannot discard it. QMux keeps a finished stream's unread bytes -// across a close this way, so the reads still owed land after the close itself. +// Buffer incoming unidirectional streams outside the transport, independently of when the +// connection accepts them: its accept budget may yield until after the session closes. QMux +// keeps a finished stream's unread bytes across a close, so the reads still owed land afterwards. function deliverWhole(transport: MockTransport) { + const incoming = transport.incomingUnidirectionalStreams; Object.defineProperty(transport, "incomingUnidirectionalStreams", { - value: transport.incomingUnidirectionalStreams.pipeThrough( - new TransformStream, ReadableStream>({ - transform(stream, controller) { + value: new ReadableStream>({ + async start(controller) { + for await (const stream of incoming) { controller.enqueue( new ReadableStream({ async start(whole) { @@ -231,9 +232,10 @@ function deliverWhole(transport: MockTransport) { }, }), ); - }, - }), - ), + } + controller.close(); + }, + }), }); } @@ -357,6 +359,8 @@ for (const [alpn, version] of [ // SUBSCRIBE FIN it sent first; the closing session reads them. for (const overtaken of hasTrackStream(version) ? ["group"] : ["group", "fin"]) { test(`a peer close drains the overtaken ${overtaken} over ${versionName(version)}`, async () => { + // Exercise a close that overtakes acceptance, regardless of the host CPU budget. + const yielding = spyOn(acceptBudget, "poll").mockImplementation(() => yieldNow()); const pair = createMockTransportPair(alpn); deliverWhole(pair.client); if (overtaken === "fin") finAfterClose(pair.client); @@ -395,6 +399,7 @@ for (const [alpn, version] of [ expect(await group?.readString()).toBeUndefined(); expect(await track.nextGroup()).toBeUndefined(); } finally { + yielding.mockRestore(); track.close(); remote.close(); client.abort();