feat(net)!: only sessions accept an origin request

js/net/MIGRATE.md

@@ -52,7 +52,7 @@ Only `publish` announces it: `prepared.broadcast.announce()` throws.
 `Prepared.reject(err)` is replaced by idempotent `close()` and `[Symbol.dispose]()`; either refuses the unpublished answer as `Unroutable` and closes its broadcast. Closing that broadcast also refuses the request. `using` handles normal scope exit and exceptions; after `publish()`, disposal leaves the returned producer live.
 Use `request.reject(err)` to refuse before preparing. A retained, unclosed `Prepared` keeps its request pending; garbage collection provides no prompt cleanup guarantee.
 Construct a `Prepared` only with `request.prepare()`. Its constructor and the `Request.publisher`, `publishPrepared`, and `rejectPrepared` completion hooks are no longer public.
-`Dynamic.update` and `Request.accept` are no longer part of the public API, and `accept` throws on a request from `dynamic`.
+`Dynamic.update` is no longer part of the public API, and `Request` has no `accept` (#6496): only the built-in sessions answer with a broadcast they already hold.
 Close the claim and claim again to change it, and close an answer's producer to end that output.
 
 ### `largest()` reports within a reader's `setGroups` bounds

js/net/bench/broadcasts.ts

@@ -6,6 +6,7 @@ import { Producer as BroadcastProducer } from "../src/broadcast.ts";
 import { BroadcastCache } from "../src/consume.ts";
 import * as Epoch from "../src/epoch.ts";
 import { HopSchema, Route } from "../src/hop.ts";
+import { hooks } from "../src/internal.ts";
 import { AnnounceRequest } from "../src/lite/announce.ts";
 import { Publisher } from "../src/lite/publisher.ts";
 import { Subscriber } from "../src/lite/subscriber.ts";
@@ -252,7 +253,10 @@ for (const pathCount of routeCounts) {
 			Array.from({ length: readerCount }, () => origin.consume().request(Path.from(`room/${index}`))),
 		).flat();
 		const broadcasts = Array.from({ length: pathCount }, () => new BroadcastProducer());
-		for (const broadcast of broadcasts) (await pending.next()).value?.accept(broadcast);
+		for (const broadcast of broadcasts) {
+			const request = (await pending.next()).value;
+			if (request) hooks.acceptRequest(request, broadcast);
+		}
 		await Promise.resolve();
 		const before = requests.map((request) => request.active.peek());
 		const start = performance.now();

js/net/bench/requests.ts

@@ -10,6 +10,7 @@ import {
 	SubscribeNamespaceEntryDone,
 } from "../src/ietf/subscribe_namespace";
 import { ALPN, Version } from "../src/ietf/version";
+import { hooks } from "../src/internal";
 import { createMockTransportPair } from "../src/mock";
 import { Producer } from "../src/origin";
 import * as Path from "../src/path";
@@ -111,7 +112,8 @@ for (const failovers of [0, 1, 16, 128])
 				const held = origin.request(path);
 				const source = new Broadcast();
 				sources.push(source);
-				(await ar.next()).value?.accept(source);
+				const answer = (await ar.next()).value;
+				if (answer) hooks.acceptRequest(answer, source);
 				await settle();
 				entries.push({ primary, ar, br, held });
 			}
@@ -148,7 +150,7 @@ for (const failovers of [0, 1, 16, 128])
 				for (let i = 0; i < entries.length; i++) {
 					const source = new Broadcast();
 					sources.push(source);
-					pending[i].accept(source);
+					hooks.acceptRequest(pending[i], source);
 					await settle();
 					if (!readsFrom(entries[i].held.active.peek(), source)) throw new Error("lost failover");
 				}

js/net/src/connection/forward.ts

@@ -8,7 +8,7 @@ import { Milli, Nano } from "@moq/clock";
 import { type Dispose, race } from "@moq/signals";
 import type * as Announce from "../announced.ts";
 import { type Route, routesEqual } from "../hop.ts";
-import { coveringPrefixes } from "../internal.ts";
+import { coveringPrefixes, hooks } from "../internal.ts";
 import type { Dynamic, Producer as OriginProducer, RequestSlot } from "../origin.ts";
 import * as Path from "../path.ts";
 import { after } from "../util/timer.ts";
