feat(net)!: ask a moq-transport peer only about watched prefixes

dart/moq_ffi/lib/src/moq.dart

@@ -20488,7 +20488,7 @@ void _checkApiChecksums() {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqoriginconsumer_request_broadcast() !=
-      65225) {
+      54627) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqorigindynamic_cancel() != 47453) {

doc/concept/standard.md

@@ -25,7 +25,7 @@ maps everything else to "not supported" or a harmless equivalent. The
 [moq-lite page](/concept/moq-lite#what-moq-lite-leaves-out) lists what the
 subset drops. What a peer actually observes against this implementation:
 
-- **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited, and we also ask for every prefix we may discover. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below): on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session.
+- **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. Rust never holds two overlapping requests at a peer that did not declare solicit: a broader one replaces the narrower ones once they are closed. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below): on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session.
 - **`SUBSCRIBE_TRACKS` lists, it never pushes.** On draft 18 and later, it lists every track of every broadcast under its prefix, later ones included. With `FORWARD=0`, each track gets a paused `PUBLISH` carrying its Track Properties, and an error `PUBLISH_DONE` once the track is no longer listed, never `TRACK_ENDED`, since the track may go on elsewhere. A `REQUEST_UPDATE` with `FORWARD=1` resumes one, served from the cache like a `SUBSCRIBE` with the update's filter and answered with `LARGEST_OBJECT`; `FORWARD=0` pauses it again. Rust opens each paused `PUBLISH` only as the peer's stream credit admits it, so a peer that grants none holds none; JavaScript caps them, and past the cap a track gets `PUBLISH_SKIPPED` (`PUBLISH_BLOCKED` on draft 18) instead. A listing that omits `FORWARD` or sets it to 1 asks for every track's objects: from draft 19 each track gets `PUBLISH_SKIPPED`, and the subscriber `SUBSCRIBE`s to what it wants; draft 18 has no such skip, so the request is refused with `NOT_SUPPORTED`. A prefix overlapping another listing gets `PREFIX_OVERLAP`, a `TRACK_PROPERTY_FILTER` gets `NOT_SUPPORTED`, and a `REQUEST_UPDATE` on the listing itself ends it. On draft 18, a `SUBSCRIBE` to a track we `PUBLISH`ed gets `DUPLICATE_SUBSCRIPTION`. Cancelling the listing stops new `PUBLISH`es, while those sent live on. A broadcast is listed only while our publish grant covers it, from scratch once it does. Rust sends it upstream for a listing (below); JavaScript does not.
 - **A Rust TRACKS listing sends `SUBSCRIBE_TRACKS` upstream, only on demand.** Listing a broadcast from a moq-transport peer on draft 18 and later sends `SUBSCRIBE_TRACKS` with `FORWARD=0` while someone lists it, never on connect, announce, or subscribe, so a peer that refuses, resets, or ignores it affects only that listing. Before draft 18 the listing is refused as unsupported. Each listed track's paused `PUBLISH` is its upstream copy, which a reader resumes rather than subscribing again. Only a `PUBLISH_DONE` with `TRACK_ENDED` removes a listed track, and it lists as finished; any other end, including the error our own answerer sends for a track that left unfinished, fails the listing, and readers list again.
 - **History is one group.** In Rust, Object ID gaps must be marked by Prior Object ID Gap properties for a group to be fetchable. A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On drafts 14 through 19, `FETCH_OK` End Object 0 covers the entire End Group, matching the request; Rust also accepts the legacy answer naming the next group with Object 0. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed, and so does a reader whose publisher refuses `TRACK_STATUS` as `NOT_SUPPORTED`.

doc/lib/rs/moq-net.md

