From 5bf7e0b86943be8a380a1a822afb2d1303b3881c Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Wed, 7 Oct 2026 12:35:39 +0200 Subject: [PATCH 1/3] fix(net): a lite subscribe that timed out stays unregistered Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/subscriber.test.ts | 39 ++++++++++++++++++++++++++++-- js/net/src/lite/subscriber.ts | 11 ++++++--- 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index ea5ff91d91..ac8b39e9c9 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -1,13 +1,14 @@ -import { expect, spyOn, test } from "bun:test"; +import { expect, jest, spyOn, test } from "bun:test"; import { Signal } from "@moq/signals"; import type { Probe as ProbeStats } from "../connection/stats.ts"; import * as Epoch from "../epoch.ts"; import { error, fromTransport, reason, StreamCode, StreamError } from "../error.ts"; import { HopSchema, isAnonymous, MAX_HOPS, Route, UNKNOWN_HOP } from "../hop.ts"; import * as Path from "../path.ts"; -import { Writer } from "../stream.ts"; +import { type Reader, Writer } from "../stream.ts"; import * as Time from "../time.ts"; import { type AnnounceBroadcast, AnnounceInit, AnnounceOk, encodeAnnounceBroadcast } from "./announce.ts"; +import { Group } from "./group.ts"; import { Probe } from "./probe.ts"; import { Subscriber } from "./subscriber.ts"; import { TrackInfo } from "./track.ts"; @@ -882,6 +883,40 @@ test.each([ await stuck.aborted; }); +// A setup that outlived its deadline is over: TRACK_INFO arriving afterwards must neither register +// the subscription again nor send its SUBSCRIBE. +test("a subscribe that timed out waiting on TRACK_INFO stays unregistered when it arrives", async () => { + jest.useFakeTimers(); + const warn = spyOn(console, "warn").mockImplementation(() => {}); + const { quic, streams } = fakeSession(); + const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); + try { + const track = subscriber.consume(Path.from("room")).track("video").subscribe(); + await drainUntil(() => streams.length === 1); + await streams[0].reading; + + jest.advanceTimersByTime(10_000); + await drainUntil(() => track.closed.peek() !== undefined); + await answerTrackInfo(streams[0]); + for (let i = 0; i < MAX_DRAIN_TURNS; i++) await Promise.resolve(); + expect(streams.length).toBe(1); + + // A GROUP for a registered subscription is read; one for a forgotten id is ignored. + let read = false; + const reader = { + stop: () => { + read = true; + }, + } as unknown as Reader; + await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 0 }), reader); + expect(read).toBe(false); + } finally { + subscriber.close(); + warn.mockRestore(); + jest.useRealTimers(); + } +}); + test("a fetch started after the subscriber closes rejects without opening a stream", async () => { const { quic, streams } = fakeSession(); const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index eb2bae2d21..3c99512a87 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -540,8 +540,9 @@ export class Subscriber { }); // Open the stream under a timeout. The stream handle flows back via `state` - // so the timeout path can abort it if it finishes opening after the deadline. - const state: { stream?: Stream } = {}; + // so the timeout path can abort it if it finishes opening after the deadline, + // and `cancelled` stops a setup that is still running once the deadline passed. + const state: { stream?: Stream; cancelled?: boolean } = {}; const setup = this.#openSubscribe(state, msg, request, id, timescale); let opened: { stream: Stream; entry: SubscribeEntry }; @@ -556,6 +557,7 @@ export class Subscriber { // The setup outlived its deadline waiting for the first response: a control // timeout, not content that arrived late. const e = err instanceof TimeoutError ? controlTimeout(err) : await sessionCause(this.#quic, err); + state.cancelled = true; request.reject(e); this.#subscribes.delete(id); console.warn(`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${reason(e)}`); @@ -640,7 +642,7 @@ export class Subscriber { // SUBSCRIBE is accepted implicitly (no SUBSCRIBE_OK). Older drafts carry no // per-track properties, so they resolve to defaults and just drain SUBSCRIBE_OK. async #openSubscribe( - state: { stream?: Stream }, + state: { stream?: Stream; cancelled?: boolean }, msg: Subscribe, request: track.Request, id: bigint, @@ -652,6 +654,9 @@ export class Subscriber { if (supportsTrackStream(this.version)) { // Fetch the immutable properties once via the TRACK stream. const info = await this.#trackInfo(msg.broadcast, msg.epoch, msg.track); + // The deadline passed while TRACK_INFO was pending: the request is already rejected, + // so don't register it again or send its SUBSCRIBE. + if (state.cancelled) throw new Error("subscribe cancelled before it was sent"); producer = request.accept(this.#toModelInfo(info)); timescale.set(info.timescale); } else { From 40b5664c5df6f88ba36aa369b978ddf3671ba441 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Wed, 7 Oct 2026 13:17:26 +0200 Subject: [PATCH 2/3] fix(net): reset a timed-out lite subscribe's TRACK stream Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/subscriber.test.ts | 46 +++++++++++++++---------- js/net/src/lite/subscriber.ts | 54 ++++++++++++++++++------------ 2 files changed, 61 insertions(+), 39 deletions(-) diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index ac8b39e9c9..5409be5f43 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -10,7 +10,7 @@ import * as Time from "../time.ts"; import { type AnnounceBroadcast, AnnounceInit, AnnounceOk, encodeAnnounceBroadcast } from "./announce.ts"; import { Group } from "./group.ts"; import { Probe } from "./probe.ts"; -import { Subscriber } from "./subscriber.ts"; +import { SUBSCRIBE_SETUP_TIMEOUT_MS, Subscriber } from "./subscriber.ts"; import { TrackInfo } from "./track.ts"; import { Version } from "./version.ts"; @@ -883,33 +883,45 @@ test.each([ await stuck.aborted; }); -// A setup that outlived its deadline is over: TRACK_INFO arriving afterwards must neither register -// the subscription again nor send its SUBSCRIBE. -test("a subscribe that timed out waiting on TRACK_INFO stays unregistered when it arrives", async () => { +// A setup that outlived its deadline is over: its TRACK stream is reset, even one still waiting for a +// slot, and the subscription is neither registered again nor sent as a SUBSCRIBE. +test.each([ + ["lite-05 subscribe waiting on the TRACK_INFO", Version.DRAFT_05, false], + ["lite-06 subscribe waiting on the TRACK_INFO", Version.DRAFT_06, false], + ["lite-07 subscribe waiting on the TRACK_INFO", Version.DRAFT_07, false], + ["lite-05 subscribe waiting on a stream slot for the TRACK", Version.DRAFT_05, true], +] as const)("a %s that times out leaves nothing behind", async (_, version, parked) => { jest.useFakeTimers(); const warn = spyOn(console, "warn").mockImplementation(() => {}); - const { quic, streams } = fakeSession(); - const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); + const { quic, streams } = fakeSession(parked ? [0] : []); + const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); try { const track = subscriber.consume(Path.from("room")).track("video").subscribe(); await drainUntil(() => streams.length === 1); - await streams[0].reading; + if (!parked) await streams[0].reading; - jest.advanceTimersByTime(10_000); + jest.advanceTimersByTime(SUBSCRIBE_SETUP_TIMEOUT_MS); await drainUntil(() => track.closed.peek() !== undefined); - await answerTrackInfo(streams[0]); - for (let i = 0; i < MAX_DRAIN_TURNS; i++) await Promise.resolve(); + + let aborted = false; + void streams[0].aborted.then(() => { + aborted = true; + }); + if (parked) streams[0].release(); + await drainUntil(() => aborted); + expect(aborted).toBe(true); expect(streams.length).toBe(1); - // A GROUP for a registered subscription is read; one for a forgotten id is ignored. - let read = false; - const reader = { - stop: () => { - read = true; + // A GROUP for a forgotten id is ignored without touching its stream. + const touched: PropertyKey[] = []; + const reader = new Proxy({} as Reader, { + get: (_, key) => { + touched.push(key); + return () => {}; }, - } as unknown as Reader; + }); await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 0 }), reader); - expect(read).toBe(false); + expect(touched).toEqual([]); } finally { subscriber.close(); warn.mockRestore(); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 3c99512a87..8d9f216232 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -57,7 +57,7 @@ import { // drafts, or TRACK_INFO on lite-05+) may take. Browsers cap concurrent QUIC streams // (Chrome ~100) and we open with waitUntilAvailable, so past the cap the open blocks // until the peer frees a slot. The timeout turns a stall into a clear error. -const SUBSCRIBE_SETUP_TIMEOUT_MS = 10_000; +export const SUBSCRIBE_SETUP_TIMEOUT_MS = 10_000; // The TRACK stream and implicit SUBSCRIBE acceptance are lite-05+. function supportsTrackStream(version: Version): boolean { @@ -541,8 +541,9 @@ export class Subscriber { // Open the stream under a timeout. The stream handle flows back via `state` // so the timeout path can abort it if it finishes opening after the deadline, - // and `cancelled` stops a setup that is still running once the deadline passed. - const state: { stream?: Stream; cancelled?: boolean } = {}; + // and `cancel` ends a setup still running once the deadline passed, resetting + // its TRACK stream so a peer that never answers can't hold one per attempt. + const state: { stream?: Stream; cancel: AbortController } = { cancel: new AbortController() }; const setup = this.#openSubscribe(state, msg, request, id, timescale); let opened: { stream: Stream; entry: SubscribeEntry }; @@ -557,7 +558,7 @@ export class Subscriber { // The setup outlived its deadline waiting for the first response: a control // timeout, not content that arrived late. const e = err instanceof TimeoutError ? controlTimeout(err) : await sessionCause(this.#quic, err); - state.cancelled = true; + state.cancel.abort(e); request.reject(e); this.#subscribes.delete(id); console.warn(`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${reason(e)}`); @@ -642,7 +643,7 @@ export class Subscriber { // SUBSCRIBE is accepted implicitly (no SUBSCRIBE_OK). Older drafts carry no // per-track properties, so they resolve to defaults and just drain SUBSCRIBE_OK. async #openSubscribe( - state: { stream?: Stream; cancelled?: boolean }, + state: { stream?: Stream; cancel: AbortController }, msg: Subscribe, request: track.Request, id: bigint, @@ -653,10 +654,10 @@ export class Subscriber { if (supportsTrackStream(this.version)) { // Fetch the immutable properties once via the TRACK stream. - const info = await this.#trackInfo(msg.broadcast, msg.epoch, msg.track); - // The deadline passed while TRACK_INFO was pending: the request is already rejected, - // so don't register it again or send its SUBSCRIBE. - if (state.cancelled) throw new Error("subscribe cancelled before it was sent"); + const info = await this.#trackInfo(msg.broadcast, msg.epoch, msg.track, state.cancel.signal); + // The deadline passed as TRACK_INFO landed: the request is already rejected, so don't + // register it again or send its SUBSCRIBE. + state.cancel.signal.throwIfAborted(); producer = request.accept(this.#toModelInfo(info)); timescale.set(info.timescale); } else { @@ -699,22 +700,31 @@ export class Subscriber { } // Opens a TRACK stream, reads the single TRACK_INFO, and FINs. Lite-05+ only. - async #trackInfo(broadcast: Path.Valid, epoch: Epoch.Valid | undefined, track: string): Promise { - return this.#exchange({ version: this.version }, async (stream) => { - await stream.writer.u53(StreamId.Track); - await new TrackMessage(broadcast, track, epoch).encode(stream.writer, this.version); - const info = await TrackInfo.decode(stream.reader, this.version); - // The publisher FINs after TRACK_INFO; FIN our side too. - stream.close(); - return info; - }); + async #trackInfo( + broadcast: Path.Valid, + epoch: Epoch.Valid | undefined, + track: string, + signal?: AbortSignal, + ): Promise { + return this.#exchange( + { version: this.version }, + async (stream) => { + await stream.writer.u53(StreamId.Track); + await new TrackMessage(broadcast, track, epoch).encode(stream.writer, this.version); + const info = await TrackInfo.decode(stream.reader, this.version); + // The publisher FINs after TRACK_INFO; FIN our side too. + stream.close(); + return info; + }, + signal, + ); } // Opens a stream and runs a request/response exchange on it, resetting the stream if `run` - // fails. Subscriber.close() also resets it while `run` is pending, so a peer that never - // answers cannot hold it open, and a stream that opens after the close is reset at once. - async #exchange(options: OpenOptions, run: (stream: Stream) => Promise): Promise { - const closed = this.#closed.signal; + // fails. Subscriber.close() or `signal` also resets it while `run` is pending, so a peer that + // never answers cannot hold it open, and a stream that opens after either is reset at once. + async #exchange(options: OpenOptions, run: (stream: Stream) => Promise, signal?: AbortSignal): Promise { + const closed = signal ? AbortSignal.any([this.#closed.signal, signal]) : this.#closed.signal; closed.throwIfAborted(); const stream = await Stream.open(this.#quic, options); const abort = () => stream.abort(error(closed.reason)); From de87d033d730206e702c5c00f9ff2b14015b8ae6 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Wed, 7 Oct 2026 13:25:28 +0200 Subject: [PATCH 3/3] fix(net): send no SUBSCRIBE on a stream that opens after the deadline Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/subscriber.test.ts | 44 ++++++++++++++++++++++++++++-- js/net/src/lite/subscriber.ts | 3 ++ 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 5409be5f43..1df08f571b 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -786,6 +786,8 @@ interface FakeStream { // Resolves once the subscriber waits on a read the test has not answered. reading: Promise; aborted: Promise; + // Every chunk the subscriber wrote. + written: Uint8Array[]; // Hands the stream to the subscriber, for an open the session was told to park. release: () => void; } @@ -812,10 +814,14 @@ function fakeSession(park: number[] = []) { }, { highWaterMark: 0 }, ); - const writable = new WritableStream({ abort: (reason) => void onAbort(reason) }); + const written: Uint8Array[] = []; + const writable = new WritableStream({ + write: (chunk) => void written.push(chunk), + abort: (reason) => void onAbort(reason), + }); const opened = new Promise((resolve) => (release = () => resolve({ readable, writable }))); if (!park.includes(streams.length)) release(); - streams.push({ inbound, reading, aborted, release }); + streams.push({ inbound, reading, aborted, written, release }); return opened; }, } as unknown as WebTransport; @@ -929,6 +935,40 @@ test.each([ } }); +// The SUBSCRIBE stream can open after the deadline too; it is reset without carrying a SUBSCRIBE. +test("a lite subscribe that times out waiting on a stream slot for the SUBSCRIBE sends nothing on it", async () => { + jest.useFakeTimers(); + const warn = spyOn(console, "warn").mockImplementation(() => {}); + const { quic, streams } = fakeSession([1]); + const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); + try { + const track = subscriber.consume(Path.from("room")).track("video").subscribe(); + await drainUntil(() => streams.length === 1); + await streams[0].reading; + // TRACK_INFO lands halfway, so the SUBSCRIBE open's own deadline is still ahead when the + // setup deadline fires. + jest.advanceTimersByTime(SUBSCRIBE_SETUP_TIMEOUT_MS / 2); + await answerTrackInfo(streams[0]); + await drainUntil(() => streams.length === 2); + + jest.advanceTimersByTime(SUBSCRIBE_SETUP_TIMEOUT_MS / 2); + await drainUntil(() => track.closed.peek() !== undefined); + + let aborted = false; + void streams[1].aborted.then(() => { + aborted = true; + }); + streams[1].release(); + await drainUntil(() => aborted); + expect(aborted).toBe(true); + expect(streams[1].written).toEqual([]); + } finally { + subscriber.close(); + warn.mockRestore(); + jest.useRealTimers(); + } +}); + test("a fetch started after the subscriber closes rejects without opening a stream", async () => { const { quic, streams } = fakeSession(); const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 8d9f216232..dbf1fc2263 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -685,6 +685,9 @@ export class Subscriber { this.#subscribes.set(id, entry); state.stream = await Stream.open(this.#quic, { version: this.version }); + // The deadline passed while the open waited: the late-setup handler resets the stream, so + // don't send a SUBSCRIBE on it first. + state.cancel.signal.throwIfAborted(); await state.stream.writer.u53(StreamId.Subscribe); await msg.encode(state.stream.writer, this.version);