From dd947b9f137988855dcbf1d33aceaf3f676e82a2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 22:39:44 -0700 Subject: [PATCH 1/4] chore(quest): claim quest/m1/ietf-lazy-discovery.md Quest-Claim: 27600c2eaa4bf564 From c1da6139fc6fab57e52281371d54b46fad8d994a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 22:55:24 -0700 Subject: [PATCH 2/4] fix(js): never hold overlapping SUBSCRIBE_NAMESPACEs at a foreign peer A peer that declared no MoQ Solicit may enforce the drafts' PREFIX_OVERLAP rule, as moxygen does, and refuse a broader watch held beside a narrower one, which turned the JS session blind. Mirror Rust (moq-dev/moq#6530): - forward.ts runs exclusive discovery at such a peer: a broader request closes the narrower ones beneath it, watched or lingering. - The ietf subscriber holds a request until every overlapping request already closed is off the wire: UNSUBSCRIBE_NAMESPACE written on drafts 14/15, the peer's half ended from draft 16 (FIN or CANCELLED reset, then drain). - A closed reader stays registered while its request drains, so no announcement is appended to it. Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 2 +- js/net/bench/forward.ts | 1 + js/net/src/connection/forward.test.ts | 46 +++++++++- js/net/src/connection/forward.ts | 17 ++++ js/net/src/ietf/connection.ts | 3 + js/net/src/ietf/request_stream.ts | 12 ++- js/net/src/ietf/subscriber.test.ts | 80 ++++++++++++++++- js/net/src/ietf/subscriber.ts | 121 +++++++++++++++++--------- js/net/src/lite/connection.ts | 1 + js/net/src/wire.ts | 5 ++ quest/m1/README.md | 1 - quest/m1/ietf-lazy-discovery.md | 41 --------- quest/m1/ietf-request-cancel.md | 8 +- 13 files changed, 247 insertions(+), 91 deletions(-) delete mode 100644 quest/m1/ietf-lazy-discovery.md diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 477dc62bda5..7906dd485a1 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -25,7 +25,7 @@ maps everything else to "not supported" or a harmless equivalent. The [moq-lite page](/concept/moq-lite#what-moq-lite-leaves-out) lists what the subset drops. What a peer actually observes against this implementation: -- **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. Rust never holds two overlapping requests at a peer that did not declare solicit: a broader one replaces the narrower ones once they are closed. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below): on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session. +- **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. We never hold two overlapping requests at a peer that did not declare solicit: a broader one replaces the narrower ones once they are closed. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below): on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session. - **`SUBSCRIBE_TRACKS` lists, it never pushes.** On draft 18 and later, it lists every track of every broadcast under its prefix, later ones included. With `FORWARD=0`, each track gets a paused `PUBLISH` carrying its Track Properties, and a `PUBLISH_DONE` once the track is no longer listed: `TRACK_ENDED` if it finished, even just before its broadcast ended, and an error otherwise, since the track may go on elsewhere. A `REQUEST_UPDATE` with `FORWARD=1` resumes one, served from the cache like a `SUBSCRIBE` with the update's filter and answered with `LARGEST_OBJECT`; `FORWARD=0` pauses it again. Each paused `PUBLISH` opens only as the peer's stream credit admits it, so a peer that grants none holds none. A listing that omits `FORWARD` or sets it to 1 asks for every track's objects: from draft 19 each track gets `PUBLISH_SKIPPED`, and the subscriber `SUBSCRIBE`s to what it wants; draft 18 has no such skip, so the request is refused with `NOT_SUPPORTED`. A prefix overlapping another listing gets `PREFIX_OVERLAP`, a `TRACK_PROPERTY_FILTER` gets `NOT_SUPPORTED`, and a `REQUEST_UPDATE` on the listing itself ends it. On draft 18, a `SUBSCRIBE` to a track we `PUBLISH`ed gets `DUPLICATE_SUBSCRIPTION`. Cancelling the listing stops new `PUBLISH`es, while those sent live on. A broadcast is listed only while our publish grant covers it, from scratch once it does. Rust sends it upstream for a listing (below); JavaScript does not. - **A Rust TRACKS listing sends `SUBSCRIBE_TRACKS` upstream, only on demand.** Listing a broadcast from a moq-transport peer on draft 18 and later sends `SUBSCRIBE_TRACKS` with `FORWARD=0` while someone lists it, never on connect, announce, or subscribe, so a peer that refuses, resets, or ignores it affects only that listing. Before draft 18 the listing is refused as unsupported. Each listed track's paused `PUBLISH` is its upstream copy, which a reader resumes rather than subscribing again. Only a `PUBLISH_DONE` with `TRACK_ENDED` removes a listed track, and it lists as finished; any other end, including the error our own answerer sends for a track that left unfinished, fails the listing, and readers list again. - **History is one group.** In Rust, Object ID gaps must be marked by Prior Object ID Gap properties for a group to be fetchable. A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On drafts 14 through 19, `FETCH_OK` End Object 0 covers the entire End Group, matching the request; Rust also accepts the legacy answer naming the next group with Object 0. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed, and so does a reader whose publisher refuses `TRACK_STATUS` as `NOT_SUPPORTED`. diff --git a/js/net/bench/forward.ts b/js/net/bench/forward.ts index f174a35548a..270205ccc1d 100644 --- a/js/net/bench/forward.ts +++ b/js/net/bench/forward.ts @@ -36,6 +36,7 @@ class FakeSession { registerWire(this, { consume: () => new BroadcastProducer().consume(), goaway: new Once(), + exclusive: false, auth: new AuthSession({ peerGrant: { publish: new Path.Patterns(), subscribe: new Path.Patterns() } }), }); this.closed = new Promise((resolve) => { diff --git a/js/net/src/connection/forward.test.ts b/js/net/src/connection/forward.test.ts index 9a737265a5e..9fd8a540f2a 100644 --- a/js/net/src/connection/forward.test.ts +++ b/js/net/src/connection/forward.test.ts @@ -49,10 +49,12 @@ class FakeSession { closed: Promise; #die!: (reason: Error | null) => void; - constructor() { + /** `exclusive` stands in for a peer that refuses overlapping requests. */ + constructor({ exclusive = false } = {}) { registerWire(this, { consume: (path) => this.consume(path), goaway: new Once(), + exclusive, auth: new AuthSession({ peerGrant: { publish: new Path.Patterns(), subscribe: new Path.Patterns() } }), }); this.closed = new Promise((resolve) => { @@ -341,6 +343,48 @@ test("a narrower request stays open until the broader one covering it goes", asy origin.close(); }); +test("an exclusive session closes every request beneath a broader one before it opens", async () => { + const origin = new OriginProducer(); + const session = new FakeSession({ exclusive: true }); + forwardAnnounced(session.session, origin); + + const narrow = origin.announced(Path.Pattern.subtree(Path.from("a/b"))); + const lingering = origin.announced(Path.Pattern.subtree(Path.from("a/c"))); + await settle(); + start(session.stream("a/b/**"), "a/b/x"); + lingering.close(); + await settle(); + expect(session.open()).toEqual(["a/b/**", "a/c/**"]); + + // Watched or lingering, each narrower request closes for the broader one, on purpose: the + // session stays discovering, and the route returns once the broader request reports it. + const broad = origin.announced(Path.Pattern.subtree(Path.from("a"))); + const events = record(broad); + await settle(); + expect(session.open()).toEqual(["a/**"]); + expect(wireOf(origin).routes(Path.from("a/b/x"))).toBe(false); + start(session.stream("a/**"), "a/b/x"); + await settle(); + expect(events).toEqual(["start a/b/x", "end a/b/x", "start a/b/x"]); + expect(origin.discovery.peek()).toBe(true); + + // The narrower watch rides the broader request from here on. + broad.close(); + await settle(); + jest.advanceTimersByTime(LINGER); + await settle(); + expect(session.open()).toEqual(["a/**"]); + narrow.close(); + await settle(); + jest.advanceTimersByTime(LINGER); + await settle(); + expect(session.open()).toEqual([]); + expect(session.requests.map(({ scope }) => scope)).toEqual(["a/b/**", "a/c/**", "a/**"]); + + session.die(); + origin.close(); +}); + const first = Epoch.mint(); for (const [name, epochs] of [ ["no epoch", [undefined, undefined]], diff --git a/js/net/src/connection/forward.ts b/js/net/src/connection/forward.ts index 684482c7977..641ca41c536 100644 --- a/js/net/src/connection/forward.ts +++ b/js/net/src/connection/forward.ts @@ -58,6 +58,11 @@ interface Reported { * driven by its first reporter; if it changes hands anyway (a request failing), it restarts * unless one epoch proves both requests saw one instance. * + * A peer that refuses overlapping requests gets exclusive discovery instead: a broader request + * closes the narrower ones beneath it, and the session keeps it off the wire until they have + * left. The routes a closed request carried go with it, and come back once the broader request + * reports them. + * * Each active announcement inserts a route covering the announced prefix, served lazily * through this connection, so nothing touches the wire until somebody requests a path * under it. A retraction removes the entry, and so does the session dying, which is what @@ -72,6 +77,7 @@ interface Reported { */ export function forwardAnnounced(conn: Established, origin: OriginProducer): void { const originWire = wireOf(origin); + const exclusive = wireOf(conn).exclusive; // Reassigned on a downgrade, so the origin stops counting this session as discovering. // Called through a closure so the session's death always detaches the current one. let detach = originWire.attach(true); @@ -155,6 +161,17 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi }; const open = (prefix: Path.Valid) => { + if (exclusive) { + for (const [other, narrower] of interests) { + if (other === prefix || !Path.hasPrefix(prefix, other)) continue; + interests.delete(other); + narrower.closing = true; + narrower.linger?.(); + narrower.linger = undefined; + narrower.announced.close(); + } + } + // Hidden routes are mirrored too; each local reader opts in on its own. const interest: Interest = { announced: conn.announced(Path.Pattern.subtree(prefix), { hidden: true }), diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index ca6a5b93457..5a2479f9d24 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -276,6 +276,9 @@ export class Connection implements Established { registerWire(this, { consume: (path) => this.#subscriber.consume(path), goaway: this.#goaway, + // The drafts refuse overlapping SUBSCRIBE_NAMESPACEs (PREFIX_OVERLAP). A peer that + // declared MoQ Solicit is ours and holds both; any other may enforce it. + exclusive: solicit === undefined, auth: this.#auth, }); diff --git a/js/net/src/ietf/request_stream.ts b/js/net/src/ietf/request_stream.ts index 7fe11930944..7fcbc95b4ab 100644 --- a/js/net/src/ietf/request_stream.ts +++ b/js/net/src/ietf/request_stream.ts @@ -1,4 +1,4 @@ -import { error, ProtocolViolation, StreamError } from "../error.ts"; +import { error, ProtocolViolation, StreamCode, StreamError } from "../error.ts"; import type * as Path from "../path.ts"; import { Reader, type Stream, type Writer } from "../stream.ts"; import type { Session } from "./adapter.ts"; @@ -230,6 +230,16 @@ export class RequestStream { this.close(); } + /** + * Cancel the request but keep reading, so {@link next} reports when the peer ends its + * half: FIN where that cancels, a CANCELLED reset otherwise. + */ + cancel(): void { + if (this.#closed) return; + if (finCancels(this.#session.version)) this.writer.close(); + else this.writer.reset(new StreamError(StreamCode.Cancel, { message: "cancel" })); + } + /** End the request cleanly: FIN, and stop reading. Only the first close counts. */ close(): void { if (this.#closed) return; diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index df11eabba47..854cacfe27f 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -28,7 +28,13 @@ import { } from "./publish_namespace.ts"; import { MaxRequestId, RequestError, RequestOk } from "./request.ts"; import { Subscribe, SubscribeOk, SubscribeUpdate, Unsubscribe } from "./subscribe.ts"; -import { SubscribeNamespace, SubscribeNamespaceEntry, SubscribeNamespaceEntryDone } from "./subscribe_namespace.ts"; +import { + SubscribeNamespace, + SubscribeNamespaceEntry, + SubscribeNamespaceEntryDone, + SubscribeNamespaceLegacy, + UnsubscribeNamespace, +} from "./subscribe_namespace.ts"; import { Subscriber } from "./subscriber.ts"; import { ALPN, type IetfVersion, Version } from "./version.ts"; @@ -418,11 +424,14 @@ test("an entry buffered past a consumer close does not pin the path", async () = await inlineNamespace(subscription, Path.from("buffered")); announced.close(); + // The peer ends its half on the cancel, so the request leaves the wire. + subscription.writer.close(); // Whatever the loop managed to attach, the stream gave back on its way out. await new Promise((resolve) => setTimeout(resolve, 50)); const fresh = subscriber.announced(); + await acceptSubscribeNamespace(pair.client); const seeded = await Promise.race([ fresh.next(), new Promise<"nothing">((resolve) => setTimeout(() => resolve("nothing"), 250)), @@ -2223,6 +2232,75 @@ test("an announce consumer closed while waiting on request-ID credit takes no ID void running; }); +/** + * A request on another stream can overtake a cancel, so a peer that refuses overlapping + * requests (PREFIX_OVERLAP) would see the broader one beside the narrower one it replaces. + * It goes out only once the peer has ended the narrower one's stream. + */ +test("a request overlapping one on its way out waits for the peer to end it", async () => { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const session = new NativeSession(pair.server, VERSION, true); + const subscriber = new Subscriber({ session }); + + const narrow = subscriber.announced(Path.Pattern.subtree(Path.from("a/b"))); + const stream = await acceptSubscribeNamespace(pair.client); + await inlineNamespace(stream, Path.from("x")); + expect(await narrow.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "start" }); + + // A request still open is no blocker, and hears the narrower one's route go at once. + const watcher = subscriber.announced(Path.Pattern.subtree(Path.from("a/b"))); + const watching = await acceptSubscribeNamespace(pair.client); + expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "start" }); + narrow.close(); + const broad = subscriber.announced(Path.Pattern.subtree(Path.from("a"))); + expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "end" }); + + // Draft 19 cancels with a reset, and the broader request waits on the peer's half. + expect(stream.reader.done()).rejects.toMatchObject({ code: StreamCode.Cancel }); + expect(await nextStream(pair.client)).toBeUndefined(); + + // The closed reader is still registered meanwhile, and hears nothing more. + await inlineNamespace(watching, Path.from("y")); + expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/y"), kind: "start" }); + + stream.writer.close(); + const asked = await nextStream(pair.client); + if (!asked) throw new Error("the broader SUBSCRIBE_NAMESPACE never went out"); + await asked.reader.u53(); + expect((await SubscribeNamespace.decode(asked.reader, VERSION)).namespace).toBe(Path.from("a")); + + watcher.close(); + broad.close(); + session.close(); +}); + +/** Draft 14/15 requests share the control stream, so the next one waits only for the UNSUBSCRIBE_NAMESPACE. */ +test("a draft-15 request overlapping one on its way out follows its UNSUBSCRIBE_NAMESPACE", async () => { + const version = Version.DRAFT_15; + const pair = createMockTransportPair(ALPN.DRAFT_15); + const control = await Stream.open(pair.server, { version }); + const session = new ControlStreamAdapter(pair.server, control, version, 100n, true); + void session.run().catch(() => undefined); + const peer = await Stream.accept(pair.client, version); + if (!peer) throw new Error("no control stream"); + const subscriber = new Subscriber({ session }); + + const narrow = subscriber.announced(Path.Pattern.subtree(Path.from("a/b"))); + expect(await peer.reader.u53()).toBe(SubscribeNamespaceLegacy.id); + expect((await SubscribeNamespaceLegacy.decode(peer.reader, version)).namespace).toBe(Path.from("a/b")); + + narrow.close(); + const broad = subscriber.announced(Path.Pattern.subtree(Path.from("a"))); + + expect(await peer.reader.u53()).toBe(UnsubscribeNamespace.id); + await UnsubscribeNamespace.decode(peer.reader, version); + expect(await peer.reader.u53()).toBe(SubscribeNamespaceLegacy.id); + expect((await SubscribeNamespaceLegacy.decode(peer.reader, version)).namespace).toBe(Path.from("a")); + + broad.close(); + session.close(); +}); + /** Answer the next SUBSCRIBE_NAMESPACE with REQUEST_OK, carrying `activeCount` when given. */ async function answerCounted(transport: WebTransport, activeCount: number | undefined): Promise { const stream = await nextStream(transport); diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 62147a9ccbd..0c20facf4f2 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -132,6 +132,9 @@ type Announced = { adverts: { id: number; route: Route }[]; route: Route }; /** A local announce reader's filter: its scope, the prefix it asked for, and its hidden opt-in. */ type Filter = { scope: Path.Pattern; prefix: Path.Valid; hidden: boolean }; +/** A local announce reader: its filter, and what settles once its request is off the wire. */ +type Asking = Filter & { gone?: Promise }; + /** Whether a reader with `filter` sees an announcement at `path`. */ function sees(filter: Filter, path: Path.Valid): boolean { return scopeOverlaps(filter.scope, path) && (filter.hidden || !hiddenBelow(filter.prefix, path)); @@ -191,8 +194,9 @@ export class Subscriber { // Names each advertisement the peer makes, so a later update or withdrawal finds its own. #nextAdvert = 0; - // Any consumers that want each new announcement, keyed by their local filter. - #announcedConsumers = new Map(); + // Any consumers that want each new announcement, keyed by their local filter, until + // their request is off the wire. + #announcedConsumers = new Map(); // Whether the peer understands the HIDDEN parameter (MoQ Hidden). #hidden: boolean; @@ -339,11 +343,24 @@ export class Subscriber { * `options.hidden` opts in. The opt-in rides the SUBSCRIBE_NAMESPACE when the peer * understands it (MoQ Hidden); the rule is also applied here, since an unsolicited * PUBLISH_NAMESPACE or a peer that never heard of it hides nothing. + * + * The request goes out once every overlapping request already closed here is off the + * wire, so a peer that refuses overlapping requests never sees this one beside them. */ announced(scope: Path.Pattern = Path.Pattern.all(), options?: announce.Options): announce.Consumer { // The wire speaks announce interest by prefix. const prefix = scopeHead(scope); - const filter = { scope, prefix, hidden: options?.hidden ?? false }; + const filter: Asking = { scope, prefix, hidden: options?.hidden ?? false }; + + // The overlapping requests on their way out, such as the narrower ones an exclusive + // discovery just closed for this one. Those still open overlap it only where the + // caller holds both on purpose. + const blockers: Promise[] = []; + for (const [other, { prefix: asked, gone }] of this.#announcedConsumers) { + if (gone === undefined || other.closed.peek() === undefined) continue; + if (Path.hasPrefix(asked, prefix) || Path.hasPrefix(prefix, asked)) blockers.push(gone); + } + const announced = new announce.Producer(); for (const [active, info] of this.#announced) { if (!sees(filter, active)) continue; @@ -356,7 +373,7 @@ export class Subscriber { } this.#announcedConsumers.set(announced, filter); - void this.#runAnnounced(announced, prefix, filter.hidden && this.#hidden).finally(() => { + filter.gone = this.#runAnnounced(announced, prefix, filter.hidden && this.#hidden, blockers).finally(() => { this.#announcedConsumers.delete(announced); announced.close(); }); @@ -383,7 +400,8 @@ export class Subscriber { console.debug(`announced: broadcast=${path} active=true epoch=${route.epoch}`); for (const [consumer, filter] of this.#announcedConsumers) { - if (!sees(filter, path)) continue; + // A consumer that closed stays registered until its request is off the wire. + if (!sees(filter, path) || consumer.closed.peek() !== undefined) continue; const scope = filter.scope; consumer.append({ prefix: path, captures: scopeCaptures(scope, path), kind: "start", route }); } @@ -422,9 +440,8 @@ export class Subscriber { console.debug(`announced: broadcast=${path} ${restart ? `restart=true epoch=${route.epoch}` : "rerouted"}`); const kind = restart ? "restart" : "update"; for (const [consumer, filter] of this.#announcedConsumers) { - if (!sees(filter, path)) continue; - // A consumer that closed stays registered until its reader loop notices. - if (consumer.closed.peek() !== undefined) continue; + // A consumer that closed stays registered until its request is off the wire. + if (!sees(filter, path) || consumer.closed.peek() !== undefined) continue; const scope = filter.scope; consumer.append({ prefix: path, captures: scopeCaptures(scope, path), kind, route }); } @@ -452,18 +469,10 @@ export class Subscriber { console.debug(`announced: broadcast=${path} active=false`); for (const [consumer, filter] of this.#announcedConsumers) { - if (!sees(filter, path)) continue; + // A consumer that closed stays registered until its request is off the wire. + if (!sees(filter, path) || consumer.closed.peek() !== undefined) continue; const scope = filter.scope; - try { - consumer.append({ - prefix: path, - captures: scopeCaptures(scope, path), - kind: "end", - route: existing.route, - }); - } catch { - // Consumer already closed, will be cleaned up - } + consumer.append({ prefix: path, captures: scopeCaptures(scope, path), kind: "end", route: existing.route }); } } @@ -478,7 +487,12 @@ export class Subscriber { this.#session.close(); } - async #runAnnounced(announced: announce.Producer, prefix: Path.Valid, hidden: boolean) { + async #runAnnounced( + announced: announce.Producer, + prefix: Path.Valid, + hidden: boolean, + blockers: Promise[], + ): Promise { const version = this.#session.version; // A zero-field track namespace was a protocol violation until draft-16 allowed @@ -504,13 +518,33 @@ export class Subscriber { // route is detached (`id` unset) for looping back through us. const live = new Map(); - // Set once the teardown below has given back everything this stream held. The read - // loop is not awaited when the local consumer closes first, and closing the stream - // cancels the transport without discarding what the reader already buffered, so a - // fully buffered entry can still decode afterwards. Attaching one then would take a - // reference nobody is left to release, pinning the path for the session. + // Set once `release` has given back everything this stream held. The read loop is not + // awaited when the local consumer closes first, and the stream keeps delivering what + // the peer sends until it ends its half, so an entry can still decode afterwards. + // Attaching one then would take a reference nobody is left to release, pinning the + // path for the session. let released = false; + // The stream owns every advertisement it carried, so give them back however it ends: + // as the request starts leaving, a decode error, or the peer resetting it. Without + // this each namespace keeps its count and the source never detaches, which would pin + // the path for the session even after the other source withdrew. + const release = () => { + released = true; + for (const [path, held] of live) { + if (held.id !== undefined) this.#detachAnnounce(path, held.id); + } + live.clear(); + }; + + // A peer that refuses overlapping requests would refuse this one while it still holds + // a request this overlaps, so wait for those to leave the wire. A consumer that leaves + // meanwhile sends nothing. + if (blockers.length > 0) { + await race([Promise.allSettled(blockers), announced.closed]); + if (announced.closed.peek() !== undefined) return; + } + // v14/v15: SubscribeNamespace on control stream (via adapter virtual stream) // v16+: SubscribeNamespace on its own real bidi stream @@ -559,15 +593,30 @@ export class Subscriber { } console.debug(`subscribe_namespace written: requestId=${requestId}`); - const end = async () => { - // For v14/v15: send UnsubscribeNamespace before closing - if (version === Version.DRAFT_14 || version === Version.DRAFT_15) { + // End the request, settling once it is off the wire. `reading` is whatever reads + // the stream now, which has to finish before anything else may. + const end = async (reading: Promise) => { + release(); + // Draft-14/15 adapt the request onto the control stream, so whatever we send + // after the UNSUBSCRIBE_NAMESPACE reaches the peer after it. + if (legacy) { try { const unsub = new UnsubscribeNamespace({ prefix, requestId }); await Message.write(request.writer, UnsubscribeNamespace.id, unsub, version); } catch { // Stream might already be closed } + request.close(); + return; + } + // A request on another stream can overtake the cancel, so this one is off the + // wire only once the peer ends its half. What it sends meanwhile is dropped. + request.cancel(); + await reading.catch(() => undefined); + try { + while ((await request.next()) !== undefined) {} + } catch { + // Reset, or the session ended. } request.close(); }; @@ -594,7 +643,7 @@ export class Subscriber { // A consumer that leaves before the peer answers ends the request now, not on the answer. const answered = await race([response, announced.closed.then(() => undefined)]); if (answered === undefined) { - await end(); + await end(response); return; } let initial = answered; @@ -605,6 +654,8 @@ export class Subscriber { if (initial === 0) announced.synced(); const framed = await request.next(); + // Ending: `end` drains the rest. + if (released) break; if (framed === undefined) { if (initial > 0) throw new ProtocolViolation("namespace stream ended inside the initial set"); @@ -685,7 +736,7 @@ export class Subscriber { // Wait for either the read loop or the announced to close await race([readLoop, announced.closed]); - await end(); + await end(readLoop); }; await request.serve(run()); } catch (err: unknown) { @@ -698,15 +749,7 @@ export class Subscriber { // that will never be announced. Matches the lite subscriber. announced.close(e); } finally { - // The stream owns every advertisement it carried, so release them however it - // ends: a clean close, a decode error, or the peer resetting it. Without this - // each namespace keeps its count and the source never detaches, which would - // pin the path for the session even after the other source withdrew. - released = true; - for (const [path, held] of live) { - if (held.id !== undefined) this.#detachAnnounce(path, held.id); - } - live.clear(); + release(); } } diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index adc771443e6..45dd455680d 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -223,6 +223,7 @@ export class Connection implements Established { registerWire(this, { consume: (path) => this.#subscriber.consume(path), goaway: this.#goaway, + exclusive: false, auth: this.#auth, }); diff --git a/js/net/src/wire.ts b/js/net/src/wire.ts index 4e668fe37e8..6affdc47055 100644 --- a/js/net/src/wire.ts +++ b/js/net/src/wire.ts @@ -118,6 +118,11 @@ export interface Established { consume(path: Path.Valid): broadcast.Consumer; /** Settles with the peer's GOAWAY; the session keeps serving until it closes. */ readonly goaway: GetPromise; + /** + * Whether the peer refuses an announce request overlapping one it still holds, so + * discovery never holds two where one covers the other. + */ + readonly exclusive: boolean; // private: plumbing /** * This session's side of its auth handle. diff --git a/quest/m1/README.md b/quest/m1/README.md index 68a02e52c56..0908bf773e2 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -155,7 +155,6 @@ Catalog HLS field. - [API-diff check](/quest/m1/api-diff.md) - a recipe diffs every published package's API against its last release, run on PRs into release and against main before the cut - [API Extractor checks](/quest/m1/api-extractor-checks.md) - API Extractor fails a PR on a forgotten export, an unresolved declaration, or an undocumented export, replacing the hand-rolled `.d.ts` checkers - [Generated JS API reference](/quest/m1/js-api-reference.md) - the doc site renders an API reference for every `@moq/*` package from API Extractor's doc model -- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - JS moq-transport sessions never hold overlapping SUBSCRIBE_NAMESPACE requests at a peer that refuses them - [Interop coverage](/quest/m1/interop-coverage/README.md) - the riskiest moq-transport features moq-net exposes get cross-implementation tests over drafts 18 and 22 - [AMF0 exact reads](/quest/m1/rtmp-amf0-exact-reads.md) - a truncated RTMP string fails with `UnexpectedEof` instead of parsing NUL padding - [HLS import arithmetic](/quest/m1/hls-variant-area.md) - a hostile `RESOLUTION` or media sequence cannot panic, wrap, or misorder the import diff --git a/quest/m1/ietf-lazy-discovery.md b/quest/m1/ietf-lazy-discovery.md deleted file mode 100644 index aa86be17d8e..00000000000 --- a/quest/m1/ietf-lazy-discovery.md +++ /dev/null @@ -1,41 +0,0 @@ -# [M] IETF lazy discovery in JS - -## Goal - -A JS moq-transport session never holds two overlapping SUBSCRIBE_NAMESPACE -requests at a peer that did not declare MoQ Solicit, so a peer that enforces -the drafts' PREFIX_OVERLAP rule, such as moxygen, never refuses a broader -watch and turns the session blind. - -## Plan - -Rust landed in [moq-dev/moq#6530](https://github.com/moq-dev/moq/pull/6530): -lazy SUBSCRIBE_NAMESPACE on every draft, UNSUBSCRIBE_NAMESPACE on drafts -14/15 (named by prefix on 14, fixed in both languages), and exclusive -discovery at a peer that declared no Solicit (`Discovery::exclusive`, -`rs/moq-net/src/ietf/subscriber/namespaces.rs`). JS -`connection/forward.ts` already asks only about watched prefixes, but holds a -narrower request open under a broader one on every version. - -- At a peer that declared no Solicit, a broader watch closes the narrower - requests beneath it, and its SUBSCRIBE_NAMESPACE goes out only once each - has left the wire: UNSUBSCRIBE_NAMESPACE written on drafts 14/15, the peer's - half closed from draft 16. -- `forward.ts` needs the session to say when a request is off the wire; the - announced consumer reports only its local close today. - -Settled in [moq-dev/moq#6530](https://github.com/moq-dev/moq/pull/6530): - -- JS non-overlap is this quest, not that PR: it needs the session signal - above, and a refused JS watch only downgrades the session to blind answers. -- Overlap policy is keyed on MoQ Solicit. Requests are exclusive only at a - peer that declared no Solicit, since only foreign peers enforce - PREFIX_OVERLAP; our own peers keep overlapping hold semantics, because on - drafts 14/15 our publisher withdraws what an unsubscribed request - advertised, so exclusive-everywhere would restart routes mid-handover. -- A PUBLISH_NAMESPACE that outlives an UNSUBSCRIBE_NAMESPACE (drafts 14/15), - or arrives unsolicited, is kept until the peer withdraws it. It is bounded - by request credit or streams and the `announces` cap, and cancelling it - would lose it for good, since the peer does not repeat it. - -Public API: none expected. Wire: none. diff --git a/quest/m1/ietf-request-cancel.md b/quest/m1/ietf-request-cancel.md index 37cab3dd181..41839501095 100644 --- a/quest/m1/ietf-request-cancel.md +++ b/quest/m1/ietf-request-cancel.md @@ -11,8 +11,8 @@ credit. Every request write in `rs/moq-net/src/ietf/subscriber.rs` awaits its first message without watching for cancellation, so a peer that withholds credit pins a cancelled request. -[IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) fixes only the -stream-open race. On drafts 16 and up, reset a partially written request +Lazy discovery ([moq-dev/moq#6530](https://github.com/moq-dev/moq/pull/6530)) +fixed only the stream-open race. On drafts 16 and up, reset a partially written request stream. Drafts 14 and 15 share the ordered control stream, so a partial message cannot be abandoned; decide there whether to finish the message and cancel right after, and record the choice here. @@ -20,7 +20,3 @@ cancel right after, and record the choice here. Test with a peer that never grants credit, on each draft family. Public API: none. Wire: none. - -## Required - -- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - changes the same request writes From 7888d7b175b83305a638e6d7943b3c26755e6e78 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 10:09:39 -0700 Subject: [PATCH 3/4] test(js): check the held SUBSCRIBE_NAMESPACE without a wall-clock wait Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/subscriber.test.ts | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index 854cacfe27f..7f589501807 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -2257,15 +2257,23 @@ test("a request overlapping one on its way out waits for the peer to end it", as // Draft 19 cancels with a reset, and the broader request waits on the peer's half. expect(stream.reader.done()).rejects.toMatchObject({ code: StreamCode.Cancel }); - expect(await nextStream(pair.client)).toBeUndefined(); + const incoming = pair.client.incomingBidirectionalStreams.getReader(); + let sent = false; + const opened = incoming.read().then((next) => { + sent = true; + return next; + }); // The closed reader is still registered meanwhile, and hears nothing more. await inlineNamespace(watching, Path.from("y")); expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/y"), kind: "start" }); + expect(sent).toBe(false); stream.writer.close(); - const asked = await nextStream(pair.client); - if (!asked) throw new Error("the broader SUBSCRIBE_NAMESPACE never went out"); + const next = await opened; + incoming.releaseLock(); + if (next.done) throw new Error("the broader SUBSCRIBE_NAMESPACE never went out"); + const asked = new Stream({ readable: next.value.readable, writable: next.value.writable, version: VERSION }); await asked.reader.u53(); expect((await SubscribeNamespace.decode(asked.reader, VERSION)).namespace).toBe(Path.from("a")); From 96d9a587511266832bc1cf92bd57acb02a974641 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 10:11:22 -0700 Subject: [PATCH 4/4] test(js): run the reset-cancel overlap test on draft 22 Drafts 19-21 were dropped on main, and VERSION is now draft 18, whose FIN cancels, so the test pins draft 22 and the helpers take its version. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/subscriber.test.ts | 40 +++++++++++++++++------------- 1 file changed, 23 insertions(+), 17 deletions(-) diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index c3fceeb6bb8..b93a1981800 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -49,7 +49,7 @@ const STREAM_WAIT = 500; * Reads the queue directly instead of racing {@link Stream.accept}, whose pending read * would keep the reader locked after the race resolves and could swallow a later stream. */ -async function nextStream(transport: WebTransport): Promise { +async function nextStream(transport: WebTransport, version: IetfVersion = VERSION): Promise { const reader = transport.incomingBidirectionalStreams.getReader() as ReadableStreamDefaultReader; @@ -63,7 +63,7 @@ async function nextStream(transport: WebTransport): Promise ]); if (!next || next.done) return undefined; - return new Stream({ readable: next.value.readable, writable: next.value.writable, version: VERSION }); + return new Stream({ readable: next.value.readable, writable: next.value.writable, version }); } finally { clearTimeout(timer); reader.releaseLock(); @@ -234,23 +234,28 @@ test("an advertisement with no path is anonymous", async () => { * Answer the SUBSCRIBE_NAMESPACE the subscriber just opened, then hand back its stream so * the test can feed inline NAMESPACE entries down it. */ -async function acceptSubscribeNamespace(transport: WebTransport): Promise { - const stream = await nextStream(transport); +async function acceptSubscribeNamespace(transport: WebTransport, version: IetfVersion = VERSION): Promise { + const stream = await nextStream(transport, version); if (!stream) throw new Error("no SUBSCRIBE_NAMESPACE was sent"); // Drain the request, then answer it. await stream.reader.u53(); - await SubscribeNamespace.decode(stream.reader, VERSION); + await SubscribeNamespace.decode(stream.reader, version); await stream.writer.u53(RequestOk.id); - await new RequestOk({ requestId: undefined }).encode(stream.writer, VERSION); + await new RequestOk({ requestId: undefined }).encode(stream.writer, version); return stream; } /** Advertise `path` inline on a SUBSCRIBE_NAMESPACE stream. */ -async function inlineNamespace(stream: Stream, path: Path.Valid, cluster?: Cluster.Advert): Promise { +async function inlineNamespace( + stream: Stream, + path: Path.Valid, + cluster?: Cluster.Advert, + version: IetfVersion = VERSION, +): Promise { await stream.writer.u53(SubscribeNamespaceEntry.id); - await new SubscribeNamespaceEntry({ suffix: path, params: cluster && { cluster } }).encode(stream.writer, VERSION); + await new SubscribeNamespaceEntry({ suffix: path, params: cluster && { cluster } }).encode(stream.writer, version); } /** @@ -2236,24 +2241,25 @@ test("an announce consumer closed while waiting on request-ID credit takes no ID * It goes out only once the peer has ended the narrower one's stream. */ test("a request overlapping one on its way out waits for the peer to end it", async () => { - const pair = createMockTransportPair(ALPN.DRAFT_19); - const session = new NativeSession(pair.server, VERSION, true); + const version = Version.DRAFT_22; + const pair = createMockTransportPair(ALPN.DRAFT_22); + const session = new NativeSession(pair.server, version, true); const subscriber = new Subscriber({ session }); const narrow = subscriber.announced(Path.Pattern.subtree(Path.from("a/b"))); - const stream = await acceptSubscribeNamespace(pair.client); - await inlineNamespace(stream, Path.from("x")); + const stream = await acceptSubscribeNamespace(pair.client, version); + await inlineNamespace(stream, Path.from("x"), undefined, version); expect(await narrow.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "start" }); // A request still open is no blocker, and hears the narrower one's route go at once. const watcher = subscriber.announced(Path.Pattern.subtree(Path.from("a/b"))); - const watching = await acceptSubscribeNamespace(pair.client); + const watching = await acceptSubscribeNamespace(pair.client, version); expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "start" }); narrow.close(); const broad = subscriber.announced(Path.Pattern.subtree(Path.from("a"))); expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "end" }); - // Draft 19 cancels with a reset, and the broader request waits on the peer's half. + // Draft 22 cancels with a reset, and the broader request waits on the peer's half. expect(stream.reader.done()).rejects.toMatchObject({ code: StreamCode.Cancel }); const incoming = pair.client.incomingBidirectionalStreams.getReader(); let sent = false; @@ -2263,7 +2269,7 @@ test("a request overlapping one on its way out waits for the peer to end it", as }); // The closed reader is still registered meanwhile, and hears nothing more. - await inlineNamespace(watching, Path.from("y")); + await inlineNamespace(watching, Path.from("y"), undefined, version); expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/y"), kind: "start" }); expect(sent).toBe(false); @@ -2271,9 +2277,9 @@ test("a request overlapping one on its way out waits for the peer to end it", as const next = await opened; incoming.releaseLock(); if (next.done) throw new Error("the broader SUBSCRIBE_NAMESPACE never went out"); - const asked = new Stream({ readable: next.value.readable, writable: next.value.writable, version: VERSION }); + const asked = new Stream({ readable: next.value.readable, writable: next.value.writable, version }); await asked.reader.u53(); - expect((await SubscribeNamespace.decode(asked.reader, VERSION)).namespace).toBe(Path.from("a")); + expect((await SubscribeNamespace.decode(asked.reader, version)).namespace).toBe(Path.from("a")); watcher.close(); broad.close();