@@ -19,7 +19,7 @@ What you use it for, beyond what the concept page already describes:
 - **Requests can pin an epoch.** `consumer.request_broadcast(path, Some(epoch))` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs); `None` takes whichever route wins. The resolved consumer names its epoch in `info().epoch`, pinned or not.
 - **Rank several CDNs.** Dial each with its own subscribe handle, `origin.clone().with_preference(origin::Preference::new(rank))` (higher wins; 0 is neutral and a negative rank demotes), all on one origin. Requests use the preferred link that has a route and fail over to the next; see [multiple CDNs](/concept/moq-lite#multiple-cdns).
 - **Follow restarts.** `consumer.announced_broadcast(path, None)` hands out each publisher instance serving a path from `available()`: the first once announced, then each one that takes over or returns after an end, like JS `request(path, { announced: true })`. A broadcast already resolved stays on the old instance until dropped or its route goes. `consumer.follow(path)` yields the underlying start, update, restart, and end of the route a request resolves.
-- **Discovery follows what you watch.** On moq-lite, a client session asks the relay for announcements only under the prefixes its subscribe origin's readers watch, with `announced()`, `routed_broadcast`, or `announced_broadcast`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A broadcast resolved through a discovered route keeps its path watched until dropped. `request_broadcast` watches nothing, so a path nobody watches is unroutable. A server session, or a client fed into an origin handle marked `with_link(origin::Link::Peer)` or `Link::Upstream`, asks for everything it may, so a relay discovers everything. An HTTP gateway (the moq-hls `Server`, the moq-rtc WHEP router) watches its whole scope while it lives, so the paths its callers name never open a request each. moq-transport sessions, and moq-lite-03/04 sessions (whose announcements mark no initial set), still ask for everything.
+- **Discovery follows what you watch.** A client session asks the relay for announcements only under the prefixes its subscribe origin's readers watch, with `announced()`, `routed_broadcast`, or `announced_broadcast`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A broadcast resolved through a discovered route keeps its path watched until dropped. `request_broadcast` watches nothing, so a path nobody watches is unroutable. A server session, or a client fed into an origin handle marked `with_link(origin::Link::Peer)` or `Link::Upstream`, asks for everything it may, so a relay discovers everything. An HTTP gateway (the moq-hls `Server`, the moq-rtc WHEP router) watches its whole scope while it lives, so the paths its callers name never open a request each. moq-lite-03/04 sessions, whose announcements mark no initial set, still ask for everything.
 - **One track per name.** Concurrent subscriptions share one request and producer. The application owns sequence numbers: a producer that replaces an ended one gets `Error::Duplicate` from an append that would reuse a sequence the name already wrote, so it starts with `create_group` at an explicit sequence. Only a new broadcast starts over. Readers of an aborted track resume onto the replacement only when the broadcast announces an epoch; otherwise their subscription ends with the error.
 - **End a broadcast with `finish()`.** When the input ends, finish the tracks, then call `broadcast.finish()`. The returned handle keeps the broadcast announced and serving its finished tracks for `linger()`, their longest max-age, so a subscriber still catching up can drain them. Await `wait()` on a runtime, or drop the handle once your own clock passes `linger()`. Dropping it early is the interrupt, and `close()` on another handle ends it at once.
 - **Publish only while someone is watching.** `demand()` on a track, group, or broadcast says whether a subscriber is attached, which is how capture and transcode skip work nobody asked for. Holding a broadcast or track consumer is not demand: a track consumer requests nothing until it queries, subscribes, or fetches, and those count until dropped. A shared fetch stays up until its last reader leaves.

js/net/src/ietf/adapter.ts

@@ -111,7 +111,11 @@ interface StreamEntry {
 	controller: ReadableStreamDefaultController<Uint8Array>;
 }
 
-type NamespaceDirection = "incoming" | "outgoing";
+/**
+ * Which request a v14/v15 namespace-keyed message names: an announcement, by the side that
+ * made it, or the peer's namespace subscription (draft 14's UNSUBSCRIBE_NAMESPACE).
+ */
+type NamespaceDirection = "incoming" | "outgoing" | "subscribed";
 
 /** Runs per chunk before it splits, so a split or removal shifts at most one chunk. */
 const RUN_CHUNK = 256;
@@ -252,6 +256,8 @@ export class ControlStreamAdapter implements Session {
 	// may legitimately hold the same namespace in both maps at once.
 	#incomingNamespaces = new Map<string, bigint>();
 	#outgoingNamespaces = new Map<string, bigint>();
+	// The peer's draft-14 SUBSCRIBE_NAMESPACE prefixes, which its UNSUBSCRIBE_NAMESPACE names.
+	#subscribedNamespaces = new Map<string, bigint>();
 
 	// requestId → namespace reverse lookup, one entry per request including a refused
 	// duplicate, so #forget can clean up whichever request goes away.
@@ -842,12 +848,13 @@ export class ControlStreamAdapter implements Session {
 	}
 
 	/**
-	 * Decode a PublishNamespace body and register its namespace for the v14/v15
-	 * namespace-keyed withdrawals. Returns the request ID it carried.
+	 * Decode a PublishNamespace body, or a draft-14 SubscribeNamespace one, and register its
+	 * namespace for the v14/v15 namespace-keyed messages that end it. Returns the request
+	 * ID it carried.
 	 *
-	 * The one decode site for both directions: outgoing announcements go through
-	 * #tryParseOutgoing and incoming ones through #classify. The direction selects the
-	 * map that the peer's DONE or CANCEL will later consult.
+	 * The one decode site for every direction: outgoing announcements go through
+	 * #tryParseOutgoing and incoming requests through #classify. The direction selects the
+	 * map that the peer's DONE, CANCEL, or UNSUBSCRIBE_NAMESPACE will later consult.
 	 */
 	async #registerNamespace(body: Uint8Array, direction: NamespaceDirection): Promise<bigint> {
 		const r = new Reader(undefined, body, this.version);
@@ -866,7 +873,14 @@ export class ControlStreamAdapter implements Session {
 	}
 
 	#namespaceRequests(direction: NamespaceDirection): Map<string, bigint> {
-		return direction === "incoming" ? this.#incomingNamespaces : this.#outgoingNamespaces;
+		switch (direction) {
+			case "incoming":
+				return this.#incomingNamespaces;
+			case "outgoing":
+				return this.#outgoingNamespaces;
+			case "subscribed":
+				return this.#subscribedNamespaces;
+		}
 	}
 
 	/** Create a WritableStream that buffers and writes complete messages to the control stream under mutex. */
@@ -960,7 +974,7 @@ export class ControlStreamAdapter implements Session {
 			const namespace = await Namespace.decode(r);
 			const namespaces = this.#namespaceRequests(direction);
 			const requestId = namespaces.get(namespace);
-			if (requestId === undefined) console.warn(`adapter: no announcement for namespace: ${namespace}`);
+			if (requestId === undefined) console.warn(`adapter: no request for namespace: ${namespace}`);
 			return requestId;
 		};
 
@@ -994,7 +1008,11 @@ export class ControlStreamAdapter implements Session {
 				if (this.version !== Version.DRAFT_14 && this.version !== Version.DRAFT_15) {
 					throw new Error("unexpected SubscribeNamespace on control stream");
 				}
-				const requestId = await readRequestId();
+				// Draft 14's UNSUBSCRIBE_NAMESPACE names the prefix rather than the request.
+				const requestId =
+					this.version === Version.DRAFT_14
+						? await this.#registerNamespace(body, "subscribed")
+						: await readRequestId();
 				return { route: Route.NewRequest, requestId };
 			}
 
@@ -1119,11 +1137,14 @@ export class ControlStreamAdapter implements Session {
 				return { route: Route.CloseStream, requestId };
 			}
 			case 0x14: {
-				// UnsubscribeNamespace (v14/v15 only)
-				if (this.version !== Version.DRAFT_14 && this.version !== Version.DRAFT_15) {
-					throw new Error("unexpected UnsubscribeNamespace");
+				// UnsubscribeNamespace (v14/v15 only): v14 names the prefix, v15 the request.
+				if (this.version === Version.DRAFT_15) {
+					const requestId = await readRequestId();
+					return { route: Route.CloseStream, requestId };
 				}
-				const requestId = await readRequestId();
+				if (this.version !== Version.DRAFT_14) throw new Error("unexpected UnsubscribeNamespace");
+				const requestId = await readNamespaceRequestId("subscribed");
+				if (requestId === undefined) return { route: Route.Ignore, requestId: 0n };
 				return { route: Route.CloseStream, requestId };
 			}
 
@@ -1176,6 +1197,7 @@ export class ControlStreamAdapter implements Session {
 		// Clear namespace mappings
 		this.#incomingNamespaces.clear();
 		this.#outgoingNamespaces.clear();
+		this.#subscribedNamespaces.clear();
 		this.#namespacesByRequestId.clear();
 		this.#subscribeNamespaces.clear();
 

js/net/src/ietf/ietf.test.ts

@@ -455,6 +455,20 @@ test("PublishNamespaceDone: round trip", async () => {
 	expect(decoded.trackNamespace).toBe("old/stream" as Path.Valid);
 });
 
+test("UnsubscribeNamespace: names the prefix on draft 14 and the request on draft 15", async () => {
+	const msg = new SubscribeNamespace.UnsubscribeNamespace({ prefix: Path.from("cam/front"), requestId: 6n });
+
+	const v14 = await encodeVersioned(msg, Version.DRAFT_14);
+	const named = await decodeVersioned(v14, SubscribeNamespace.UnsubscribeNamespace.decode, Version.DRAFT_14);
+	expect(named.prefix).toBe("cam/front" as Path.Valid);
+
+	const v15 = await encodeVersioned(msg, Version.DRAFT_15);
+	const numbered = await decodeVersioned(v15, SubscribeNamespace.UnsubscribeNamespace.decode, Version.DRAFT_15);
+	expect(numbered.requestId).toBe(6n);
+
+	await expect((() => encodeVersioned(msg, Version.DRAFT_16))()).rejects.toThrow(/draft-14\/15 only/);
+});
+
 test("PublishNamespaceCancel: round trip", async () => {
 	const msg = new Announce.PublishNamespaceCancel({
 		trackNamespace: Path.from("canceled"),

js/net/src/ietf/publisher.test.ts

@@ -2652,7 +2652,7 @@ test("drafts 14 to 16 refuse a namespace update without ending the subscription"
 			ask.writer.close();
 		} else {
 			await peer.writer.u53(UnsubscribeNamespace.id);
-			await new UnsubscribeNamespace({ requestId: 0n }).encode(peer.writer, version);
+			await new UnsubscribeNamespace({ prefix: asked.namespace, requestId: 0n }).encode(peer.writer, version);
 		}
 		await running;
 		origin.close();

js/net/src/ietf/subscribe_namespace.ts

@@ -253,30 +253,42 @@ export class SubscribeNamespaceError {
 	}
 }
 
+/** UNSUBSCRIBE_NAMESPACE (0x14), drafts 14 and 15 only: draft 14 names the prefix, draft 15 the request. */
 export class UnsubscribeNamespace {
 	static id = 0x14;
 
+	/** Draft 14: the Track Namespace Prefix of the SUBSCRIBE_NAMESPACE it cancels. */
+	prefix: Path.Valid;
+	/** Draft 15: the Request ID of the SUBSCRIBE_NAMESPACE it cancels. */
 	requestId: bigint;
 
-	constructor({ requestId }: { requestId: bigint }) {
+	constructor({ prefix = "" as Path.Valid, requestId = 0n }: { prefix?: Path.Valid; requestId?: bigint }) {
+		this.prefix = prefix;
 		this.requestId = requestId;
 	}
 
-	async #encode(w: Writer): Promise<void> {
-		await w.u62(this.requestId);
+	async #encode(w: Writer, version: IetfVersion): Promise<void> {
+		if (version === Version.DRAFT_14) {
+			await Namespace.encode(w, this.prefix);
+		} else if (version === Version.DRAFT_15) {
+			await w.u62(this.requestId);
+		} else {
+			throw new Error(`UNSUBSCRIBE_NAMESPACE is draft-14/15 only, not ${version}`);
+		}
 	}
 
-	async encode(w: Writer, _version: IetfVersion): Promise<void> {
-		return Message.encode(w, this.#encode.bind(this));
+	async encode(w: Writer, version: IetfVersion): Promise<void> {
+		return Message.encode(w, (wr) => this.#encode(wr, version));
 	}
 
-	static async decode(r: Reader, _version: IetfVersion): Promise<UnsubscribeNamespace> {
-		return Message.decode(r, UnsubscribeNamespace.#decode);
+	static async decode(r: Reader, version: IetfVersion): Promise<UnsubscribeNamespace> {
+		return Message.decode(r, (rd) => UnsubscribeNamespace.#decode(rd, version));
 	}
 
-	static async #decode(r: Reader): Promise<UnsubscribeNamespace> {
-		const requestId = await r.u62();
-		return new UnsubscribeNamespace({ requestId });
+	static async #decode(r: Reader, version: IetfVersion): Promise<UnsubscribeNamespace> {
+		if (version === Version.DRAFT_14) return new UnsubscribeNamespace({ prefix: await Namespace.decode(r) });
+		if (version === Version.DRAFT_15) return new UnsubscribeNamespace({ requestId: await r.u62() });
+		throw new Error(`UNSUBSCRIBE_NAMESPACE is draft-14/15 only, not ${version}`);
 	}
 }
 

js/net/src/ietf/subscriber.ts

@@ -563,7 +563,7 @@ export class Subscriber {
 					// For v14/v15: send UnsubscribeNamespace before closing
 					if (version === Version.DRAFT_14 || version === Version.DRAFT_15) {
 						try {
-							const unsub = new UnsubscribeNamespace({ requestId });
+							const unsub = new UnsubscribeNamespace({ prefix, requestId });
 							await Message.write(request.writer, UnsubscribeNamespace.id, unsub, version);
 						} catch {
 							// Stream might already be closed