@@ -276,7 +276,7 @@ async function drive(handle: Dynamic, conn: Established): Promise<void> {
 	const session = wireOf(conn);
 	try {
 		for await (const request of handle.requested()) {
-			request.accept(session.consume(request.path));
+			hooks.acceptRequest(request, session.consume(request.path));
 		}
 	} catch {
 		handle.close();

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

@@ -7,6 +7,7 @@ import * as Epoch from "../epoch.ts";
 import { error, ProtocolViolation, SessionCode, SessionError, StreamCode, unauthorized } from "../error.ts";
 import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "../group.ts";
 import { type Hop, HopSchema } from "../hop.ts";
+import { hooks } from "../internal.ts";
 import { createMockTransportPair } from "../mock.ts";
 import { Producer as OriginProducer } from "../origin.ts";
 import * as Path from "../path.ts";
@@ -435,7 +436,7 @@ test("a cancelled TRACK_STATUS releases its demand on every draft", async () =>
 		expect(early.written).toEqual([]);
 		// Withdrawn, so a late answer is closed rather than served.
 		const answer = new BroadcastProducer().consume();
-		pending?.accept(answer);
+		hooks.acceptRequest(pending, answer);
 		expect(answer.closed.peek()).toBeNull();
 
 		// The broadcast resolves, but its track is never answered.
@@ -445,7 +446,7 @@ test("a cancelled TRACK_STATUS releases its demand on every draft", async () =>
 		if (!request) throw new Error("no request for room/late");
 		const served = new BroadcastProducer();
 		const trackRequest = wireOf(served).requested();
-		request.accept(served);
+		hooks.acceptRequest(request, served);
 		expect((await trackRequest)?.name).toBe("video");
 		expect(served.demand().used.peek()).toBe(true);
 		late.cancel();
@@ -485,7 +486,7 @@ test("a SUBSCRIBE withdrawn during setup releases its demand on every draft", as
 		expect(early.written).toEqual([]);
 		// Withdrawn, so a late answer is closed rather than served.
 		const answer = new BroadcastProducer().consume();
-		pending?.accept(answer);
+		if (pending) hooks.acceptRequest(pending, answer);
 		expect(answer.closed.peek()).toBeNull();
 
 		// The broadcast resolves, but its track is never answered.
@@ -495,7 +496,7 @@ test("a SUBSCRIBE withdrawn during setup releases its demand on every draft", as
 		if (!request) throw new Error("no request for room/late");
 		const served = new BroadcastProducer();
 		const trackRequest = wireOf(served).requested();
-		request.accept(served);
+		hooks.acceptRequest(request, served);
 		expect((await trackRequest)?.name).toBe("video");
 		expect(served.demand().used.peek()).toBe(true);
 		late.withdraw();

js/net/src/integration.test.ts

@@ -19,6 +19,7 @@ import { Parameters } from "./ietf/parameters.ts";
 import { Publisher as IetfPublisher } from "./ietf/publisher.ts";
 import { RequestOk } from "./ietf/request.ts";
 import { SubscribeTracks } from "./ietf/subscribe_tracks.ts";
+import { hooks } from "./internal.ts";
 import * as Lite from "./lite/index.ts";
 import { Publisher as LitePublisher } from "./lite/publisher.ts";
 import { createMockTransportPair, textFrame } from "./mock.ts";
@@ -2739,7 +2740,7 @@ test("a handle serves a request under live/** over the wire", async () => {
 		for await (const request of handle.requested()) {
 			const produced = new BroadcastProducer();
 			produced.createTrack("chat", { timescale: Timescale.MILLI }).writeFrame(textFrame("hello"));
-			request.accept(produced);
+			hooks.acceptRequest(request, produced);
 		}
 	})();
 

js/net/src/internal.ts

@@ -14,6 +14,7 @@ import type * as Epoch from "./epoch.ts";
 import { toStreamCode } from "./error.ts";
 import type { Frame, Consumer as GroupConsumer, Producer as GroupProducer } from "./group.ts";
 import { isAnonymous, type Route } from "./hop.ts";
+import type { Request as OriginRequest } from "./origin.ts";
 import * as Path from "./path.ts";
 import type { Timestamp } from "./time.ts";
 import type { Demand, Groups, Info, Position, Producer, Request, Subscriber, Subscription } from "./track.ts";
@@ -553,6 +554,11 @@ export const hooks: {
 	stampEpoch: (target: BroadcastConsumer, epoch: Epoch.Valid | undefined) => void;
 	/** Mark a broadcast handle served through a route claiming a shorter prefix: it lists nothing. */
 	stampUnlisted: (target: BroadcastConsumer) => void;
+	/**
+	 * Resolve a session-forwarded request with its peer's routed broadcast. An application
+	 * claim answers through `prepare` and `publish` instead. Assigned by `origin.ts`.
+	 */
+	acceptRequest: (request: OriginRequest, source: BroadcastProducer | BroadcastConsumer) => void;
 } = {
 	endTrack: () => {
 		throw new Error("track.ts not loaded");
@@ -734,6 +740,9 @@ export const hooks: {
 	stampUnlisted: () => {
 		throw new Error("broadcast.ts not loaded");
 	},
+	acceptRequest: () => {
+		throw new Error("origin.ts not loaded");
+	},
 };
 
 /**

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

@@ -2503,7 +2503,7 @@ test("lite draft-06: a grant that shrinks while a TRACK resolves refuses it", as
 		const request = await requests.next();
 		if (request.done) throw new Error("the handler never saw the request");
 		grant.set(grantOf("other"));
-		request.value.accept(produced);
+		hooks.acceptRequest(request.value, produced);
 		await serving;
 
 		expect(unauthorizedResets(resets)).toContain("unauthorized: live/cam");
@@ -2559,7 +2559,7 @@ test("lite draft-06: a grant that shrinks while TRACK_INFO is blocked resets it"
 		const serving = publisher.runTrackInfo(new TrackMessage(Path.from("live/cam"), "video"), stream);
 		const request = await requests.next();
 		if (request.done) throw new Error("the handler never saw the request");
-		request.value.accept(produced);
+		hooks.acceptRequest(request.value, produced);
 		await writing;
 
 		grant.set(grantOf("other"));
@@ -2608,7 +2608,7 @@ test("lite draft-06: a grant that shrinks while the broadcast resolves resets th
 		if (request.done) throw new Error("the handler never saw the request");
 		grant.set(grantOf("other"));
 		await Promise.resolve();
-		request.value.accept(produced);
+		hooks.acceptRequest(request.value, produced);
 		await serving;
 
 		expect(unauthorizedResets(resets)).toContain("unauthorized: live/cam");

js/net/src/lite/tracks.test.ts

@@ -7,6 +7,7 @@ import { accept, connect } from "../connection/index.ts";
 import * as Epoch from "../epoch.ts";
 import { fromTransport, NotFound, SessionCode, StreamCode, StreamError } from "../error.ts";
 import { randomHop } from "../hop.ts";
+import { hooks } from "../internal.ts";
 import { createMockTransportPair } from "../mock.ts";
 import { Producer as OriginProducer } from "../origin.ts";
 import * as Path from "../path.ts";
@@ -412,7 +413,7 @@ test("a path served through a covering prefix route is NOT_FOUND", async () => {
 		for await (const request of dynamic.requested()) {
 			const served = new BroadcastProducer();
 			served.createTrack("video");
-			request.accept(served);
+			hooks.acceptRequest(request, served);
 		}
 	})();
 	const { list } = publisher(origin);
@@ -429,7 +430,7 @@ test("a local read through a covering prefix route is NOT_FOUND, and through its
 			for await (const request of dynamic.requested()) {
 				const served = new BroadcastProducer();
 				served.createTrack("video");
-				request.accept(served);
+				hooks.acceptRequest(request, served);
 			}
 		})();
 	}
@@ -529,7 +530,7 @@ test("an epoch-pinned front served through a covering prefix route is NOT_FOUND"
 		for await (const request of dynamic.requested()) {
 			const served = new BroadcastProducer();
 			served.createTrack("video");
-			request.accept(served);
+			hooks.acceptRequest(request, served);
 		}
 	})();
 	const request = origin.consume().request(Path.from("room/alice"));