From a8e132afdbfa2d905580ba12d24db5010a5b3ec6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 10:39:42 -0700 Subject: [PATCH 1/3] fix(net): refuse new requests from retired sources Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 2 + js/net/bench/broadcasts.ts | 38 +++ js/net/src/consume.test.ts | 72 +++++ js/net/src/consume.ts | 84 +++++- js/net/src/ietf/adapter.test.ts | 27 ++ js/net/src/ietf/adapter.ts | 28 +- js/net/src/ietf/subscriber.test.ts | 93 ++++++- js/net/src/ietf/subscriber.ts | 35 ++- js/net/src/lite/subscriber.test.ts | 161 +++++++++++- js/net/src/lite/subscriber.ts | 88 +++++-- quest/m0/broadcast-epoch/README.md | 1 - .../m0/broadcast-epoch/catalog-references.md | 2 +- quest/m0/broadcast-epoch/retired-requests.md | 65 ----- quest/m1/announce-takeover.md | 2 +- quest/m1/fetch-max-delay.md | 2 +- quest/m1/js-group-handover.md | 2 +- quest/m1/pipeline-requests/fetch.md | 2 +- rs/moq-net/Cargo.toml | 6 + rs/moq-net/benches/retirement.rs | 70 +++++ rs/moq-net/src/fuzz.rs | 18 ++ rs/moq-net/src/ietf/subscriber.rs | 245 +++++++++++------- rs/moq-net/src/lite/subscriber.rs | 45 +++- rs/moq-net/src/model/broadcast.rs | 51 ++++ rs/moq-net/src/model/origin.rs | 245 +++++++++++++++--- rs/moq-net/tests/restart.rs | 159 ++++++++++++ 25 files changed, 1279 insertions(+), 264 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/retired-requests.md create mode 100644 rs/moq-net/benches/retirement.rs diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 743272c3b6..861b1dead5 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -87,6 +87,8 @@ authorization and hidden names work exactly as without one. winning the path, is announced as a restart. Subscriptions already open stay on the old instance until dropped or its route goes, while new requests get the new one, so a player follows the restart by requesting the path again. + Held handles can still read their old cached content, but cannot open new + epochless requests after retirement. Request the path again for new content. That is how a restarted encoder, whose group numbers start over, takes the name instead of stalling viewers on the old sequence. A relay drops its copy of the old instance, so a new subscriber never sees its cached groups. diff --git a/js/net/bench/broadcasts.ts b/js/net/bench/broadcasts.ts index 11c0761f70..4ed0a19e14 100644 --- a/js/net/bench/broadcasts.ts +++ b/js/net/bench/broadcasts.ts @@ -327,3 +327,41 @@ for (const routeCount of routeCounts) { source.close(); } } + +// Retiring a covering route cancels its pending descendants, not unrelated consumers. +console.log("retirement,routes,consumers,retire_us,refused"); +for (const routeCount of routeCounts) { + for (const consumerCount of observerCounts) { + let elapsed = 0; + let refused = 0; + for (let round = 0; round < 256; round++) { + const cache = new BroadcastCache(); + const source = new BroadcastProducer(); + const owners = Array.from({ length: routeCount }, () => ({})); + const held = []; + for (let route = 0; route < routeCount; route++) { + const prefix = Path.from(`route/${route}`); + cache.announce(prefix, owners[route], Route.default); + for (let consumer = 0; consumer < (route === 0 ? consumerCount : 1); consumer++) { + const path = Path.from(`${prefix}/job/${consumer}`); + const instance = cache.instance(path); + if (route === 0) + instance.validity?.signal.addEventListener( + "abort", + () => { + refused++; + }, + { once: true }, + ); + held.push(cache.insert(path, instance, source.consume())); + } + } + const start = performance.now(); + cache.withdraw(Path.from("route/0"), owners[0]); + elapsed += performance.now() - start; + for (const consumer of held) consumer.close(); + source.close(); + } + console.log(`retirement,${routeCount},${consumerCount},${((elapsed * 1000) / 256).toFixed(2)},${refused}`); + } +} diff --git a/js/net/src/consume.test.ts b/js/net/src/consume.test.ts index 8d84a1ca54..9637b61a86 100644 --- a/js/net/src/consume.test.ts +++ b/js/net/src/consume.test.ts @@ -99,3 +99,75 @@ test.each([false, true])("route alternation reuses every held instance (epoch: % a.close(); b.close(); }); + +test("request validity cancels one selection period and permits the held fallback again", () => { + const cache = new BroadcastCache(); + const path = Path.from("pool/job"); + const broad = Path.from("pool"); + const a = {}; + const b = {}; + cache.announce(broad, a, Route.default); + const sourceA = new Producer(); + const instanceA = cache.instance(path); + const heldA = cache.insert(path, instanceA, sourceA.consume()); + const first = instanceA.validity?.signal; + expect(first?.aborted).toBe(false); + cache.announce(path, b, Route.default); + expect(first?.aborted).toBe(true); + const sourceB = new Producer(); + const instanceB = cache.instance(path); + const heldB = cache.insert(path, instanceB, sourceB.consume()); + expect(instanceB.validity?.signal.aborted).toBe(false); + cache.withdraw(path, b); + expect(instanceB.validity?.signal.aborted).toBe(true); + expect(instanceA.validity?.signal).not.toBe(first); + expect(instanceA.validity?.signal.aborted).toBe(false); + expect(first?.aborted).toBe(true); + expect(heldA.closed.peek()).toBeUndefined(); + expect(heldB.closed.peek()).toBeUndefined(); + heldA.close(); + heldB.close(); + sourceA.close(); + sourceB.close(); +}); + +test("request validity follows the selected interest and ignores unrelated withdrawals", () => { + const cache = new BroadcastCache(); + const path = Path.from("pool/job"); + const broad = {}; + const first = {}; + const second = {}; + cache.announce(Path.from("pool"), broad, Route.default); + cache.announce(path, first, Route.default); + cache.announce(path, second, Route.default); + const source = new Producer(); + const instance = cache.instance(path); + const held = cache.insert(path, instance, source.consume()); + const signal = instance.validity?.signal; + cache.withdraw(Path.from("pool"), broad); + cache.withdraw(path, second); + cache.announce(path, first, { ...Route.default, cost: 1n }); + expect(instance.validity?.signal).toBe(signal); + expect(signal?.aborted).toBe(false); + cache.withdraw(path, first); + expect(signal?.aborted).toBe(true); + held.close(); + source.close(); +}); + +test("a blind selection retires on discovery while explicit epochs need no path gate", () => { + const cache = new BroadcastCache(); + const path = Path.from("pool/job"); + const blind = cache.instance(path); + const source = new Producer(); + const held = cache.insert(path, blind, source.consume()); + const owner = {}; + cache.announce(Path.from("pool"), owner, { ...Route.default, epoch: Epoch.mint() }); + expect(blind.validity?.signal.aborted).toBe(true); + expect(cache.instance(path).validity).toBeUndefined(); + cache.withdraw(Path.from("pool"), owner); + expect(blind.validity?.signal.aborted).toBe(true); + expect(cache.instance(path).identity).not.toBe(blind.identity); + held.close(); + source.close(); +}); diff --git a/js/net/src/consume.ts b/js/net/src/consume.ts index c6e4a392d4..8f39f638dd 100644 --- a/js/net/src/consume.ts +++ b/js/net/src/consume.ts @@ -3,19 +3,39 @@ import type * as Epoch from "./epoch.ts"; import type { Route } from "./hop.ts"; import { compareRouteCandidates, coveringPrefixes } from "./internal.ts"; import type * as Path from "./path.ts"; -import type { Instance } from "./wire.ts"; +import { type Instance, sameInstance } from "./wire.ts"; -class Blind implements Instance { +class Validity { + #current = new AbortController(); + + get signal(): AbortSignal { + return this.#current.signal; + } + + update(allowed: boolean): void { + if (allowed) { + if (this.#current.signal.aborted) this.#current = new AbortController(); + } else if (!this.#current.signal.aborted) { + this.#current.abort(new Error("broadcast instance retired")); + } + } +} + +type Selection = Instance & { readonly validity?: Validity }; + +class Blind implements Selection { readonly identity = {}; readonly route = {}; + readonly validity = new Validity(); } /** Consumed broadcasts shared only within the selected serving instance. @internal */ export class BroadcastCache { - #cache = new Map>(); + #cache = new Map>(); #announced = new Map>(); #blind = new Map(); #blindPrefixes = new Map>(); + #paths = new Map>(); /** Record this interest's live advertisement, preserving identity across route updates. */ announce(path: Path.Valid, owner: object, route: Route, restart = false): void { @@ -28,6 +48,7 @@ export class BroadcastCache { } const identity = (!restart && entries.get(owner)?.identity) || {}; entries.set(owner, { identity, route }); + this.#update(path); } /** Withdraw this interest's advertisement without disturbing another interest's routes. */ @@ -35,10 +56,11 @@ export class BroadcastCache { const entries = this.#announced.get(path); entries?.delete(owner); if (entries?.size === 0) this.#announced.delete(path); + this.#update(path); } /** Resolve the same most-specific-prefix and route preference as the origin. */ - instance(path: Path.Valid): Instance { + instance(path: Path.Valid): Selection { for (const prefix of coveringPrefixes(path)) { const entries = this.#announced.get(prefix); if (!entries) continue; @@ -46,7 +68,18 @@ export class BroadcastCache { for (const entry of entries.values()) { if (!best || compareRouteCandidates(path, entry.route, best.route) < 0) best = entry; } - if (best) return best; + if (best) { + const cached = this.#cache.get(path)?.get(best.route.epoch ?? best.identity); + return { + ...best, + validity: + best.route.epoch === undefined + ? ((cached?.consumer.closed.peek() === undefined + ? cached?.selection.validity + : undefined) ?? new Validity()) + : undefined, + }; + } } return this.#blind.get(path) ?? new Blind(); } @@ -54,18 +87,27 @@ export class BroadcastCache { /** Clone the live cached consumer only when it serves the selected instance. */ get(path: Path.Valid, instance: Instance): broadcast.Consumer | undefined { const cached = this.#cache.get(path)?.get(instance.route.epoch ?? instance.identity); - return cached?.closed.peek() === undefined ? cached?.clone() : undefined; + return cached?.consumer.closed.peek() === undefined ? cached?.consumer.clone() : undefined; } /** Cache each live instance so a temporary route winner never displaces a held fallback. */ - insert(path: Path.Valid, instance: Instance, consumer: broadcast.Consumer): broadcast.Consumer { + insert(path: Path.Valid, instance: Selection, consumer: broadcast.Consumer): broadcast.Consumer { let entries = this.#cache.get(path); if (!entries) { entries = new Map(); this.#cache.set(path, entries); } const key = instance.route.epoch ?? instance.identity; - entries.set(key, consumer); + const entry = { consumer, selection: instance }; + entries.set(key, entry); + for (const prefix of coveringPrefixes(path)) { + let paths = this.#paths.get(prefix); + if (!paths) { + paths = new Set(); + this.#paths.set(prefix, paths); + } + paths.add(path); + } if (instance instanceof Blind) { this.#blind.set(path, instance); for (const prefix of coveringPrefixes(path)) { @@ -78,9 +120,10 @@ export class BroadcastCache { } } void consumer.closed.then(() => { - if (entries.get(key) !== consumer) return; + instance.validity?.update(false); + if (entries.get(key) !== entry) return; entries.delete(key); - if (entries.size === 0 && this.#cache.get(path) === entries) this.#cache.delete(path); + if (entries.size === 0 && this.#cache.get(path) === entries) this.#forgetPath(path); if (this.#blind.get(path) === instance) this.#forgetBlind(path); }); return consumer; @@ -89,14 +132,33 @@ export class BroadcastCache { #forgetBlind(path: Path.Valid): void { const blind = this.#blind.get(path); if (!blind) return; + blind.validity.update(false); this.#blind.delete(path); const entries = this.#cache.get(path); entries?.delete(blind.identity); - if (entries?.size === 0) this.#cache.delete(path); + if (entries?.size === 0) this.#forgetPath(path); for (const prefix of coveringPrefixes(path)) { const paths = this.#blindPrefixes.get(prefix); paths?.delete(path); if (paths?.size === 0) this.#blindPrefixes.delete(prefix); } } + + #forgetPath(path: Path.Valid): void { + this.#cache.delete(path); + for (const prefix of coveringPrefixes(path)) { + const paths = this.#paths.get(prefix); + paths?.delete(path); + if (paths?.size === 0) this.#paths.delete(prefix); + } + } + + #update(prefix: Path.Valid): void { + for (const path of this.#paths.get(prefix) ?? []) { + const selected = this.instance(path); + for (const entry of this.#cache.get(path)?.values() ?? []) { + entry.selection.validity?.update(sameInstance(entry.selection, selected)); + } + } + } } diff --git a/js/net/src/ietf/adapter.test.ts b/js/net/src/ietf/adapter.test.ts index 934cd66ecc..6d55936b4d 100644 --- a/js/net/src/ietf/adapter.test.ts +++ b/js/net/src/ietf/adapter.test.ts @@ -542,3 +542,30 @@ test("more requests than the setup window complete on one session", async () => await reading; } }); + +// Canceling the wait must remove the allocation itself, not just stop awaiting it. +test("a canceled request ID wait never spends a later grant", async () => { + for (const [version, alpn] of WINDOW_DRAFTS) { + const pair = createMockTransportPair(alpn); + const control = await Stream.open(pair.server, { version }); + const adapter = new ControlStreamAdapter(pair.server, control, version, 0n, true); + const running = adapter.run().catch(() => undefined); + const peer = await Stream.accept(pair.client, version); + if (!peer) throw new Error("no control stream"); + try { + const signal = new AbortController(); + const canceled = adapter.nextRequestId(signal.signal); + signal.abort(new Error("retired")); + await expect(canceled).rejects.toThrow("retired"); + await peer.writer.u53(MaxRequestId.id); + await new MaxRequestId({ requestId: 4n }).encode(peer.writer, version); + expect(await adapter.nextRequestId()).toBe(0n); + await expect(adapter.nextRequestId(signal.signal)).rejects.toThrow("retired"); + expect(await adapter.nextRequestId()).toBe(2n); + } finally { + peer.close(); + adapter.close(); + void running; + } + } +}); diff --git a/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index b5811b928a..ba0b429b45 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -3,6 +3,7 @@ import { Mutex } from "async-mutex"; import type { Drain } from "../connection/goaway.ts"; import { error, ProtocolViolation, SessionCode } from "../error.ts"; import { Reader, Stream, type Writer } from "../stream.ts"; +import { untilAborted } from "../util/abort.ts"; import * as Varint from "../varint.ts"; import { GoAway } from "./goaway.ts"; import * as Namespace from "./namespace.ts"; @@ -31,7 +32,7 @@ export interface Session { openBi(): Stream | Promise; openNativeBi?(): Promise; acceptBi(): Promise; - nextRequestId(): Promise; + nextRequestId(signal?: AbortSignal): Promise; close(): void; readonly version: IetfVersion; } @@ -61,7 +62,8 @@ export class NativeSession implements Session { return Stream.accept(this.#quic, this.version); } - async nextRequestId(): Promise { + async nextRequestId(signal?: AbortSignal): Promise { + signal?.throwIfAborted(); const id = this.#requestId; this.#requestId += 2n; return id; @@ -142,7 +144,7 @@ export class ControlStreamAdapter implements Session { // delivered to the wrong virtual stream). #requestId: bigint; #maxRequestId: bigint; - #maxRequestIdResolves: (() => void)[] = []; + #maxRequestIdResolves = new Set<() => void>(); // The peer's requests, flow controlled as moq-net does: at most `#peerWindow` open at // once, every ID below `#peerMax`, and once half the window has closed the bound rises @@ -279,19 +281,25 @@ export class ControlStreamAdapter implements Session { } /** - * Allocate the next request ID, blocking if flow control limit reached. + * Allocate the next request ID, canceling before allocation while waiting for credit. + * A returned legacy ID is spent and its request must be sent, even if then canceled. */ - async nextRequestId(): Promise { + async nextRequestId(signal?: AbortSignal): Promise { for (;;) { + signal?.throwIfAborted(); if (this.#closed) return undefined; const id = this.#requestId; if (id < this.#maxRequestId) { this.#requestId += 2n; return id; } - await new Promise((resolve) => { - this.#maxRequestIdResolves.push(resolve); - }); + const { promise, resolve } = Promise.withResolvers(); + this.#maxRequestIdResolves.add(resolve); + try { + await untilAborted(promise, signal); + } finally { + this.#maxRequestIdResolves.delete(resolve); + } } } @@ -362,7 +370,7 @@ export class ControlStreamAdapter implements Session { case Route.MaxRequestId: this.#maxRequestId = requestId; for (const resolve of this.#maxRequestIdResolves) resolve(); - this.#maxRequestIdResolves = []; + this.#maxRequestIdResolves.clear(); break; } } @@ -923,6 +931,6 @@ export class ControlStreamAdapter implements Session { // Unblock maxRequestId waiters for (const resolve of this.#maxRequestIdResolves) resolve(); - this.#maxRequestIdResolves = []; + this.#maxRequestIdResolves.clear(); } } diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index 6d1f4ad260..fa22122500 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -17,7 +17,7 @@ import { ObjectDatagram } from "./datagram.ts"; import { encodeObjectExtensions, type GroupFlags, Group as GroupMessage } from "./object.ts"; import type { Properties } from "./properties.ts"; import { PublishNamespace, PublishNamespaceUpdate } from "./publish_namespace.ts"; -import { RequestError, RequestOk } from "./request.ts"; +import { MaxRequestId, RequestError, RequestOk } from "./request.ts"; import { Subscribe, SubscribeOk, Unsubscribe } from "./subscribe.ts"; import { SubscribeNamespace, SubscribeNamespaceEntry, SubscribeNamespaceEntryDone } from "./subscribe_namespace.ts"; import { Subscriber } from "./subscriber.ts"; @@ -1846,3 +1846,94 @@ for (const version of [ expect(await group?.readFrame()).toBeUndefined(); }); } + +// Retirement before allocation spends nothing; retirement after allocation drains that request. +test.each(["waiting", "allocated"] as const)( + "legacy retirement while %s keeps subsequent request IDs contiguous", + async (phase) => { + 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, phase === "waiting" ? 0n : 4n, true); + const running = session.run().catch(() => undefined); + const peer = await Stream.accept(pair.client, version); + if (!peer) throw new Error("no control stream"); + const entered = Promise.withResolvers(); + const allocated = Promise.withResolvers(); + const resume = Promise.withResolvers(); + const allocate = session.nextRequestId.bind(session); + const intercepted = spyOn(session, "nextRequestId").mockImplementationOnce(async (signal) => { + entered.resolve(); + const id = await allocate(signal); + allocated.resolve(); + await resume.promise; + return id; + }); + const subscriber = new Subscriber({ session }); + const announced = subscriber.announced(); + const path = Path.from("room"); + const held = subscriber.consume(path); + const old = held + .track("video") + .info() + .then( + () => undefined, + (err: unknown) => err, + ); + await entered.promise; + if (phase === "allocated") await allocated.promise; + + const source = createMockTransportPair(ALPN.DRAFT_15); + const sourceStream = await Stream.open(source.server, { version }); + const sourcePeer = await Stream.accept(source.client, version); + if (!sourcePeer) throw new Error("no announcement stream"); + const advertising = subscriber.runPublishNamespace( + new PublishNamespace({ requestId: 1n, trackNamespace: path }), + sourceStream, + ); + let fresh: ReturnType | undefined; + try { + expect(await sourcePeer.reader.u53()).toBe(RequestOk.id); + await RequestOk.decode(sourcePeer.reader, version); + await announced.next(); + resume.resolve(); + expect(await old).toBeInstanceOf(Error); + if (phase === "waiting") { + await peer.writer.u53(MaxRequestId.id); + await new MaxRequestId({ requestId: 4n }).encode(peer.writer, version); + } else { + expect(await peer.reader.u53()).toBe(Subscribe.id); + const canceled = await Subscribe.decode(peer.reader, version); + expect(canceled.requestId).toBe(0n); + expect(await peer.reader.u53()).toBe(Unsubscribe.id); + expect((await Unsubscribe.decode(peer.reader, version)).requestId).toBe(0n); + } + fresh = subscriber.consume(path); + const info = fresh + .track("video") + .info() + .catch((err: unknown) => err); + expect(await peer.reader.u53()).toBe(Subscribe.id); + const request = await Subscribe.decode(peer.reader, version); + expect(request.requestId).toBe(phase === "waiting" ? 0n : 2n); + await peer.writer.u53(SubscribeOk.id); + await new SubscribeOk({ + requestId: request.requestId, + trackAlias: request.requestId, + properties: {}, + }).encode(peer.writer, version); + expect(await info).not.toBeInstanceOf(Error); + } finally { + intercepted.mockRestore(); + resume.resolve(); + fresh?.close(); + held.close(); + announced.close(); + sourcePeer.close(); + await advertising; + peer.close(); + session.close(); + void running; + } + }, +); diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 81b8e7fb3f..fd33200d55 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -24,6 +24,7 @@ import { type Cursor, Reader, type Stream, UnexpectedEnd } from "../stream.ts"; import { Tail } from "../tail.ts"; import { Milli, type Timescale, type Timestamp } from "../time.ts"; import type * as track from "../track.ts"; +import { untilAborted } from "../util/abort.ts"; import { TimeoutError, withTimeout } from "../util/timeout.ts"; import { overrideBroadcastWire, wireOf } from "../wire.ts"; import type { Session } from "./adapter.ts"; @@ -573,23 +574,20 @@ export class Subscriber { */ consume(path: Path.Valid): broadcast.Consumer { const instance = this.#consumes.instance(path); - return this.#consumes.get(path, instance) ?? this.#consumes.insert(path, instance, this.#createConsume(path)); + return this.#consumes.get(path, instance) ?? this.#createConsume(path, instance); } - #createConsume(path: Path.Valid): broadcast.Consumer { - // moq-transport has no one-shot group fetch; ConsumeBroadcast rejects it. Track info - // is resolved by the subscribe path (the inherited resolveTrackInfo). + #createConsume(path: Path.Valid, instance: ReturnType): broadcast.Consumer { + // moq-transport resolves metadata through SUBSCRIBE and has no one-shot group fetch here. const consumer = new ConsumeBroadcast(); - void (async () => { for (;;) { const request = await wireOf(consumer).requested(); if (!request) break; - void this.#runSubscribe(path, request); + void this.#runSubscribe(path, request, instance.validity?.signal); } })(); - - return consumer; + return this.#consumes.insert(path, instance, consumer); } // The adapter is gone. If the transport has already closed, that close is the @@ -604,14 +602,23 @@ export class Subscriber { ]); } - async #runSubscribe(broadcast: Path.Valid, request: track.Request) { + async #runSubscribe(broadcast: Path.Valid, request: track.Request, valid?: AbortSignal) { const refused = unauthorized(broadcast); if (this.#denied(broadcast)) { request.reject(refused); return; } - const requestId = await this.#session.nextRequestId(); + let requestId: bigint | undefined; + try { + valid?.throwIfAborted(); + // The allocator cancels its own wait. Once it returns an ID, setup must spend + // it on the wire before cancellation, or the next legacy request would skip it. + requestId = await this.#session.nextRequestId(valid); + } catch (err) { + request.reject(error(err)); + return; + } if (requestId === undefined) { request.reject(await this.#closedSession()); return; @@ -657,7 +664,7 @@ export class Subscriber { `subscribe timed out after ${SUBSCRIBE_OK_TIMEOUT_MS}ms waiting for SUBSCRIBE_OK (browser stream limit reached?)`, ); for (;;) { - const result = await race([accepted, waitAbandoned()]); + const result = await untilAborted(race([accepted, waitAbandoned()]), valid); if (result !== null) { stream = result.stream; trackAlias = result.alias; @@ -860,7 +867,11 @@ export class Subscriber { // The timeout can fire while the open is still in flight, in which case cleanup ran // with nothing to release. Bail now that the stream is recorded, so settling this // promise hands it back to be torn down. - if (state.cancelled) throw new Error("subscribe cancelled before it was sent"); + // A legacy ID is already spent and cannot be skipped. Its virtual open is local, + // so finish this request and let cleanup send UNSUBSCRIBE, even if retirement + // won the await above. No response data is released from the canceled setup. + const legacy = version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16; + if (state.cancelled && !legacy) throw new Error("subscribe cancelled before it was sent"); await state.stream.writer.u53(Subscribe.id); const msg = new Subscribe({ diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 499a3f782a..3f37884050 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -898,13 +898,13 @@ function fakeSession(park: number[] = [], stall: number[] = []) { return { quic, streams, opened }; } -async function answerTrackInfo(stream: FakeStream): Promise { +async function answerTrackInfo(stream: FakeStream, version: Version = Version.DRAFT_05): Promise { const chunks: Uint8Array[] = []; const writer = new Writer( new WritableStream({ write: (chunk) => void chunks.push(new Uint8Array(chunk)) }), - Version.DRAFT_05, + version, ); - await new TrackInfo({}).encode(writer, Version.DRAFT_05); + await new TrackInfo({}).encode(writer, version); for (const chunk of chunks) stream.inbound.enqueue(chunk); stream.inbound.close(); } @@ -1340,6 +1340,114 @@ test.each(["initial", "track", "fetch", "body"] as const)( }, ); +// Serve announcements and metadata on separate mock streams, with no wall-clock wait. +test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( + "a retired epochless handle refuses new metadata on %s", + async (version) => { + const { quic, streams } = fakeSession(); + const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); + const announced = subscriber.announced(); + await drainUntil(() => streams.length === 1); + const send = async (encode: (writer: Writer) => Promise) => { + const writer = new Writer( + new WritableStream({ + write: (chunk) => streams[0].inbound.enqueue(chunk), + }), + version, + ); + await encode(writer); + }; + const path = Path.from("pool"); + await send((writer) => new AnnounceOk(PEER, 0).encode(writer, version)); + await send((writer) => + encodeAnnounceBroadcast(writer, { status: "active", suffix: path, hops: [PUBLISHER_A] }, version), + ); + await announced.next(); + const held = subscriber.consume(Path.from("pool/job")); + const track = held.track("video"); + await send((writer) => + encodeAnnounceBroadcast( + writer, + version === Version.DRAFT_05 ? { status: "ended", suffix: path } : { status: "endedId", id: 0n }, + version, + ), + ); + await announced.next(); + await send((writer) => + encodeAnnounceBroadcast(writer, { status: "active", suffix: path, hops: [PUBLISHER_B] }, version), + ); + await announced.next(); + let settled = false; + const result = track.info().then( + (info) => { + settled = true; + return info; + }, + (err: unknown) => { + settled = true; + return err; + }, + ); + await drainUntil(() => settled || streams.length === 2); + // The unfixed implementation asks B and accepts its metadata. + if (streams.length === 2) await answerTrackInfo(streams[1], version); + expect(await result).toBeInstanceOf(Error); + expect(streams.length).toBe(1); + expect(held.closed.peek()).toBeUndefined(); + held.close(); + announced.close(); + subscriber.close(); + }, +); + +test.each(["track", "fetch", "open"] as const)( + "retirement settles coalesced fetches during %s setup", + async (stage) => { + const { quic, streams } = fakeSession(stage === "open" ? [1] : []); + const subscriber = new Subscriber(quic, Version.DRAFT_06, HopSchema.parse(1n)); + const retired = new AbortController(); + const consumed = { id: 1, valid: retired.signal }; + const path = Path.from("room"); + const first = subscriber.fetchGroup(path, "video", 0, {}, consumed).catch((err: unknown) => err); + const second = subscriber.fetchGroup(path, "video", 0, {}, consumed).catch((err: unknown) => err); + await drainUntil(() => streams.length === 1); + if (stage !== "track") { + await answerTrackInfo(streams[0], Version.DRAFT_06); + await drainUntil(() => streams.length === 2); + } + const cause = new Error("broadcast instance retired"); + retired.abort(cause); + expect(await first).toBe(cause); + expect(await second).toBe(cause); + if (stage === "open") streams[1].release(); + const active = streams[stage === "track" ? 0 : 1]; + expect(await active.aborted).toBeDefined(); + subscriber.close(); + }, +); + +test("retirement preserves an accepted fetch and its coalesced reads", async () => { + const { quic, streams } = fakeSession(); + const subscriber = new Subscriber(quic, Version.DRAFT_06, HopSchema.parse(1n)); + const retired = new AbortController(); + const consumed = { id: 1, valid: retired.signal }; + const path = Path.from("room"); + const fetch = subscriber.fetchGroup(path, "video", 0, {}, consumed); + await drainUntil(() => streams.length === 1); + await answerTrackInfo(streams[0], Version.DRAFT_06); + await drainUntil(() => streams.length === 2); + await streams[1].reading; + streams[1].inbound.enqueue(new Uint8Array([0, 4, ...new TextEncoder().encode("old!")])); + const first = await fetch; + retired.abort(new Error("broadcast instance retired")); + const cached = await subscriber.fetchGroup(path, "video", 0, {}, consumed); + expect(streams.length).toBe(2); + expect((await first.readFrame())?.payload).toEqual((await cached.readFrame())?.payload); + first.close(); + cached.close(); + subscriber.close(); +}); + // Drive discovery and inspect the TRACK request before answering it. Waiting for the // reader means the whole request has reached the wire, without a clock-based settle. test("consumes resolve the serving prefix and pin its epoch", async () => { @@ -1423,3 +1531,50 @@ test.each([Version.DRAFT_06, Version.DRAFT_07])( subscriber.close(); }, ); + +test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( + "pending metadata cancels on selection change and a live fallback can request again on %s", + async (version) => { + const { quic, streams } = fakeSession(); + const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); + const announced = subscriber.announced(); + await drainUntil(() => streams.length === 1); + const peer = new Writer( + new WritableStream({ + write: (chunk) => streams[0].inbound.enqueue(new Uint8Array(chunk)), + }), + version, + ); + await new AnnounceOk(PEER, 0).encode(peer, version); + const pool = Path.from("pool"); + const job = Path.from("pool/job"); + await encodeAnnounceBroadcast(peer, { status: "active", suffix: pool, hops: [PUBLISHER_A] }, version); + await announced.next(); + const held = subscriber.consume(job); + const pending = held + .track("pending") + .info() + .catch((err: unknown) => err); + await drainUntil(() => streams.length === 2); + await streams[1].reading; + await encodeAnnounceBroadcast(peer, { status: "active", suffix: job, hops: [PUBLISHER_B] }, version); + await announced.next(); + expect(await pending).toBeInstanceOf(Error); + await streams[1].aborted; + await expect(held.track("during").info()).rejects.toThrow(); + expect(streams.length).toBe(2); + await encodeAnnounceBroadcast( + peer, + version === Version.DRAFT_05 ? { status: "ended", suffix: job } : { status: "endedId", id: 1n }, + version, + ); + await announced.next(); + const info = held.track("after").info(); + await drainUntil(() => streams.length === 3); + await answerTrackInfo(streams[2], version); + expect((await info).priority).toBe(0); + held.close(); + announced.close(); + subscriber.close(); + }, +); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 28f286df46..19cfcca08e 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -110,6 +110,7 @@ interface SubscribeSetup { interface Consumed { readonly epoch?: Epoch.Valid; readonly id?: number; + readonly valid?: AbortSignal; } interface SubscribeEntry { @@ -589,30 +590,31 @@ export class Subscriber { */ consume(path: Path.Valid): broadcast.Consumer { const instance = this.#consumes.instance(path); - return ( - this.#consumes.get(path, instance) ?? - this.#consumes.insert(path, instance, this.#createConsume(path, instance.route.epoch)) - ); + return this.#consumes.get(path, instance) ?? this.#createConsume(path, instance); } - #createConsume(path: Path.Valid, epoch?: Epoch.Valid): broadcast.Consumer { - // A consumed broadcast resolves info() and fetchGroup() over the wire by reaching - // back into this Subscriber (see ConsumeBroadcast below), rather than the wire - // installing callbacks on the broadcast. - const consumer = new ConsumeBroadcast(this, path, { epoch, id: this.#consumeNext++ }); + #createConsume(path: Path.Valid, instance: ReturnType): broadcast.Consumer { + const consumed = { + epoch: instance.route.epoch, + id: this.#consumeNext++, + get valid() { + return instance.validity?.signal; + }, + }; + const consumer = new ConsumeBroadcast(this, path, consumed); void (async () => { for (;;) { const request = await wireOf(consumer).requested(); if (!request) break; - void this.#runSubscribe(path, epoch, request); + void this.#runSubscribe(path, consumed, request); } })(); - return consumer; + return this.#consumes.insert(path, instance, consumer); } - async #runSubscribe(broadcast: Path.Valid, epoch: Epoch.Valid | undefined, request: track.Request) { + async #runSubscribe(broadcast: Path.Valid, consumed: Consumed, request: track.Request) { const id = this.#subscribeNext++; const subscription = request.subscription; const initialBounds = groupBounds(subscription.groups); @@ -637,7 +639,7 @@ export class Subscriber { request.reject(refused); return; } - await this.#serveSubscribe(id, broadcast, epoch, request, state); + await this.#serveSubscribe(id, broadcast, consumed, request, state); } finally { disposeGrant?.(); } @@ -646,10 +648,11 @@ export class Subscriber { async #serveSubscribe( id: bigint, broadcast: Path.Valid, - epoch: Epoch.Valid | undefined, + consumed: Consumed, request: track.Request, state: SubscribeSetup, ) { + const { epoch, valid } = consumed; const subscription = request.subscription; // `timescale` stays undefined until TRACK_INFO (or, on older drafts, @@ -674,12 +677,19 @@ export class Subscriber { // so the timeout path can close them if they finish opening after the deadline, // 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 retire = () => { + state.cancel.abort(valid?.reason); + state.producer?.close(error(valid?.reason)); + state.stream?.abort(error(valid?.reason)); + }; + valid?.addEventListener("abort", retire); + if (valid?.aborted) retire(); const setup = this.#openSubscribe(state, msg, request, id, timescale); let opened: { stream: Stream; entry: SubscribeEntry }; try { opened = await withTimeout( - setup, + untilAborted(setup, valid), SUBSCRIBE_SETUP_TIMEOUT_MS, `subscribe timed out after ${SUBSCRIBE_SETUP_TIMEOUT_MS}ms waiting for the first response (browser stream limit reached?)`, ); @@ -691,6 +701,7 @@ export class Subscriber { state.revoked ?? (err instanceof TimeoutError ? controlTimeout(err) : await sessionCause(this.#quic, err)); state.cancel.abort(e); + valid?.removeEventListener("abort", retire); request.reject(e); this.#subscribes.delete(id); console.warn(`subscribe error: id=${id} broadcast=${broadcast} track=${request.name} error=${reason(e)}`); @@ -707,6 +718,7 @@ export class Subscriber { return; } + valid?.removeEventListener("abort", retire); const { stream, entry } = opened; const producer = entry.track; try { @@ -786,6 +798,7 @@ export class Subscriber { id: bigint, timescale: Signal, ): Promise<{ stream: Stream; entry: SubscribeEntry }> { + state.cancel.signal.throwIfAborted(); let producer: track.Producer; let drainOk = false; @@ -868,12 +881,22 @@ export class Subscriber { 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 opening = Stream.open(this.#quic, options); + let stream: Stream; + try { + stream = await untilAborted(opening, closed); + } catch (err) { + void opening.then( + (late) => late.abort(error(err)), + () => {}, + ); + throw err; + } const abort = () => stream.abort(error(closed.reason)); closed.addEventListener("abort", abort); try { closed.throwIfAborted(); - return await run(stream); + return await untilAborted(run(stream), closed); } catch (err) { stream.abort(error(err)); throw err; @@ -894,7 +917,7 @@ export class Subscriber { } // Resolve a track's immutable model info via a TRACK stream (lite-05+), for the - // ConsumeBroadcast backing track.Consumer.query(). On older drafts there's no TRACK + // ConsumeBroadcast backing track.Consumer.info(). On older drafts there's no TRACK // stream, so this rejects rather than fabricating defaults. The TRACK stream stays // open until `hold` aborts, so the publisher keeps the track for whoever asked here; // aborting it before TRACK_INFO resets the stream. @@ -942,7 +965,7 @@ export class Subscriber { consumer = group.mirror(); entry = { group, - accepted: this.#runFetch(broadcast, epoch, track, sequence, options.priority ?? 0, group), + accepted: this.#runFetch(broadcast, consumed, track, sequence, options.priority ?? 0, group), }; this.#fetches.set(key, entry); void group.closed.then(() => { @@ -964,7 +987,7 @@ export class Subscriber { // has abandoned is cancelled the same way. async #runFetch( broadcast: Path.Valid, - epoch: Epoch.Valid | undefined, + consumed: Consumed, track: string, sequence: number, priority: number, @@ -986,7 +1009,7 @@ export class Subscriber { // Lite has no FETCH_OK, so a publisher that never answers would hold the setup forever. // Subscriber.close() closing the group releases every caller at any stage, and resets // the streams the setup opened. - const setup = this.#fetchSetup(broadcast, epoch, track, sequence, priority, group); + const setup = this.#fetchSetup(broadcast, consumed, track, sequence, priority, group); let accepted: { stream: Stream; info: TrackInfo }; try { accepted = await untilAbandoned(group, setup); @@ -1010,15 +1033,17 @@ export class Subscriber { // Closing the group during that wait resets the stream. async #fetchSetup( broadcast: Path.Valid, - epoch: Epoch.Valid | undefined, + consumed: Consumed, track: string, sequence: number, priority: number, group: netGroup.Producer, ): Promise<{ stream: Stream; info: TrackInfo }> { + const { epoch, valid } = consumed; const cancel = new AbortController(); void group.closed.then((err) => cancel.abort(err ?? new StreamError(StreamCode.Cancel))); - const answered = this.#trackInfo(broadcast, epoch, track, cancel.signal); + const signal = AbortSignal.any([valid, cancel.signal]); + const answered = this.#trackInfo(broadcast, epoch, track, signal); // Only the properties are needed, so the TRACK stream closes however the wait ends. answered.then( ({ stream }) => stream.close(), @@ -1038,7 +1063,7 @@ export class Subscriber { await untilClosed(group, stream.reader.done()); return { stream, info }; }, - cancel.signal, + signal, ); } @@ -1500,7 +1525,7 @@ async function untilAbandoned(group: netGroup.Producer, step: Promise): Pr } /** - * A broadcast consumed from a lite session. It resolves `track.Consumer.query()` and + * A broadcast consumed from a lite session. It resolves `track.Consumer.info()` and * `.fetchGroup()` over the wire (lite-05+ TRACK / FETCH streams) by reaching into the * {@link Subscriber} it was opened from, the way the Rust `BroadcastConsumer` holds its * session. Live subscribes still flow through the inherited requested() queue. @@ -1513,7 +1538,18 @@ class ConsumeBroadcast extends broadcast.Consumer { constructor(subscriber: Subscriber, path: Path.Valid, consumed: Consumed, state?: never) { super(state); overrideBroadcastWire(this, { - resolveTrackInfo: (name, hold) => subscriber.resolveTrackInfo(path, name, consumed.epoch, hold), + resolveTrackInfo: async (name, hold) => { + if (!consumed.valid) return subscriber.resolveTrackInfo(path, name, consumed.epoch, hold); + // Retirement cancels setup, but a one-shot query still releases its TRACK + // as soon as it answers instead of holding it for the broadcast's lifetime. + const query = new AbortController(); + const signal = AbortSignal.any([consumed.valid, hold ?? query.signal]); + try { + return await subscriber.resolveTrackInfo(path, name, consumed.epoch, signal); + } finally { + if (!hold) query.abort(); + } + }, fetchGroup: (name, sequence, options) => subscriber.fetchGroup(path, name, sequence, options, consumed), }); this.#subscriber = subscriber; diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 93990ba0bc..18605c77e2 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -123,7 +123,6 @@ End+Start. The child quests own their focused regressions. ## Required -- [Retired requests](/quest/m0/broadcast-epoch/retired-requests.md) - a held broadcast or track cannot open unpinned wire work against its replacement - [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - a delayed group is delivered once even beyond the old 1024-entry history - [Catalog references](/quest/m0/broadcast-epoch/catalog-references.md) - players and exporters pin siblings consistently, with optional reference epochs and documented legacy best effort - [Publish catalog restart](/quest/m0/broadcast-epoch/publish-catalog-restart.md) - `@moq/publish` never reuses catalog group numbers under one name and epoch after a re-announce diff --git a/quest/m0/broadcast-epoch/catalog-references.md b/quest/m0/broadcast-epoch/catalog-references.md index c21c596b80..8c7de1f372 100644 --- a/quest/m0/broadcast-epoch/catalog-references.md +++ b/quest/m0/broadcast-epoch/catalog-references.md @@ -72,7 +72,7 @@ guarantee and the selected legacy best-effort behavior. - [Source pin](https://github.com/moq-dev/moq/pull/5252) - exporters share their resolved catalog handle - [JS consume identity](https://github.com/moq-dev/moq/pull/5254) - JS resolves the actual serving identity -- [Retired requests](/quest/m0/broadcast-epoch/retired-requests.md) - held handles cannot emit unpinned requests after retirement +- [Retired requests](https://github.com/moq-dev/moq/pull/5294) - held handles cannot emit unpinned requests after retirement - [Catalog rendition IDs](/quest/m1/catalog-track-id.md) - the adjacent optional wire-track field - [Catalog track identity](/quest/m1/catalog-tracks.md) - catalog updates cannot redefine a track's content - [IETF epochs](/quest/m1/ietf-epochs.md) - verified catalog identity over negotiated IETF sessions diff --git a/quest/m0/broadcast-epoch/retired-requests.md b/quest/m0/broadcast-epoch/retired-requests.md deleted file mode 100644 index 38e725752b..0000000000 --- a/quest/m0/broadcast-epoch/retired-requests.md +++ /dev/null @@ -1,65 +0,0 @@ -# [L] Retired handles never request a replacement's content - -## Goal - -A Rust or JS broadcast or track handle keeps its resolved instance for -TRACK, SUBSCRIBE, and FETCH. Once that instance retires, a new wire request -cannot resolve the replacement by path. Existing reads and cached content -from the old instance remain usable within their normal lifetime and budgets. - -## Plan - -Reproduced in the 2026-10-10 resume audit: resolve catalog and media from A, -publish B at the same path, wait for the replacement announcement, then -FETCH a missing group through A's held media track. Rust returns B's bytes -on lite-05/06, epochless lite-07, and IETF drafts 14-22. Explicit-epoch -lite-07 refuses the stale request. JS also accepts a new TRACK_INFO through -a held epochless broadcast after End+Start on lite-06/07. - -Decided by the maintainer in that audit: - -- This correctness fix gates m0. Supporting epochless peers on updated - relays remains required; neither new paths nor the IETF epoch extension - is a prerequisite. -- Preserve instance validity through every handle and deferred operation, - including a track handed out before retirement, coalesced requests, and - a request whose setup races withdrawal. A fresh request may proceed only - when it still names the held instance. An explicit wire epoch can prove - that; an epochless path after retirement cannot. -- Keep already-open reads sticky and allow old cached bytes. Do not close - every old subscription to prevent new requests. A rejected operation must - settle its waiter and release its demand, without migrating it to B. -- Same explicit epoch permits route replacement and recovery. Absent - epochs never prove continuity. Before an epochless peer observes a - replacement, its wire cannot identify the old instance; do not claim - this fixes arbitrary legacy caches or that unavoidable ambiguity. - -Fix validity at the request/source boundary shared by these operations, -rather than patching individual exporters. `Source` handle pinning and the -JS consume cache fix select the right initial handle but do not prevent a -surviving handle from opening new path-based wire work. Coordinate with -those active quests without duplicating their cache-selection changes. -IETF standalone FETCH resolves a path, whereas joining FETCH reads the -saved subscription cache; preserve that distinction and check both. - -Add mocked-time regressions with A still alive after B is announced: held -catalog plus held media FETCH, late track lookup, re-subscription, pending -TRACK/FETCH setup, concurrent old/new consumers, exact and covering routes, -and source withdrawal without replacement. Test matching, different, and -absent epochs. Keep the explicit lite-07 epoch control and the failing -lite-05/06, epochless lite-07, and IETF14-22 matrix in CI; cover supported -legacy request forms and refusals too. Run `just check` and -`just test interop --all`. Benchmark any routing or invalidation fanout over -both routes and held consumers. - -Public API: corrected request lifetime and refusal behavior; preserve -readable old subscriptions. Wire: none; published framing is unchanged. -Update existing identity and request-lifetime documentation inline. - -## Related - -- [Source pin](https://github.com/moq-dev/moq/pull/5252) - the exporter holds the exact catalog broadcast -- [JS consume identity](https://github.com/moq-dev/moq/pull/5254) - fresh consumers select the serving announcement's identity -- [JS restart keeps the request](https://github.com/moq-dev/moq/pull/5199) - an existing resolved request stays sticky -- [IETF epochs](/quest/m1/ietf-epochs.md) - explicit wire identity removes the epochless ambiguity -- [Pipelined first FETCH](/quest/m1/pipeline-requests/fetch.md) - staged requests must retain the same instance validity diff --git a/quest/m1/announce-takeover.md b/quest/m1/announce-takeover.md index 5720d91287..3f1b732b31 100644 --- a/quest/m1/announce-takeover.md +++ b/quest/m1/announce-takeover.md @@ -93,5 +93,5 @@ ANNOUNCE_TAKEOVER; published versions retain their framing. - [IETF takeover](/quest/m1/ietf-takeover.md) - negotiated IETF counterpart - [Catalog references](/quest/m0/broadcast-epoch/catalog-references.md) - a proven continuation preserves the same pin -- [Retired requests](/quest/m0/broadcast-epoch/retired-requests.md) - replacements never substitute content into held handles +- [Retired requests](https://github.com/moq-dev/moq/pull/5294) - replacements never substitute content into held handles - [Identity and recovery guide](/quest/m1/identity-guide.md) - explains takeover alongside restart and legacy behavior diff --git a/quest/m1/fetch-max-delay.md b/quest/m1/fetch-max-delay.md index 067b7b6fa5..64ea0adf8c 100644 --- a/quest/m1/fetch-max-delay.md +++ b/quest/m1/fetch-max-delay.md @@ -71,4 +71,4 @@ exemption and the distinction between age and upstream wait budgets. - [IETF fill timeout](/quest/m1/ietf-fill-timeout.md) - upstream waiting bounded through the existing FETCH parameter - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - historical ranges retain this optional budget and publisher clamp -- [Retired requests](/quest/m0/broadcast-epoch/retired-requests.md) - identity stays pinned independently of age +- [Retired requests](https://github.com/moq-dev/moq/pull/5294) - identity stays pinned independently of age diff --git a/quest/m1/js-group-handover.md b/quest/m1/js-group-handover.md index 40e0fe60bb..7ff59c9dcb 100644 --- a/quest/m1/js-group-handover.md +++ b/quest/m1/js-group-handover.md @@ -70,6 +70,6 @@ subscriptions across a swap. Report it in the PR. ## Related -- [Retired requests](/quest/m0/broadcast-epoch/retired-requests.md) - old handles cannot open unpinned requests after an identity change +- [Retired requests](https://github.com/moq-dev/moq/pull/5294) - old handles cannot open unpinned requests after an identity change - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - a terminal group disposition stops recovery even for the newest group - [FETCH max-delay](/quest/m1/fetch-max-delay.md) - recovery fetches retain the reader's budget diff --git a/quest/m1/pipeline-requests/fetch.md b/quest/m1/pipeline-requests/fetch.md index b8a23f2b11..42034dc7f2 100644 --- a/quest/m1/pipeline-requests/fetch.md +++ b/quest/m1/pipeline-requests/fetch.md @@ -63,5 +63,5 @@ Public API: none. Wire: none (ordering only). - [Pipelined SUBSCRIBE](/quest/m1/pipeline-requests/subscribe.md) - the same change for subscriptions - [moq-transport ranges](/quest/m1/subscribe-ranges/ietf.md) - adjacent code in `ietf/subscriber.rs` and `model/origin.rs`; this lands first and ranges rebases -- [Retired requests](/quest/m0/broadcast-epoch/retired-requests.md) - staged requests retain their selected instance; retries cannot resolve a replacement by path +- [Retired requests](https://github.com/moq-dev/moq/pull/5294) - staged requests retain their selected instance; retries cannot resolve a replacement by path - [FETCH max-delay](/quest/m1/fetch-max-delay.md) - setup and recovery do not reset or bypass the reader's age budget diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index 0aaecfb0ed..8d629c23a4 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -121,3 +121,9 @@ harness = false [[bench]] name = "session" harness = false + +# Uses the hidden fuzz module to observe source selection without public API. +[[bench]] +name = "retirement" +harness = false +required-features = ["fuzz"] diff --git a/rs/moq-net/benches/retirement.rs b/rs/moq-net/benches/retirement.rs new file mode 100644 index 0000000000..5ff1db672e --- /dev/null +++ b/rs/moq-net/benches/retirement.rs @@ -0,0 +1,70 @@ +//! Cost of observing replacement at a covering source, swept over unrelated routes and held requests. +//! The route-only control models the old gate: a more-specific source does not retract the broad route. + +use criterion::{BatchSize, BenchmarkId, Criterion, criterion_group, criterion_main}; +use moq_net::{Path, fuzz::RequestValidity, kio, origin}; +use std::{hint::black_box, time::Duration}; + +fn retirement(c: &mut Criterion) { + let mut group = c.benchmark_group("retirement"); + group.sample_size(20); + group.warm_up_time(Duration::from_millis(500)); + group.measurement_time(Duration::from_secs(2)); + for selected in [false, true] { + for routes in [8, 32, 128] { + for consumers in [1, 8, 32] { + let name = if selected { "selection" } else { "route-only" }; + group.bench_with_input( + BenchmarkId::new(name, format!("{routes}x{consumers}")), + &(routes, consumers), + |b, &(routes, consumers)| { + b.iter_batched( + || { + let (origin, driver) = origin::Producer::new(origin::Config::default()); + let routes: Vec<_> = (0..routes) + .map(|i| origin.dynamic(format!("route/{i}"), origin::Route::default()).unwrap()) + .collect(); + let mut watches = Vec::new(); + let mut controls = Vec::new(); + let mut pending = Vec::new(); + for (i, route) in routes.iter().enumerate() { + for reader in 0..if i == 0 { consumers } else { 1 } { + if selected { + let watch = RequestValidity::new( + route, + &Path::new(&format!("route/{i}/jobs/{reader}")), + ); + pending.push(watch.start()); + watches.push(watch); + } else { + let control = kio::Producer::new(()); + pending.push(control.consume()); + controls.push(control); + } + } + } + (origin, driver, routes, watches, controls, pending) + }, + |fixture| { + let replacement = fixture.0.dynamic("route/0/jobs", origin::Route::default()).unwrap(); + let refused = fixture + .5 + .iter() + .take(consumers) + .filter(|request| request.is_closed()) + .count(); + assert_eq!(refused, if selected { consumers } else { 0 }); + black_box((fixture, replacement)) + }, + BatchSize::SmallInput, + ); + }, + ); + } + } + } + group.finish(); +} + +criterion_group!(benches, retirement); +criterion_main!(benches); diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index fbf13059ad..ada1fc1121 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -1171,6 +1171,24 @@ impl crate::transport::Error for NoError { } } +/// A source-selection watch for the private request-lifetime benchmark. +pub struct RequestValidity(crate::model::broadcast::RequestValidity); + +impl RequestValidity { + /// Watch which instance this route's peer selects for a path. + pub fn new(route: &crate::origin::Dynamic, path: &crate::Path) -> Self { + Self(crate::model::broadcast::RequestValidity::for_source(route, path)) + } + + /// Capture the lifetime of one pending request. + pub fn start(&self) -> kio::Consumer<()> { + match self.0.start() { + crate::model::broadcast::RequestValidity::Route(route) => route, + _ => unreachable!("a source watch always captures a route lease"), + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index e02933df2a..a02c7b3bc8 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -573,6 +573,8 @@ struct Advertised { #[derive(Clone)] pub(super) struct Subscriber { + // Scoped to a minted source on the clone that serves its requests. + validity: crate::model::broadcast::RequestValidity, // Arms the track-alias and request-id timeouts. runtime: crate::time::Clock, session: S, @@ -694,6 +696,7 @@ where auth: crate::auth::Handle::new(false), frames: Default::default(), announces: Default::default(), + validity: Default::default(), } } @@ -1780,6 +1783,17 @@ where // The handler exists before the requester sees the source, so a track it asks // for before `run_broadcast` first polls queues instead of failing `NotFound`. let dynamic = source.dynamic(); + let validity = self + .state + .lock() + .broadcasts + .get(&path) + .filter(|entry| entry.generation == generation) + .and_then(|entry| entry.dynamic.as_ref()) + .map(|route| crate::model::broadcast::RequestValidity::for_source(route, request.path())) + .unwrap_or_else(|| { + crate::model::broadcast::RequestValidity::token(Some(kio::Producer::new(()).consume())) + }); request.accept(&*source); // The namespace's token closes the source as it is retracted. If it was @@ -1790,12 +1804,14 @@ where .lock() .broadcasts .get(&path) + .filter(|entry| entry.generation == generation) .map(|entry| entry.live.consume()); if route.is_none() { source.close(); } - let this = self.clone(); + let mut this = self.clone(); + this.validity = validity; broadcasts.push(async move { if let Err(err) = this.run_broadcast(requested.borrow(), source, dynamic, route).await { tracing::debug!(%err, "error running broadcast"); @@ -1959,6 +1975,7 @@ where group_fetches: &mut TaskSet, gate: &mut crate::auth::Gate, ) -> Option { + let validity = self.validity.start(); let subscription = target.subscription(); let start = subscription.as_ref().and_then(|s| s.start); // A live join delivers nothing below the group SUBSCRIBE_OK names as Largest. @@ -1975,7 +1992,7 @@ where } }; - let request_id = match self.control.next_request_id(&self.runtime).await { + let request_id = match validity.run(self.control.next_request_id(&self.runtime)).await { Ok(id) => id, Err(err) => { target.fail(err); @@ -1983,7 +2000,10 @@ where } }; - let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { + let mut stream = match validity + .run(Stream::open(&mut self.session.clone(), self.version)) + .await + { Ok(s) => s, Err(err) => { tracing::debug!(%err, "failed to open subscribe stream"); @@ -2018,26 +2038,27 @@ where // Write Subscribe message. The aggregate is read now: a subscriber can join while // the request ID and stream were awaited, and nothing updates the priority after. let priority = target.subscription().map(|s| s.priority).unwrap_or(0); - let written = async { - stream.writer.varint(ietf::Subscribe::ID).await?; - stream - .writer - .encode(&ietf::Subscribe { - request_id, - track_namespace: broadcast_path.to_owned(), - track_name: track_name.into(), - subscriber_priority: super::priority::to_wire(priority), - group_order: GroupOrder::Descending, - filter: join.filter, - fill: join.fill, - // A resumed copy already knows the track, from the answer that accepted it. - properties_wanted: matches!(target, Target::Request(_)), - forward: true, - range_filters: false, - }) - .await - } - .await; + let written = validity + .run(async { + stream.writer.varint(ietf::Subscribe::ID).await?; + stream + .writer + .encode(&ietf::Subscribe { + request_id, + track_namespace: broadcast_path.to_owned(), + track_name: track_name.into(), + subscriber_priority: super::priority::to_wire(priority), + group_order: GroupOrder::Descending, + filter: join.filter, + fill: join.fill, + // A resumed copy already knows the track, from the answer that accepted it. + properties_wanted: matches!(target, Target::Request(_)), + forward: true, + range_filters: false, + }) + .await + }) + .await; if let Err(err) = written { tracing::debug!(%err, "failed to write subscribe"); self.remove_subscribe(request_id); @@ -2076,15 +2097,24 @@ where Gone, } + let mut retired = false; let setup = { let mut response = std::pin::pin!(self.read_subscribe_response(&mut stream)); loop { let setup = kio::wait(|waiter| { + let response = waiter.poll_future(response.as_mut()); + if let Poll::Ready(Err(err)) = response { + return Poll::Ready(Setup::Response(Err(err))); + } + if let Err(err) = validity.check(waiter) { + retired = true; + return Poll::Ready(Setup::Response(Err(err))); + } // An answer that has already arrived wins over the local terminals. Both can // be ready in one poll, and taking abandonment there would discard a response // the publisher has already sent: if it was a rejection, the request is gone // and cancelling it names a dead id back at a peer entitled to object. - if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) { + if let Poll::Ready(res) = response { return Poll::Ready(Setup::Response(res)); } let mut state = self.state.lock(); @@ -2167,6 +2197,9 @@ where } Err(err) => { tracing::debug!(%err, "subscribe response error"); + if retired { + self.cancel_subscribe(stream, request_id).await; + } if let Some(pending) = self.take_pending(request_id) { pending.reject(err.clone()); } @@ -2764,33 +2797,38 @@ where track_name: &str, request: track::Request, ) -> Option { - let request_id = match self.control.next_request_id(&self.runtime).await { + let validity = self.validity.start(); + let request_id = match validity.run(self.control.next_request_id(&self.runtime)).await { Ok(id) => id, Err(err) => { request.reject(err); return None; } }; - let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { + let mut stream = match validity + .run(Stream::open(&mut self.session.clone(), self.version)) + .await + { Ok(stream) => stream, Err(err) => { request.reject(err); return None; } }; - let written = async { - stream.writer.varint(ietf::TrackStatus::ID).await?; - stream - .writer - .encode(&ietf::TrackStatus { - request_id, - track_namespace: broadcast_path.to_owned(), - track_name: track_name.into(), - properties_wanted: true, - }) - .await - } - .await; + let written = validity + .run(async { + stream.writer.varint(ietf::TrackStatus::ID).await?; + stream + .writer + .encode(&ietf::TrackStatus { + request_id, + track_namespace: broadcast_path.to_owned(), + track_name: track_name.into(), + properties_wanted: true, + }) + .await + }) + .await; if let Err(err) = written { request.reject(err); return None; @@ -2804,6 +2842,9 @@ where let mut response = std::pin::pin!(self.read_track_status_response(&mut stream)); loop { let res = kio::wait(|waiter| { + if let Err(err) = validity.check(waiter) { + return Poll::Ready(Some(Err(err))); + } // An answer that already arrived wins over abandonment. if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) { return Poll::Ready(Some(res)); @@ -3743,11 +3784,12 @@ where request: group::Request, timescale: Option, ) { + let validity = self.validity.start(); let sequence = request.sequence(); let start = request.frame_start(); let draft20 = Filter::is_draft20(self.version); - let fetch_id = match self.control.next_request_id(&self.runtime).await { + let fetch_id = match validity.run(self.control.next_request_id(&self.runtime)).await { Ok(id) => id, Err(err) => return request.reject(err), }; @@ -3756,7 +3798,10 @@ where let slot = kio::Producer::new(GroupFetch::Pending); let registered = GroupFetchEntry::new(&self.state, fetch_id, slot.clone()); - let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { + let mut stream = match validity + .run(Stream::open(&mut self.session.clone(), self.version)) + .await + { Ok(stream) => stream, Err(err) => return request.reject(err), }; @@ -3791,32 +3836,42 @@ where }, }, }; - let res = async { - stream.writer.varint(ietf::Fetch::ID).await?; - stream - .writer - .encode(&ietf::Fetch { - request_id: fetch_id, - subscriber_priority: super::priority::to_wire(request.priority()), - group_order: GroupOrder::Ascending, - fetch_type, - range_filters: false, - fill_timeout: false, - // The copy learned the track before any fetch, from SUBSCRIBE_OK or - // TRACK_STATUS_OK, so a FETCH_OK repeating it is waste. - properties_wanted: false, - }) - .await?; - Ok::<_, Error>(()) - } - .await; + let res = validity + .run(async { + stream.writer.varint(ietf::Fetch::ID).await?; + stream + .writer + .encode(&ietf::Fetch { + request_id: fetch_id, + subscriber_priority: super::priority::to_wire(request.priority()), + group_order: GroupOrder::Ascending, + fetch_type, + range_filters: false, + fill_timeout: false, + // The copy learned the track before any fetch, from SUBSCRIBE_OK or + // TRACK_STATUS_OK, so a FETCH_OK repeating it is waste. + properties_wanted: false, + }) + .await?; + Ok::<_, Error>(()) + }) + .await; + let mut retired = false; let res = match res { Err(err) => Some(Err(err)), Ok(()) => { let mut response = std::pin::pin!(self.read_group_fetch_response(&mut stream)); kio::wait(|waiter| { + let response = waiter.poll_future(response.as_mut()); + if let Poll::Ready(Err(err)) = response { + return Poll::Ready(Some(Err(err))); + } + if let Err(err) = validity.check(waiter) { + retired = true; + return Poll::Ready(Some(Err(err))); + } // A refusal retires the peer's request, so it wins over abandonment. - if let Poll::Ready(res) = waiter.poll_future(response.as_mut()) { + if let Poll::Ready(res) = response { return Poll::Ready(Some(res)); } request.demand().poll_unused(waiter).map(|_| None) @@ -3836,7 +3891,11 @@ where Some(Err(err)) => { tracing::debug!(%err, group = sequence, "group fetch refused"); request.reject(err); - let _ = stream.writer.close().await; + if retired { + self.cancel_group_fetch(stream, fetch_id).await; + } else { + let _ = stream.writer.close().await; + } return; } }; @@ -5341,12 +5400,10 @@ mod tests { ); } - /// A retraction does not disturb subscriptions already in flight, and one whose - /// SUBSCRIBE_OK has not arrived yet is in flight too: the publisher may already be - /// serving it. The broadcast ending in that window must not abort the track or cancel - /// the subscription. + /// An unanswered epochless request cannot prove which instance the peer resolved. + /// Retiring its source cancels it even while the caller keeps its demand. #[moq_net_sim::test] - async fn a_retraction_before_subscribe_ok_keeps_the_subscription() { + async fn a_retirement_before_subscribe_ok_refuses_the_subscription() { const VERSION: Version = Version::Draft16; // A peer that accepts the stream and then says nothing at all. @@ -5368,6 +5425,9 @@ mod tests { Default::default(), ); + let route = kio::Producer::new(()); + subscriber.validity = crate::model::broadcast::RequestValidity::token(Some(route.consume())); + let producer = crate::broadcast::Info::default().produce(); let mut dynamic = producer.dynamic(); let consumer = producer.consume(); @@ -5382,27 +5442,11 @@ mod tests { // Let the SUBSCRIBE go out, then retract the broadcast before any response. settle().await; - producer.close(); - settle().await; - - assert!( - !serving.is_finished(), - "a retraction ended a subscription still in flight" - ); - assert_eq!( - occurrences(&log, &[ietf::Unsubscribe::ID as u8]), - 0, - "a retraction must not cancel a subscription still in flight", - ); - - // The reader leaving is still what ends it. - drop(subscription); - drop(track); - drop(consumer); - moq_net_sim::timeout(std::time::Duration::from_secs(1), serving) - .await - .expect("run_subscribe parked after its reader left") - .unwrap(); + drop(route); + serving.await.unwrap(); + assert!(matches!(subscription.await, Err(Error::Unroutable))); + assert!(occurrences(&log, &[ietf::Unsubscribe::ID as u8]) > 0); + assert_eq!(log.stops(), vec![crate::ietf::error::CANCELLED]); } /// The control messages that actually reached the wire, by type id. @@ -9680,6 +9724,7 @@ mod joining_fetch_tests { #[derive(Clone, Copy, PartialEq, Eq)] enum FetchStage { Unanswered, + Retired, Accepted, Receiving, Complete, @@ -9694,7 +9739,7 @@ mod joining_fetch_tests { let complete = matches!(stage, FetchStage::Complete | FetchStage::CompleteReset); let session = ScriptedSession::new(Vec::new()); let (tasks, _task_set) = TaskSet::new(); - let subscriber = Subscriber::new( + let mut subscriber = Subscriber::new( crate::time::Clock::sim(), session.clone(), crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), @@ -9707,6 +9752,8 @@ mod joining_fetch_tests { tasks, Default::default(), ); + let route = kio::Producer::new(()); + subscriber.validity = crate::model::broadcast::RequestValidity::token(Some(route.consume())); let track = track::Producer::new( std::sync::Arc::new(crate::broadcast::Info::default()), "video", @@ -9727,13 +9774,20 @@ mod joining_fetch_tests { assert!(futures::poll!(run.as_mut()).is_pending()); let slot = subscriber.state.lock().group_fetches[&RequestId(1)].clone(); - if stage == FetchStage::Unanswered { - drop(fetch); + if matches!(stage, FetchStage::Unanswered | FetchStage::Retired) { + if stage == FetchStage::Retired { + drop(route); + } else { + drop(fetch); + } assert!( futures::poll!(run.as_mut()).is_ready(), "{version:?}: unanswered FETCH stays alive" ); - assert!(matches!(outcome.read().rejected, Some(Error::Cancel))); + assert!(matches!( + outcome.read().rejected, + Some(Error::Cancel | Error::Unroutable) + )); } else { session.push(&message_bytes( ietf::FetchOk::ID, @@ -9861,6 +9915,13 @@ mod joining_fetch_tests { assert!(subscriber.state.lock().group_fetches.is_empty(), "request retired"); } + #[moq_net_sim::test] + async fn a_retired_group_fetch_before_fetch_ok_is_cancelled() { + for version in JOINING_DRAFTS { + abandon_group_fetch(version, FetchStage::Retired).await; + } + } + #[moq_net_sim::test] async fn an_abandoned_group_fetch_before_fetch_ok_is_cancelled() { for version in JOINING_DRAFTS { diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 806b4bb6da..e0b7fafd50 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -354,7 +354,7 @@ impl Subscriber { /// including a restart of one already attached: a connection on its way out must /// not win selection, however good the path it advertises looks. /// - /// The epoch is the one the peer announced at `path`: a restart carries none, + /// The epoch is the one the peer announced at `path`: an update carries none, /// since it never changes the content. fn announced_route( &self, @@ -423,7 +423,7 @@ impl Subscriber { tracing::debug!(route = %self.log_path(&path), hops = hops.len(), "update"); let metadata = self.announced_route(&path, hops, cost, link_cost, responder_origin, announced); - // A restart is a metadata update: the route keeps its prefix (and its + // A re-price keeps the route's prefix (and its // served paths) and re-prices in place. In-flight tracks keep flowing. if let Some(entry) = announced.attached(&path) { entry.update(metadata); @@ -1484,6 +1484,7 @@ impl AnnouncePrefix { /// A source the announce half minted for one requested path, for the driver to serve. struct MintedSource { + validity: crate::model::broadcast::RequestValidity, path: PathOwned, /// The publisher instance the route announced, asked for on every request. epoch: Option, @@ -1500,6 +1501,7 @@ struct MintedSource { /// retraction does not disturb subscriptions already in flight); the session /// dying drops them. struct SourceServe { + validity: crate::model::broadcast::RequestValidity, subscriber: Subscriber, path: PathOwned, epoch: Option, @@ -1518,6 +1520,7 @@ impl SourceServe { let closed = subscriber.session.clone(); Self { subscriber, + validity: minted.validity, path: minted.path, epoch: minted.epoch, source: minted.source, @@ -1549,6 +1552,7 @@ impl kio::Task for SourceServe { match self.dynamic.poll_requested_track(waiter) { Poll::Ready(Ok(request)) => { let serve = TrackServe { + validity: self.validity.clone(), subscriber: self.subscriber.clone(), path: self.path.clone(), epoch: self.epoch.clone(), @@ -1977,6 +1981,7 @@ mod tests { auth: crate::auth::Handle::new(false), }); let serve = TrackServe { + validity: Default::default(), subscriber, path: Path::new("room").to_owned(), epoch: None, @@ -2033,6 +2038,7 @@ mod tests { auth: crate::auth::Handle::new(false), }); let serve = TrackServe { + validity: Default::default(), subscriber, path: Path::new("room").to_owned(), epoch: None, @@ -2206,6 +2212,7 @@ mod tests { let run = TrackServeRun::new( TrackServe { + validity: Default::default(), subscriber: subscriber.clone(), path: Path::new("room").to_owned(), epoch: None, @@ -2264,6 +2271,7 @@ mod tests { let mut run = TrackServeRun::new( TrackServe { + validity: Default::default(), subscriber, path: Path::new("room").to_owned(), epoch: None, @@ -2314,6 +2322,7 @@ mod tests { }); let subscribes = subscriber.subscribes.clone(); let serve = TrackServe { + validity: Default::default(), subscriber, path: Path::new("room/host").to_owned(), epoch: None, @@ -2382,6 +2391,7 @@ mod tests { }); let subscribes = subscriber.subscribes.clone(); let serve = TrackServe { + validity: Default::default(), subscriber, path: Path::new("room/host").to_owned(), epoch: None, @@ -2450,6 +2460,7 @@ mod tests { Self { serve: TrackServe { + validity: Default::default(), subscriber, path: Path::new("room/host").to_owned(), epoch: None, @@ -4005,8 +4016,14 @@ impl Announced { // `NotFound`. Accepted before the push, so the requester holds the source // before its serve machine can see it unheld. let dynamic = source.dynamic(); + let validity = if entry.route.epoch.is_some() { + Default::default() + } else { + crate::model::broadcast::RequestValidity::for_source(&entry.dynamic, request.path()) + }; request.accept(&source); let _ = subscriber.sources.try_push(MintedSource { + validity, path, epoch: entry.route.epoch.clone(), source: crate::model::broadcast::SourceGuard::new(source), @@ -4092,7 +4109,7 @@ impl AnnouncedRoute { } } - /// Update the announced route in place (a restart). + /// Update the route metadata without changing its content identity. fn update(&mut self, route: crate::origin::Route) { self.route = route.clone(); self.drained = false; @@ -4150,6 +4167,7 @@ enum ServeEnd { /// the last one leaves) concurrently with any number of one-shot fetches. #[derive(Clone)] struct TrackServe { + validity: crate::model::broadcast::RequestValidity, subscriber: Subscriber, path: PathOwned, /// The publisher instance to ask for; the peer refuses a request for another. @@ -4336,6 +4354,7 @@ impl TrackServe { let session = self.subscriber.session.clone(); Establish { + validity: self.validity.start(), serve: self.clone(), closed: session.clone(), session, @@ -4435,6 +4454,7 @@ fn buffer_update( /// lite-05) wait for the SUBSCRIBE_OK. Resolves with the live [`SubStream`]; /// the caller deregisters the id on failure. struct Establish { + validity: crate::model::broadcast::RequestValidity, serve: TrackServe, session: S, // A dedicated close-watch handle for the SUBSCRIBE_OK wait. @@ -4460,6 +4480,7 @@ enum EstablishState { impl Establish { fn poll(&mut self, waiter: &kio::Waiter) -> Poll, Error>> { + self.validity.check(waiter)?; let mut cx = waiter.context(); loop { match &mut self.state { @@ -4728,6 +4749,7 @@ impl Drop for TrackServeRun { /// model's [`track::Info`], handing back the stream to hold. Lite05+ only. Bails if the /// session dies meanwhile. struct TrackInfoFetch { + validity: crate::model::broadcast::RequestValidity, session: S, // A dedicated close-watch handle for the read. closed: S, @@ -4744,6 +4766,7 @@ impl TrackInfoFetch { fn new(serve: &TrackServe) -> Self { let session = serve.subscriber.session.clone(); Self { + validity: serve.validity.start(), closed: session.clone(), session, state: TrackInfoState::Open, @@ -4755,6 +4778,7 @@ impl TrackInfoFetch { serve: &TrackServe, waiter: &kio::Waiter, ) -> Poll), Error>> { + self.validity.check(waiter)?; let mut cx = waiter.context(); loop { match &mut self.state { @@ -5239,6 +5263,7 @@ impl ServeLoop { /// comes from this track's TRACK_INFO (already known), and the group sequence /// is implicit from the request. struct FetchServeRun { + validity: crate::model::broadcast::RequestValidity, serve: TrackServe, session: S, timescale: Option, @@ -5288,6 +5313,7 @@ impl FetchServeRun { let session = serve.subscriber.session.clone(); let group = request.sequence(); Self { + validity: serve.validity.start(), serve, session, timescale, @@ -5340,6 +5366,19 @@ impl kio::Task for FetchServeRun { type Output = (); fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { + if self.state.request().is_some() + && let Err(err) = self.validity.check(waiter) + { + match std::mem::replace(&mut self.state, FetchRunState::Done) { + FetchRunState::Open { request } => request.expect("request pending").reject(err), + FetchRunState::Send { request, stream, .. } | FetchRunState::Answer { request, stream, .. } => { + stream.writer.abort(&err); + request.expect("request pending").reject(err); + } + _ => unreachable!(), + } + return Poll::Ready(()); + } let mut cx = waiter.context(); loop { // A fetch nobody waits on any more is cancelled upstream, so the publisher stops diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index bc0b2a3c81..2b5742f8ed 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -527,6 +527,57 @@ impl Drop for Alive { } } +/// Whether a path request still names the source that handed out its track. +#[derive(Clone, Default)] +pub(crate) enum RequestValidity { + #[default] + Unrestricted, + Route(kio::Consumer<()>), + Selected(std::sync::Arc), +} + +impl RequestValidity { + pub(crate) fn token(route: Option>) -> Self { + route.map(Self::Route).unwrap_or_default() + } + + pub(crate) fn for_source(route: &crate::origin::Dynamic, path: &crate::Path) -> Self { + route + .request_validity(path) + .map(|selected| Self::Selected(std::sync::Arc::new(selected))) + .unwrap_or_else(|| Self::Route(kio::Producer::new(()).consume())) + } + + /// Capture one selection period so a pending request never revives after a replacement. + pub(crate) fn start(&self) -> Self { + match self { + Self::Selected(selected) => Self::Route(selected.current()), + other => other.clone(), + } + } + + pub(crate) fn check(&self, waiter: &kio::Waiter) -> Result<(), crate::Error> { + match self { + Self::Route(route) if route.poll_closed(waiter).is_ready() => Err(crate::Error::Unroutable), + Self::Selected(_) => self.start().check(waiter), + _ => Ok(()), + } + } + + pub(crate) async fn run( + &self, + future: impl std::future::Future>, + ) -> Result { + let validity = self.start(); + let mut future = std::pin::pin!(future); + kio::wait(|waiter| { + validity.check(waiter)?; + waiter.poll_future(future.as_mut()) + }) + .await + } +} + #[cfg(test)] #[allow(missing_docs)] // test-only assertion helpers impl Producer { diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index b587f6299d..8832b3cbc7 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3112,7 +3112,7 @@ struct RouteNode { cursors_below: usize, /// Who is waiting on the routes covering this prefix: the fronts serving it /// and the requesters parked on it (see [`Watch`]). - watches: Vec<(u64, kio::Producer)>, + watches: Vec<(u64, WatchProducer)>, /// Watches at this node or below, so a route change walks only the subtrees /// holding one. watches_below: usize, @@ -3126,6 +3126,43 @@ struct Watched { generation: u64, } +enum WatchProducer { + Changes(kio::Producer), + Selection(kio::Producer), +} + +struct Selected { + path: PathOwned, + via: Hop, + instance: Instance, + current: Option>, +} + +/// The lifetime of path requests while this peer still selects the held instance. +pub(crate) struct Selection { + shared: kio::Shared, + path: PathOwned, + id: u64, + signal: kio::Consumer, +} + +impl Selection { + pub(crate) fn current(&self) -> kio::Consumer<()> { + self.signal + .read() + .current + .as_ref() + .map(kio::Producer::consume) + .unwrap_or_else(|| kio::Producer::new(()).consume()) + } +} + +impl Drop for Selection { + fn drop(&mut self) { + self.shared.lock().routes.remove_watch(&self.path, self.id); + } +} + /// A registration in the route table for changes to the routes covering one /// path. The table pokes it; the holder waits on it, so an announcement wakes /// only the fronts and requesters it can affect rather than every one of them. @@ -3226,24 +3263,48 @@ impl RouteNode { Some(result) } - /// Wake the watches at this node. - fn poke(&self) { + /// Wake route watchers, and revoke only leases whose selected instance changed. + fn poke(&self, routes: Option<&RouteTable>) { for (_, watch) in &self.watches { - if let Ok(mut watched) = watch.write() { - watched.generation += 1; + match watch { + WatchProducer::Changes(watch) => { + if let Ok(mut watched) = watch.write() { + watched.generation += 1; + } + } + WatchProducer::Selection(watch) => { + if let Ok(mut selected) = watch.write() { + let active = routes + .and_then(|routes| { + routes.best_route( + &selected.path, + Horizon { + exclude: None, + local: false, + upstream: false, + }, + &HashSet::new(), + |entry| entry.via == selected.via, + ) + }) + .is_some_and(|entry| entry.instance() == selected.instance); + if active != selected.current.is_some() { + selected.current = active.then(kio::Producer::default); + } + } + } } } } - /// Wake the watches at this node and below: a route here covers every one - /// of their paths. Skips subtrees holding none. - fn poke_below(&self) { + /// Wake only the watched subtrees covered by this route. + fn poke_below(&self, routes: &RouteTable) { if self.watches_below == 0 { return; } - self.poke(); + self.poke(Some(routes)); for child in self.children.values() { - child.poke_below(); + child.poke_below(routes); } } @@ -3268,6 +3329,37 @@ impl RouteNode { } impl RouteTable { + fn best_route( + &self, + path: &Path, + horizon: Horizon, + excluded: &HashSet, + only: impl Fn(&RouteEntry) -> bool, + ) -> Option<&RouteEntry> { + // Covering prefixes of one path form a chain, so the deepest node with a + // candidate holds the unique longest prefix; walking down, the last such + // node decides. + let (above, at) = self.split(path); + let mut best = None; + for node in above.into_iter().chain(at) { + let mut candidates = node + .entries + .iter() + .filter(|entry| entry.live()) + .filter(|entry| entry.scope.matches(path.as_str())) + .filter(|entry| horizon.admits(entry)) + .filter(|entry| !excluded.contains(&entry.id)) + .filter(|entry| only(entry)) + .peekable(); + if candidates.peek().is_some() { + best = candidates + .filter(|entry| entry.serves(path)) + .min_by_key(|entry| route_order(path, entry)); + } + } + best + } + /// The nodes above `path` and the node at it, as far as the table has them. /// The entries of those nodes are exactly the routes covering `path`. fn split(&self, path: &Path) -> (Vec<&RouteNode>, Option<&RouteNode>) { @@ -3369,7 +3461,10 @@ impl RouteTable { fn add_watch(&mut self, path: &Path, id: u64) -> kio::Consumer { let producer = kio::Producer::::default(); let consumer = producer.consume(); - self.root.reach(path.parts(), Below::WATCH).watches.push((id, producer)); + self.root + .reach(path.parts(), Below::WATCH) + .watches + .push((id, WatchProducer::Changes(producer))); consumer } @@ -3384,13 +3479,13 @@ impl RouteTable { /// Wake the watches of every path a route at `prefix` covers. fn poke_below(&self, prefix: &Path) { if let (_, Some(node)) = self.split(prefix) { - node.poke_below(); + node.poke_below(self); } } /// Wake every watch: the origin is tearing down. fn poke_all(&self) { - self.root.walk(&mut |node| node.poke()); + self.root.walk(&mut |node| node.poke(None)); } /// The cursors a route at `prefix` can present on: a cursor sees a route @@ -3676,28 +3771,7 @@ impl OriginState { excluded: &HashSet, only: impl Fn(&RouteEntry) -> bool, ) -> Option<&RouteEntry> { - // Covering prefixes of one path form a chain, so the deepest node with a - // candidate holds the unique longest prefix; walking down, the last such - // node decides. - let (above, at) = self.routes.split(path); - let mut best = None; - for node in above.into_iter().chain(at) { - let mut candidates = node - .entries - .iter() - .filter(|entry| entry.live()) - .filter(|entry| entry.scope.matches(path.as_str())) - .filter(|entry| horizon.admits(entry)) - .filter(|entry| !excluded.contains(&entry.id)) - .filter(|entry| only(entry)) - .peekable(); - if candidates.peek().is_some() { - best = candidates - .filter(|entry| entry.serves(path)) - .min_by_key(|entry| route_order(path, entry)); - } - } - best + self.routes.best_route(path, horizon, excluded, only) } /// Whether every cursor presenting `prefix` restarted it since @@ -3814,6 +3888,64 @@ pub struct Dynamic { } impl Dynamic { + pub(crate) fn request_validity(&self, path: &Path) -> Option { + let shared = &self.announcement.shared; + let mut table = shared.lock(); + if table.closed { + return None; + } + let entry = table.best_route( + path, + Horizon { + exclude: None, + local: false, + upstream: false, + }, + &HashSet::new(), + |entry| { + entry + .server + .as_ref() + .is_some_and(|server| server.same_channel(&self.state)) + }, + )?; + let via = entry.via; + let instance = entry.instance(); + let active = table + .best_route( + path, + Horizon { + exclude: None, + local: false, + upstream: false, + }, + &HashSet::new(), + |entry| entry.via == via, + ) + .is_some_and(|entry| entry.instance() == instance); + let producer = kio::Producer::new(Selected { + path: path.to_owned(), + via, + instance, + current: active.then(kio::Producer::default), + }); + let signal = producer.consume(); + let id = table.next_watch; + table.next_watch += 1; + table + .routes + .root + .reach(path.parts(), Below::WATCH) + .watches + .push((id, WatchProducer::Selection(producer))); + Some(Selection { + shared: shared.clone(), + path: path.to_owned(), + id, + signal, + }) + } + /// Retire an explicit peer withdrawal, hiding derived paths only once its /// last live advertisement at the prefix is gone. A lost session just drops. pub(crate) fn withdrawn(self) { @@ -9887,6 +10019,49 @@ mod tests { .unwrap(); } + #[test] + fn source_selection_leases_survive_reprices_but_never_revive_pending_work() { + let producer = origin(1).produce(); + let broad = producer.dynamic("room", Route::default().with_via(origin(2))).unwrap(); + let selected = broad.request_validity(&Path::new("room/a")).unwrap(); + let pending = selected.current(); + assert!(!pending.is_closed()); + broad.update(broad.route().with_cost(Cost::from(7))).unwrap(); + assert!(!pending.is_closed()); + // Another peer's winning route cannot change what a request on peer 2 names. + let other = producer + .dynamic("room/a", Route::default().with_via(origin(3))) + .unwrap(); + assert!(!pending.is_closed()); + let replacement = producer + .dynamic("room/a", Route::default().with_via(origin(2))) + .unwrap(); + assert!(pending.is_closed()); + assert!(selected.current().is_closed()); + // No waiter has polled between these changes: the old lease must still be dead. + drop(replacement); + let fresh = selected.current(); + assert!(!fresh.is_closed()); + assert!(pending.is_closed()); + drop(other); + assert!(!fresh.is_closed()); + drop(broad); + assert!(fresh.is_closed()); + } + + #[test] + fn source_selection_unregisters_without_retaining_the_source() { + let producer = origin(1).produce(); + let route = producer.dynamic("room", Route::default().with_via(origin(2))).unwrap(); + let path = Path::new("room/a"); + let selected = route.request_validity(&path).unwrap(); + let lease = selected.current(); + drop(selected); + assert!(lease.is_closed()); + let table = producer.shared.lock(); + assert_eq!(table.routes.root.watches_below, 0); + } + #[test] fn watch_wakes_only_for_covering_changes() { let producer = origin(1).produce(); diff --git a/rs/moq-net/tests/restart.rs b/rs/moq-net/tests/restart.rs index f5563210d3..e77c6140ee 100644 --- a/rs/moq-net/tests/restart.rs +++ b/rs/moq-net/tests/restart.rs @@ -407,3 +407,162 @@ async fn dynamic_epoch_second_route_keeps_winning() { second.write(0, "second:1"); assert_eq!(read(&mut sticky).await, "second:1"); } + +/// A held track must not fill an old instance's cache from its replacement. +async fn retired_fetch(version: &str, explicit: bool) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let viewer = produce_origin(2); + let _session = link(version, &publisher, &viewer).await; + let (a, b) = epochs(); + let old_route = if explicit { + origin::Route::default().with_epoch(a) + } else { + origin::Route::default() + }; + let new_route = if explicit { + origin::Route::default().with_epoch(b) + } else { + origin::Route::default() + }; + let old = publisher.publish("live", old_route).unwrap(); + let old_track = old.create_track("video", None).unwrap(); + let mut announced = viewer.consume().announced(); + assert!(matches!(next_event(&mut announced).await, announce::Event::Start(_))); + let (held, mut subscription) = subscribe(&viewer, "live").await; + let track = held.track("video").unwrap(); + write(&old_track, "old:0"); + assert_eq!(read(&mut subscription).await, "old:0"); + + let replacement = publisher.publish("live", new_route).unwrap(); + let replacement_track = replacement.create_track("video", None).unwrap(); + let _late = replacement + .create_track("late", Some(track::Info::default().with_priority(17))) + .unwrap(); + write(&replacement_track, "new:0"); + write(&replacement_track, "new:1"); + expect_restart(version, &mut announced, "live").await; + + let fetched = track.fetch_group(1, None).await; + assert!( + fetched.is_err(), + "{version}: retired handle fetched replacement content" + ); + let mut cached = track.fetch_group(0, None).await.expect("old cache stays readable"); + assert_eq!(cached.read_frame().await.unwrap().unwrap().payload.as_ref(), b"old:0"); + write(&old_track, "old:1"); + assert_eq!(read(&mut subscription).await, "old:1"); + if let Ok(late) = held.track("late") + && let Ok(info) = late.query().await + { + assert_ne!(info.priority, 17, "{version}: late track resolved replacement"); + } + drop(subscription); + settle().await; + let resumed = track.subscribe(None).await; + settle().await; + assert!( + replacement_track.subscription().is_none(), + "{version}: resubscribed to replacement" + ); + drop(resumed); + let mut cached = track + .fetch_group(0, None) + .await + .expect("old cache survives resubscribe refusal"); + assert_eq!(cached.read_frame().await.unwrap().unwrap().payload.as_ref(), b"old:0"); +} + +macro_rules! retired_fetch_tests { + ($($name:ident: $version:literal,)*) => { $( + #[moq_net_sim::test] + async fn $name() { + moq_net_sim::timeout(TEST_TIMEOUT, retired_fetch($version, false)).await.expect("timed out"); + } + )* }; +} + +retired_fetch_tests! { + retired_fetch_lite05: "moq-lite-05", + retired_fetch_lite06: "moq-lite-06", + retired_fetch_lite07: "moq-lite-07-wip", + retired_fetch_ietf14: "moq-transport-14", + retired_fetch_ietf15: "moq-transport-15", + retired_fetch_ietf16: "moq-transport-16", + retired_fetch_ietf17: "moq-transport-17", + retired_fetch_ietf18: "moq-transport-18", + retired_fetch_ietf19: "moq-transport-19", + retired_fetch_ietf20: "moq-transport-20", + retired_fetch_ietf21: "moq-transport-21", + retired_fetch_ietf22: "moq-transport-22", +} + +#[moq_net_sim::test] +async fn retired_fetch_explicit_epoch_control() { + moq_net_sim::timeout(TEST_TIMEOUT, retired_fetch("moq-lite-07-wip", true)) + .await + .expect("timed out"); +} + +#[moq_net_sim::test] +async fn withdrawal_refuses_new_fetches_but_keeps_open_reads() { + for version in [ + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-14", + "moq-transport-22", + ] { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let viewer = produce_origin(2); + let _session = link(version, &publisher, &viewer).await; + let source = publisher.publish("live", origin::Route::default()).unwrap(); + let produced = source.create_track("video", None).unwrap(); + let mut announced = viewer.consume().announced(); + assert!(matches!(next_event(&mut announced).await, announce::Event::Start(_))); + let (held, mut subscription) = subscribe(&viewer, "live").await; + let track = held.track("video").unwrap(); + write(&produced, "old:0"); + assert_eq!(read(&mut subscription).await, "old:0"); + source.unannounce(); + assert!(matches!(next_event(&mut announced).await, announce::Event::End(_))); + assert!(track.fetch_group(1, None).await.is_err(), "{version}"); + write(&produced, "old:1"); + assert_eq!(read(&mut subscription).await, "old:1"); + } +} + +#[moq_net_sim::test] +async fn a_more_specific_source_cannot_fill_a_held_covering_track() { + for name in [ + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-14", + "moq-transport-22", + ] { + let version: Version = name.parse().unwrap(); + let publisher = produce_origin(1); + let viewer = produce_origin(2); + let _session = link(version, &publisher, &viewer).await; + let old = Claim::new(&publisher, None); + let mut announced = viewer.consume().announced(); + assert!(matches!(next_event(&mut announced).await, announce::Event::Start(_))); + let (held, mut subscription) = subscribe(&viewer, "live/job").await; + let track = held.track("video").unwrap(); + old.write(0, "old:0"); + assert_eq!(read(&mut subscription).await, "old:0"); + let replacement = publisher.publish("live/job", origin::Route::default()).unwrap(); + let replacement_track = replacement.create_track("video", None).unwrap(); + write(&replacement_track, "new:0"); + write(&replacement_track, "new:1"); + assert!(matches!(next_event(&mut announced).await, announce::Event::Start(_))); + assert!( + track.fetch_group(1, None).await.is_err(), + "{version}: covering handle fetched replacement" + ); + old.write(0, "old:1"); + assert_eq!(read(&mut subscription).await, "old:1"); + } +} From 7ff0e0225c0a12897ca16c1c5eb914b33258d272 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 10:45:08 -0700 Subject: [PATCH 2/3] fix(net): send every allocated IETF request ID despite retirement Drafts 14-16 grant MAX_REQUEST_ID credit back only for requests the peer receives. Retirement now cancels only the wait for an ID; an allocated SUBSCRIBE, TRACK_STATUS or FETCH is always written, then canceled by the response wait. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/subscriber.rs | 224 +++++++++++++++++++++--------- 1 file changed, 157 insertions(+), 67 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index a02c7b3bc8..485f933582 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1992,6 +1992,9 @@ where } }; + // Retirement cancels only the wait for an ID. On drafts 14-16 the peer grants + // MAX_REQUEST_ID credit back only for requests it receives, so an allocated ID is + // always sent; the response wait below cancels it if the source retired meanwhile. let request_id = match validity.run(self.control.next_request_id(&self.runtime)).await { Ok(id) => id, Err(err) => { @@ -2000,10 +2003,7 @@ where } }; - let mut stream = match validity - .run(Stream::open(&mut self.session.clone(), self.version)) - .await - { + let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { Ok(s) => s, Err(err) => { tracing::debug!(%err, "failed to open subscribe stream"); @@ -2038,27 +2038,26 @@ where // Write Subscribe message. The aggregate is read now: a subscriber can join while // the request ID and stream were awaited, and nothing updates the priority after. let priority = target.subscription().map(|s| s.priority).unwrap_or(0); - let written = validity - .run(async { - stream.writer.varint(ietf::Subscribe::ID).await?; - stream - .writer - .encode(&ietf::Subscribe { - request_id, - track_namespace: broadcast_path.to_owned(), - track_name: track_name.into(), - subscriber_priority: super::priority::to_wire(priority), - group_order: GroupOrder::Descending, - filter: join.filter, - fill: join.fill, - // A resumed copy already knows the track, from the answer that accepted it. - properties_wanted: matches!(target, Target::Request(_)), - forward: true, - range_filters: false, - }) - .await - }) - .await; + let written = async { + stream.writer.varint(ietf::Subscribe::ID).await?; + stream + .writer + .encode(&ietf::Subscribe { + request_id, + track_namespace: broadcast_path.to_owned(), + track_name: track_name.into(), + subscriber_priority: super::priority::to_wire(priority), + group_order: GroupOrder::Descending, + filter: join.filter, + fill: join.fill, + // A resumed copy already knows the track, from the answer that accepted it. + properties_wanted: matches!(target, Target::Request(_)), + forward: true, + range_filters: false, + }) + .await + } + .await; if let Err(err) = written { tracing::debug!(%err, "failed to write subscribe"); self.remove_subscribe(request_id); @@ -2798,6 +2797,7 @@ where request: track::Request, ) -> Option { let validity = self.validity.start(); + // An allocated ID is always sent, as in `subscribe_once`. let request_id = match validity.run(self.control.next_request_id(&self.runtime)).await { Ok(id) => id, Err(err) => { @@ -2805,30 +2805,26 @@ where return None; } }; - let mut stream = match validity - .run(Stream::open(&mut self.session.clone(), self.version)) - .await - { + let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { Ok(stream) => stream, Err(err) => { request.reject(err); return None; } }; - let written = validity - .run(async { - stream.writer.varint(ietf::TrackStatus::ID).await?; - stream - .writer - .encode(&ietf::TrackStatus { - request_id, - track_namespace: broadcast_path.to_owned(), - track_name: track_name.into(), - properties_wanted: true, - }) - .await - }) - .await; + let written = async { + stream.writer.varint(ietf::TrackStatus::ID).await?; + stream + .writer + .encode(&ietf::TrackStatus { + request_id, + track_namespace: broadcast_path.to_owned(), + track_name: track_name.into(), + properties_wanted: true, + }) + .await + } + .await; if let Err(err) = written { request.reject(err); return None; @@ -3789,6 +3785,7 @@ where let start = request.frame_start(); let draft20 = Filter::is_draft20(self.version); + // An allocated ID is always sent, as in `subscribe_once`. let fetch_id = match validity.run(self.control.next_request_id(&self.runtime)).await { Ok(id) => id, Err(err) => return request.reject(err), @@ -3798,10 +3795,7 @@ where let slot = kio::Producer::new(GroupFetch::Pending); let registered = GroupFetchEntry::new(&self.state, fetch_id, slot.clone()); - let mut stream = match validity - .run(Stream::open(&mut self.session.clone(), self.version)) - .await - { + let mut stream = match Stream::open(&mut self.session.clone(), self.version).await { Ok(stream) => stream, Err(err) => return request.reject(err), }; @@ -3836,26 +3830,25 @@ where }, }, }; - let res = validity - .run(async { - stream.writer.varint(ietf::Fetch::ID).await?; - stream - .writer - .encode(&ietf::Fetch { - request_id: fetch_id, - subscriber_priority: super::priority::to_wire(request.priority()), - group_order: GroupOrder::Ascending, - fetch_type, - range_filters: false, - fill_timeout: false, - // The copy learned the track before any fetch, from SUBSCRIBE_OK or - // TRACK_STATUS_OK, so a FETCH_OK repeating it is waste. - properties_wanted: false, - }) - .await?; - Ok::<_, Error>(()) - }) - .await; + let res = async { + stream.writer.varint(ietf::Fetch::ID).await?; + stream + .writer + .encode(&ietf::Fetch { + request_id: fetch_id, + subscriber_priority: super::priority::to_wire(request.priority()), + group_order: GroupOrder::Ascending, + fetch_type, + range_filters: false, + fill_timeout: false, + // The copy learned the track before any fetch, from SUBSCRIBE_OK or + // TRACK_STATUS_OK, so a FETCH_OK repeating it is waste. + properties_wanted: false, + }) + .await?; + Ok::<_, Error>(()) + } + .await; let mut retired = false; let res = match res { Err(err) => Some(Err(err)), @@ -5449,6 +5442,103 @@ mod tests { assert_eq!(log.stops(), vec![crate::ietf::error::CANCELLED]); } + #[derive(Clone, Copy, Debug)] + enum Allocated { + Subscribe, + TrackStatus, + GroupFetch, + } + + /// Drafts 14-16 return MAX_REQUEST_ID credit only for requests the peer receives, so a + /// source retiring after its request ID was allocated must still send that request, + /// then cancel it if the draft can. Skipping it would leak the credit for good. + #[moq_net_sim::test] + async fn a_retirement_after_allocation_still_sends_the_request() { + const VERSION: Version = Version::Draft16; + + for kind in [Allocated::Subscribe, Allocated::TrackStatus, Allocated::GroupFetch] { + // Every request write parks until the gate opens, after the retirement. + let gate = kio::Producer::new(false); + let session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); + let log = session.log.clone(); + let (tasks, _task_set) = crate::util::TaskSet::new(); + let mut subscriber = Subscriber::new( + crate::time::Clock::sim(), + session, + crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + Control::new(None, false), + None, + peer::PeerSetup::default(), + crate::Hop::new(1).unwrap(), + None, + VERSION, + tasks, + Default::default(), + ); + let route = kio::Producer::new(()); + subscriber.validity = crate::model::broadcast::RequestValidity::token(Some(route.consume())); + + let producer = crate::broadcast::Info::default().produce(); + let mut dynamic = producer.dynamic(); + let consumer = producer.consume(); + let track = consumer.track("video").unwrap(); + let _subscription = track.subscribe(None); + let request = dynamic.requested_track().await.expect("no track requested"); + + let mut fetch = None; + let serving = match kind { + Allocated::Subscribe => moq_net_sim::spawn(async move { + subscriber.run_subscribe(Path::new("broadcast"), dynamic, request).await; + }), + Allocated::TrackStatus => moq_net_sim::spawn(async move { + assert!( + subscriber + .track_status(&Path::new("broadcast"), "video", request) + .await + .is_none() + ); + }), + Allocated::GroupFetch => { + let track = track::Producer::new( + std::sync::Arc::new(crate::broadcast::Info::default()), + "video", + track::Info::default().with_timescale(None), + ); + let requested = track.dynamic(); + let mut pending = Box::pin(track.consume().fetch_group(0, None)); + assert!(futures::poll!(pending.as_mut()).is_pending()); + let group = requested.requested_group().await.expect("group request"); + fetch = Some((track, pending)); + moq_net_sim::spawn(subscriber.run_group_fetch( + Path::new("broadcast").to_owned(), + "video".into(), + group, + None, + )) + } + }; + + // The ID is allocated and its request parked mid-write when the source retires. + settle().await; + drop(route); + settle().await; + let Ok(mut open) = gate.write() else { + panic!("gate closed") + }; + *open = true; + drop(open); + serving.await.unwrap(); + drop(fetch); + + let expected = match kind { + Allocated::Subscribe => vec![ietf::Subscribe::ID, ietf::Unsubscribe::ID], + Allocated::TrackStatus => vec![ietf::TrackStatus::ID], + Allocated::GroupFetch => vec![ietf::Fetch::ID, ietf::FetchCancel::ID], + }; + assert_eq!(control_message_types(&log, VERSION), expected, "{kind:?}"); + } + } + /// The control messages that actually reached the wire, by type id. /// /// Decoding the framing rather than scanning for a byte: a type id is one varint among From 9698ea3aa2b5b08c2d2c9b0a497f2d269a19238b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 10:49:40 -0700 Subject: [PATCH 3/3] fix(net): compose fetch retirement with group-close cancellation Main now cancels a lite fetch setup when its group closes. Both signals end the setup, and the retirement tests use main's stream-open helper. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/subscriber.test.ts | 40 +++++++++++------------------- js/net/src/lite/subscriber.ts | 2 +- 2 files changed, 16 insertions(+), 26 deletions(-) diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 3f37884050..265ea00eb1 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -1344,14 +1344,14 @@ test.each(["initial", "track", "fetch", "body"] as const)( test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( "a retired epochless handle refuses new metadata on %s", async (version) => { - const { quic, streams } = fakeSession(); + const { quic, streams, opened } = fakeSession(); const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); const announced = subscriber.announced(); - await drainUntil(() => streams.length === 1); + await opened(0); const send = async (encode: (writer: Writer) => Promise) => { const writer = new Writer( new WritableStream({ - write: (chunk) => streams[0].inbound.enqueue(chunk), + write: (chunk) => streams[0].inbound.enqueue(new Uint8Array(chunk)), }), version, ); @@ -1377,18 +1377,8 @@ test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( encodeAnnounceBroadcast(writer, { status: "active", suffix: path, hops: [PUBLISHER_B] }, version), ); await announced.next(); - let settled = false; - const result = track.info().then( - (info) => { - settled = true; - return info; - }, - (err: unknown) => { - settled = true; - return err; - }, - ); - await drainUntil(() => settled || streams.length === 2); + const result = track.info().catch((err: unknown) => err); + await Promise.race([result, opened(1)]); // The unfixed implementation asks B and accepts its metadata. if (streams.length === 2) await answerTrackInfo(streams[1], version); expect(await result).toBeInstanceOf(Error); @@ -1403,17 +1393,17 @@ test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( test.each(["track", "fetch", "open"] as const)( "retirement settles coalesced fetches during %s setup", async (stage) => { - const { quic, streams } = fakeSession(stage === "open" ? [1] : []); + const { quic, streams, opened } = fakeSession(stage === "open" ? [1] : []); const subscriber = new Subscriber(quic, Version.DRAFT_06, HopSchema.parse(1n)); const retired = new AbortController(); const consumed = { id: 1, valid: retired.signal }; const path = Path.from("room"); const first = subscriber.fetchGroup(path, "video", 0, {}, consumed).catch((err: unknown) => err); const second = subscriber.fetchGroup(path, "video", 0, {}, consumed).catch((err: unknown) => err); - await drainUntil(() => streams.length === 1); + await opened(0); if (stage !== "track") { await answerTrackInfo(streams[0], Version.DRAFT_06); - await drainUntil(() => streams.length === 2); + await opened(1); } const cause = new Error("broadcast instance retired"); retired.abort(cause); @@ -1427,15 +1417,15 @@ test.each(["track", "fetch", "open"] as const)( ); test("retirement preserves an accepted fetch and its coalesced reads", async () => { - const { quic, streams } = fakeSession(); + const { quic, streams, opened } = fakeSession(); const subscriber = new Subscriber(quic, Version.DRAFT_06, HopSchema.parse(1n)); const retired = new AbortController(); const consumed = { id: 1, valid: retired.signal }; const path = Path.from("room"); const fetch = subscriber.fetchGroup(path, "video", 0, {}, consumed); - await drainUntil(() => streams.length === 1); + await opened(0); await answerTrackInfo(streams[0], Version.DRAFT_06); - await drainUntil(() => streams.length === 2); + await opened(1); await streams[1].reading; streams[1].inbound.enqueue(new Uint8Array([0, 4, ...new TextEncoder().encode("old!")])); const first = await fetch; @@ -1535,10 +1525,10 @@ test.each([Version.DRAFT_06, Version.DRAFT_07])( test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( "pending metadata cancels on selection change and a live fallback can request again on %s", async (version) => { - const { quic, streams } = fakeSession(); + const { quic, streams, opened } = fakeSession(); const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); const announced = subscriber.announced(); - await drainUntil(() => streams.length === 1); + await opened(0); const peer = new Writer( new WritableStream({ write: (chunk) => streams[0].inbound.enqueue(new Uint8Array(chunk)), @@ -1555,7 +1545,7 @@ test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( .track("pending") .info() .catch((err: unknown) => err); - await drainUntil(() => streams.length === 2); + await opened(1); await streams[1].reading; await encodeAnnounceBroadcast(peer, { status: "active", suffix: job, hops: [PUBLISHER_B] }, version); await announced.next(); @@ -1570,7 +1560,7 @@ test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( ); await announced.next(); const info = held.track("after").info(); - await drainUntil(() => streams.length === 3); + await opened(2); await answerTrackInfo(streams[2], version); expect((await info).priority).toBe(0); held.close(); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 19cfcca08e..6b83d6a669 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -1042,7 +1042,7 @@ export class Subscriber { const { epoch, valid } = consumed; const cancel = new AbortController(); void group.closed.then((err) => cancel.abort(err ?? new StreamError(StreamCode.Cancel))); - const signal = AbortSignal.any([valid, cancel.signal]); + const signal = valid ? AbortSignal.any([valid, cancel.signal]) : cancel.signal; const answered = this.#trackInfo(broadcast, epoch, track, signal); // Only the properties are needed, so the TRACK stream closes however the wait ends. answered.then(