fix(js): never hold overlapping SUBSCRIBE_NAMESPACEs at a foreign peer

doc/concept/standard.md

@@ -25,7 +25,7 @@ maps everything else to "not supported" or a harmless equivalent. The
 [moq-lite page](/concept/moq-lite#what-moq-lite-leaves-out) lists what the
 subset drops. What a peer actually observes against this implementation:
 
-- **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. Rust never holds two overlapping requests at a peer that did not declare solicit: a broader one replaces the narrower ones once they are closed. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below): on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session.
+- **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. We never hold two overlapping requests at a peer that did not declare solicit: a broader one replaces the narrower ones once they are closed. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below): on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session.
 - **`SUBSCRIBE_TRACKS` lists, it never pushes.** On draft 18 and later, it lists every track of every broadcast under its prefix, later ones included. With `FORWARD=0`, each track gets a paused `PUBLISH` carrying its Track Properties, and a `PUBLISH_DONE` once the track is no longer listed: `TRACK_ENDED` if it finished, even just before its broadcast ended, and an error otherwise, since the track may go on elsewhere. A `REQUEST_UPDATE` with `FORWARD=1` resumes one, served from the cache like a `SUBSCRIBE` with the update's filter and answered with `LARGEST_OBJECT`; `FORWARD=0` pauses it again. Each paused `PUBLISH` opens only as the peer's stream credit admits it, so a peer that grants none holds none. A listing that omits `FORWARD` or sets it to 1 asks for every track's objects: on draft 22 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 18 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 22 `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On drafts 14 through 18, `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 22, `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`.

js/net/bench/forward.ts

