diff --git a/js/net/src/ietf/join.test.ts b/js/net/src/ietf/join.test.ts index def2ff01ff9..8a9f23aed27 100644 --- a/js/net/src/ietf/join.test.ts +++ b/js/net/src/ietf/join.test.ts @@ -20,10 +20,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; @@ -53,7 +58,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(); @@ -167,6 +172,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); @@ -468,56 +474,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(); @@ -559,7 +574,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 }), @@ -586,9 +601,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); @@ -599,7 +613,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/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(); diff --git a/quest/m1/README.md b/quest/m1/README.md index a9ed35f1c1f..96017b5e0bd 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -181,7 +181,6 @@ per-frame CPU growth since #5926. - [Unbounded groups](/quest/m1/unbounded-groups/README.md) - groups and frames lose their size caps; a group keeps a 32 MB tail, and readers behind it replay through paced flights or refresh - [A follower is told when the serving epochless route leaves](/quest/m1/epochless-leave-restart.md) - wire-level tests pin the restart a follower gets when the serving epochless route is withdrawn or dies while another remains - [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 -- [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 - [Authored tracks default to a 30 s max-age](/quest/m1/max-age-defaults.md) - raw and media tracks advertise 30 s unless their author chooses otherwise, with explicit unlimited in every binding - [JS ordered reads wait only on owed groups](/quest/m1/js-ordered-owed.md) - @moq/net matches Rust's owed-group wait 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.