@@ -36,6 +36,7 @@ class FakeSession {
 		registerWire(this, {
 			consume: () => new BroadcastProducer().consume(),
 			goaway: new Once(),
+			exclusive: false,
 			auth: new AuthSession({ peerGrant: { publish: new Path.Patterns(), subscribe: new Path.Patterns() } }),
 		});
 		this.closed = new Promise((resolve) => {

js/net/src/connection/forward.test.ts

@@ -49,10 +49,12 @@ class FakeSession {
 	closed: Promise<Error | null>;
 	#die!: (reason: Error | null) => void;
 
-	constructor() {
+	/** `exclusive` stands in for a peer that refuses overlapping requests. */
+	constructor({ exclusive = false } = {}) {
 		registerWire(this, {
 			consume: (path) => this.consume(path),
 			goaway: new Once(),
+			exclusive,
 			auth: new AuthSession({ peerGrant: { publish: new Path.Patterns(), subscribe: new Path.Patterns() } }),
 		});
 		this.closed = new Promise((resolve) => {
@@ -341,6 +343,48 @@ test("a narrower request stays open until the broader one covering it goes", asy
 	origin.close();
 });
 
+test("an exclusive session closes every request beneath a broader one before it opens", async () => {
+	const origin = new OriginProducer();
+	const session = new FakeSession({ exclusive: true });
+	forwardAnnounced(session.session, origin);
+
+	const narrow = origin.announced(Path.Pattern.subtree(Path.from("a/b")));
+	const lingering = origin.announced(Path.Pattern.subtree(Path.from("a/c")));
+	await settle();
+	start(session.stream("a/b/**"), "a/b/x");
+	lingering.close();
+	await settle();
+	expect(session.open()).toEqual(["a/b/**", "a/c/**"]);
+
+	// Watched or lingering, each narrower request closes for the broader one, on purpose: the
+	// session stays discovering, and the route returns once the broader request reports it.
+	const broad = origin.announced(Path.Pattern.subtree(Path.from("a")));
+	const events = record(broad);
+	await settle();
+	expect(session.open()).toEqual(["a/**"]);
+	expect(wireOf(origin).routes(Path.from("a/b/x"))).toBe(false);
+	start(session.stream("a/**"), "a/b/x");
+	await settle();
+	expect(events).toEqual(["start a/b/x", "end a/b/x", "start a/b/x"]);
+	expect(origin.discovery.peek()).toBe(true);
+
+	// The narrower watch rides the broader request from here on.
+	broad.close();
+	await settle();
+	jest.advanceTimersByTime(LINGER);
+	await settle();
+	expect(session.open()).toEqual(["a/**"]);
+	narrow.close();
+	await settle();
+	jest.advanceTimersByTime(LINGER);
+	await settle();
+	expect(session.open()).toEqual([]);
+	expect(session.requests.map(({ scope }) => scope)).toEqual(["a/b/**", "a/c/**", "a/**"]);
+
+	session.die();
+	origin.close();
+});
+
 const first = Epoch.mint();
 for (const [name, epochs] of [
 	["no epoch", [undefined, undefined]],

js/net/src/connection/forward.ts

@@ -58,6 +58,11 @@ interface Reported {
  * driven by its first reporter; if it changes hands anyway (a request failing), it restarts
  * unless one epoch proves both requests saw one instance.
  *
+ * A peer that refuses overlapping requests gets exclusive discovery instead: a broader request
+ * closes the narrower ones beneath it, and the session keeps it off the wire until they have
+ * left. The routes a closed request carried go with it, and come back once the broader request
+ * reports them.
+ *
  * Each active announcement inserts a route covering the announced prefix, served lazily
  * through this connection, so nothing touches the wire until somebody requests a path
  * under it. A retraction removes the entry, and so does the session dying, which is what
@@ -72,6 +77,7 @@ interface Reported {
  */
 export function forwardAnnounced(conn: Established, origin: OriginProducer): void {
 	const originWire = wireOf(origin);
+	const exclusive = wireOf(conn).exclusive;
 	// Reassigned on a downgrade, so the origin stops counting this session as discovering.
 	// Called through a closure so the session's death always detaches the current one.
 	let detach = originWire.attach(true);
@@ -155,6 +161,17 @@ export function forwardAnnounced(conn: Established, origin: OriginProducer): voi
 	};
 
 	const open = (prefix: Path.Valid) => {
+		if (exclusive) {
+			for (const [other, narrower] of interests) {
+				if (other === prefix || !Path.hasPrefix(prefix, other)) continue;
+				interests.delete(other);
+				narrower.closing = true;
+				narrower.linger?.();
+				narrower.linger = undefined;
+				narrower.announced.close();
+			}
+		}
+
 		// Hidden routes are mirrored too; each local reader opts in on its own.
 		const interest: Interest = {
 			announced: conn.announced(Path.Pattern.subtree(prefix), { hidden: true }),

js/net/src/ietf/connection.ts

@@ -276,6 +276,9 @@ export class Connection implements Established {
 		registerWire(this, {
 			consume: (path) => this.#subscriber.consume(path),
 			goaway: this.#goaway,
+			// The drafts refuse overlapping SUBSCRIBE_NAMESPACEs (PREFIX_OVERLAP). A peer that
+			// declared MoQ Solicit is ours and holds both; any other may enforce it.
+			exclusive: solicit === undefined,
 			auth: this.#auth,
 		});
 

js/net/src/ietf/request_stream.ts

@@ -1,4 +1,4 @@
-import { error, ProtocolViolation, StreamError } from "../error.ts";
+import { error, ProtocolViolation, StreamCode, StreamError } from "../error.ts";
 import type * as Path from "../path.ts";
 import { Reader, type Stream, type Writer } from "../stream.ts";
 import type { Session } from "./adapter.ts";
@@ -234,6 +234,16 @@ export class RequestStream {
 		this.close();
 	}
 
+	/**
+	 * Cancel the request but keep reading, so {@link next} reports when the peer ends its
+	 * half: FIN where that cancels, a CANCELLED reset otherwise.
+	 */
+	cancel(): void {
+		if (this.#closed) return;
+		if (finCancels(this.#session.version)) this.writer.close();
+		else this.writer.reset(new StreamError(StreamCode.Cancel, { message: "cancel" }));
+	}
+
 	/** End the request cleanly: FIN, and stop reading. Only the first close counts. */
 	close(): void {
 		if (this.#closed) return;

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

@@ -28,7 +28,13 @@ import {
 } from "./publish_namespace.ts";
 import { MaxRequestId, RequestError, RequestOk } from "./request.ts";
 import { Subscribe, SubscribeOk, SubscribeUpdate, Unsubscribe } from "./subscribe.ts";
-import { SubscribeNamespace, SubscribeNamespaceEntry, SubscribeNamespaceEntryDone } from "./subscribe_namespace.ts";
+import {
+	SubscribeNamespace,
+	SubscribeNamespaceEntry,
+	SubscribeNamespaceEntryDone,
+	SubscribeNamespaceLegacy,
+	UnsubscribeNamespace,
+} from "./subscribe_namespace.ts";
 import { Subscriber } from "./subscriber.ts";
 import { ALPN, type IetfVersion, Version } from "./version.ts";
 
@@ -43,7 +49,7 @@ const STREAM_WAIT = 500;
  * Reads the queue directly instead of racing {@link Stream.accept}, whose pending read
  * would keep the reader locked after the race resolves and could swallow a later stream.
  */
-async function nextStream(transport: WebTransport): Promise<Stream | undefined> {
+async function nextStream(transport: WebTransport, version: IetfVersion = VERSION): Promise<Stream | undefined> {
 	const reader =
 		transport.incomingBidirectionalStreams.getReader() as ReadableStreamDefaultReader<WebTransportBidirectionalStream>;
 
@@ -57,7 +63,7 @@ async function nextStream(transport: WebTransport): Promise<Stream | undefined>
 		]);
 
 		if (!next || next.done) return undefined;
-		return new Stream({ readable: next.value.readable, writable: next.value.writable, version: VERSION });
+		return new Stream({ readable: next.value.readable, writable: next.value.writable, version });
 	} finally {
 		clearTimeout(timer);
 		reader.releaseLock();
@@ -228,23 +234,28 @@ test("an advertisement with no path is anonymous", async () => {
  * Answer the SUBSCRIBE_NAMESPACE the subscriber just opened, then hand back its stream so
  * the test can feed inline NAMESPACE entries down it.
  */
-async function acceptSubscribeNamespace(transport: WebTransport): Promise<Stream> {
-	const stream = await nextStream(transport);
+async function acceptSubscribeNamespace(transport: WebTransport, version: IetfVersion = VERSION): Promise<Stream> {
+	const stream = await nextStream(transport, version);
 	if (!stream) throw new Error("no SUBSCRIBE_NAMESPACE was sent");
 
 	// Drain the request, then answer it.
 	await stream.reader.u53();
-	await SubscribeNamespace.decode(stream.reader, VERSION);
+	await SubscribeNamespace.decode(stream.reader, version);
 	await stream.writer.u53(RequestOk.id);
-	await new RequestOk({ requestId: undefined }).encode(stream.writer, VERSION);
+	await new RequestOk({ requestId: undefined }).encode(stream.writer, version);
 
 	return stream;
 }
 
 /** Advertise `path` inline on a SUBSCRIBE_NAMESPACE stream. */
-async function inlineNamespace(stream: Stream, path: Path.Valid, cluster?: Cluster.Advert): Promise<void> {
+async function inlineNamespace(
+	stream: Stream,
+	path: Path.Valid,
+	cluster?: Cluster.Advert,
+	version: IetfVersion = VERSION,
+): Promise<void> {
 	await stream.writer.u53(SubscribeNamespaceEntry.id);
-	await new SubscribeNamespaceEntry({ suffix: path, params: cluster && { cluster } }).encode(stream.writer, VERSION);
+	await new SubscribeNamespaceEntry({ suffix: path, params: cluster && { cluster } }).encode(stream.writer, version);
 }
 
 /**
@@ -419,11 +430,14 @@ test("an entry buffered past a consumer close does not pin the path", async () =
 	await inlineNamespace(subscription, Path.from("buffered"));
 
 	announced.close();
+	// The peer ends its half on the cancel, so the request leaves the wire.
+	subscription.writer.close();
 
 	// Whatever the loop managed to attach, the stream gave back on its way out.
 	await new Promise((resolve) => setTimeout(resolve, 50));
 
 	const fresh = subscriber.announced();
+	await acceptSubscribeNamespace(pair.client);
 	const seeded = await Promise.race([
 		fresh.next(),
 		new Promise<"nothing">((resolve) => setTimeout(() => resolve("nothing"), 250)),
@@ -2221,6 +2235,84 @@ test("an announce consumer closed while waiting on request-ID credit takes no ID
 	void running;
 });
 
+/**
+ * A request on another stream can overtake a cancel, so a peer that refuses overlapping
+ * requests (PREFIX_OVERLAP) would see the broader one beside the narrower one it replaces.
+ * It goes out only once the peer has ended the narrower one's stream.
+ */
+test("a request overlapping one on its way out waits for the peer to end it", async () => {
+	const version = Version.DRAFT_22;
+	const pair = createMockTransportPair(ALPN.DRAFT_22);
+	const session = new NativeSession(pair.server, version, true);
+	const subscriber = new Subscriber({ session });
+
+	const narrow = subscriber.announced(Path.Pattern.subtree(Path.from("a/b")));
+	const stream = await acceptSubscribeNamespace(pair.client, version);
+	await inlineNamespace(stream, Path.from("x"), undefined, version);
+	expect(await narrow.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "start" });
+
+	// A request still open is no blocker, and hears the narrower one's route go at once.
+	const watcher = subscriber.announced(Path.Pattern.subtree(Path.from("a/b")));
+	const watching = await acceptSubscribeNamespace(pair.client, version);
+	expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "start" });
+	narrow.close();
+	const broad = subscriber.announced(Path.Pattern.subtree(Path.from("a")));
+	expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/x"), kind: "end" });
+
+	// Draft 22 cancels with a reset, and the broader request waits on the peer's half.
+	expect(stream.reader.done()).rejects.toMatchObject({ code: StreamCode.Cancel });
+	const incoming = pair.client.incomingBidirectionalStreams.getReader();
+	let sent = false;
+	const opened = incoming.read().then((next) => {
+		sent = true;
+		return next;
+	});
+
+	// The closed reader is still registered meanwhile, and hears nothing more.
+	await inlineNamespace(watching, Path.from("y"), undefined, version);
+	expect(await watcher.next()).toMatchObject({ prefix: Path.from("a/b/y"), kind: "start" });
+	expect(sent).toBe(false);
+
+	stream.writer.close();
+	const next = await opened;
+	incoming.releaseLock();
+	if (next.done) throw new Error("the broader SUBSCRIBE_NAMESPACE never went out");
+	const asked = new Stream({ readable: next.value.readable, writable: next.value.writable, version });
+	await asked.reader.u53();
+	expect((await SubscribeNamespace.decode(asked.reader, version)).namespace).toBe(Path.from("a"));
+
+	watcher.close();
+	broad.close();
+	session.close();
+});
+
+/** Draft 14/15 requests share the control stream, so the next one waits only for the UNSUBSCRIBE_NAMESPACE. */
+test("a draft-15 request overlapping one on its way out follows its UNSUBSCRIBE_NAMESPACE", async () => {
+	const version = Version.DRAFT_15;
+	const pair = createMockTransportPair(ALPN.DRAFT_15);
+	const control = await Stream.open(pair.server, { version });
+	const session = new ControlStreamAdapter(pair.server, control, version, 100n, true);
+	void session.run().catch(() => undefined);
+	const peer = await Stream.accept(pair.client, version);
+	if (!peer) throw new Error("no control stream");
+	const subscriber = new Subscriber({ session });
+
+	const narrow = subscriber.announced(Path.Pattern.subtree(Path.from("a/b")));
+	expect(await peer.reader.u53()).toBe(SubscribeNamespaceLegacy.id);
+	expect((await SubscribeNamespaceLegacy.decode(peer.reader, version)).namespace).toBe(Path.from("a/b"));
+
+	narrow.close();
+	const broad = subscriber.announced(Path.Pattern.subtree(Path.from("a")));
+
+	expect(await peer.reader.u53()).toBe(UnsubscribeNamespace.id);
+	await UnsubscribeNamespace.decode(peer.reader, version);
+	expect(await peer.reader.u53()).toBe(SubscribeNamespaceLegacy.id);
+	expect((await SubscribeNamespaceLegacy.decode(peer.reader, version)).namespace).toBe(Path.from("a"));
+
+	broad.close();
+	session.close();
+});
+
 /** Answer the next SUBSCRIBE_NAMESPACE with REQUEST_OK, carrying `activeCount` when given. */
 async function answerCounted(transport: WebTransport, activeCount: number | undefined): Promise<Stream> {
 	const stream = await nextStream(transport);