refactor(net)!: keep session-only model items private

js/net/MIGRATE.md

@@ -38,15 +38,9 @@ Path normalization, relative references, and subscription paths retain their sem
 Drop the `await`. Every caller reads the same value: the old shared cursor let one caller take an update and leave another waiting forever.
 To wait for each newer one, keep a reader and follow its stream: `const reader = producer.subscribe();`, then `await reader.largest()`.
 
-### Stage a requested track before subscribers see it (#6111)
-
-`Track.Request.stage(info)` returns a `Track.Staged` whose `producer` writes while subscribers waiting on the request stay pending; `commit()` publishes it with everything written so far and returns the `Producer`.
-Disposing a `Staged` uncommitted (`using`, or `[Symbol.dispose]()`) rejects the request. `accept(info)` is unchanged and equals `stage(info).commit()`.
-
 ### A frameless group is not a successor (#6111)
 
 A group followed by a frameless one is judged stale against the next group that has a frame, where the frameless group used to count.
-`Track.Producer.dropGroups(start, end, error)` marks groups `[start, end)` the peer said it skipped; a reader holding one of them rejects with `error`.
 
 ### Priorities are signed (#6142)
 

js/net/api/index.api.md

@@ -632,7 +632,6 @@ class Producer {
     close(abort?: Error): void;
     get closed(): GetPromise<Error | null>;
     demand(): Demand;
-    dropGroups(start: number, end: number, error: Error): void;
     finishAt(end: Position_2): void;
     info(): Promise<Info>;
     insertDatagram(sequence: number, timestamp: Timestamp | undefined, payload: Uint8Array): void;
@@ -737,7 +736,6 @@ class Request_2 {
     readonly name: string;
     get priority(): number;
     reject(err?: Error): void;
-    stage(info?: Partial<Info>): Staged;
     get subscription(): Readonly<Subscription>;
 }
 
@@ -825,12 +823,6 @@ export { Signals }
 
 function size(v: number): number;
 
-class Staged {
-    [Symbol.dispose](): void;
-    commit(): Producer;
-    readonly producer: Producer;
-}
-
 interface Stats {
     bytesLost?: number;
     bytesReceived?: number;
@@ -972,7 +964,6 @@ declare namespace Track {
         Subscription,
         Order,
         Request_2 as Request,
-        Staged,
         FetchGroupOptions,
         Consumer_3 as Consumer,
         Demand,
@@ -993,7 +984,6 @@ declare namespace track {
         Subscription,
         Order,
         Request_2 as Request,
-        Staged,
         FetchGroupOptions,
         Consumer_3 as Consumer,
         Demand,

js/net/api/zod.api.md

@@ -141,7 +141,6 @@ class Producer {
     close(abort?: Error): void;
     get closed(): GetPromise<Error | null>;
     demand(): Demand;
-    dropGroups(start: number, end: number, error: Error): void;
     finishAt(end: Position_2): void;
     info(): Promise<Info>;
     insertDatagram(sequence: number, timestamp: Timestamp | undefined, payload: Uint8Array): void;
@@ -176,16 +175,9 @@ class Request_2 {
     readonly name: string;
     get priority(): number;
     reject(err?: Error): void;
-    stage(info?: Partial<Info>): Staged;
     get subscription(): Readonly<Subscription>;
 }
 
-class Staged {
-    [Symbol.dispose](): void;
-    commit(): Producer;
-    readonly producer: Producer;
-}
-
 class Subscriber {
     close(abort?: Error): void;
     get closed(): GetPromise<Error | null>;
@@ -245,7 +237,6 @@ declare namespace track {
         Subscription,
         Order,
         Request_2 as Request,
-        Staged,
         FetchGroupOptions,
         Consumer_2 as Consumer,
         Demand,

js/net/src/broadcast.test.ts

@@ -3,6 +3,7 @@ import { Milli } from "@moq/clock";
 import { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts";
 import { GroupTooLarge, NotFound } from "./error.ts";
 import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "./group.ts";
+import { hooks } from "./internal.ts";
 import { textFrame } from "./mock.ts";
 import { Timescale, Timestamp } from "./time.ts";
 import type { Request as TrackRequest } from "./track.ts";
@@ -62,7 +63,7 @@ test("a staged track resolves its subscribers at commit, with what was written",
 	const request = await pendingRequest(consumer);
 	if (!request) throw new Error("expected request");
 
-	const staged = request.stage({ timescale: Timescale.MILLI });
+	const staged = hooks.stageRequest(request, { timescale: Timescale.MILLI });
 	staged.producer.writeFrame(textFrame("hello"));
 	const pending = Symbol("pending");
 	expect(await Promise.race([track.info(), Promise.resolve(pending)])).toBe(pending);
@@ -72,31 +73,6 @@ test("a staged track resolves its subscribers at commit, with what was written",
 	expect(await reader.readString()).toBe("hello");
 });
 
-test("disposing an uncommitted staged track rejects the request", async () => {
-	const consumer = new TestConsumer();
-	const info = consumer.track("video").info();
-	const request = await pendingRequest(consumer);
-	if (!request) throw new Error("expected request");
-
-	{
-		using staged = request.stage({ timescale: Timescale.MILLI });
-		staged.producer.writeFrame(textFrame("hello"));
-	}
-	await expect(info).rejects.toThrow();
-});
-
-test("a disposed staged track refuses a later commit", async () => {
-	const consumer = new TestConsumer();
-	const info = consumer.track("video").info();
-	const request = await pendingRequest(consumer);
-	if (!request) throw new Error("expected request");
-
-	const staged = request.stage({ timescale: Timescale.MILLI });
-	staged[Symbol.dispose]();
-	expect(() => staged.commit()).toThrow("staged track already committed or dropped");
-	await expect(info).rejects.toThrow("staged track dropped");
-});
-
 // A broadcast keeps nothing of a name once its track is gone: a replacement numbers and times its
 // groups from scratch, and the application owns keeping them past its predecessor's.
 test("a replacement producer starts over", async () => {

js/net/src/ietf/subscriber.ts

@@ -1282,7 +1282,7 @@ export class Subscriber {
 		}
 		// Staged until the answer is written in below, so no reader sees the track before its
 		// Largest and start.
-		const staged = request.stage({
+		const staged = hooks.stageRequest(request, {
 			// No TIMESCALE (always so on drafts 14-16, which can't carry it) means no timeline,
 			// and the track must not claim one when served onward.
 			timescale: ok.properties.timescale,

js/net/src/internal.ts

@@ -330,6 +330,17 @@ export interface TrackRequestOptions {
 	pending: Set<Request>;
 }
 
+/**
+ * A requested track written before any subscriber sees it; see `hooks.stageRequest`.
+ * Write through {@link producer}; {@link commit} publishes it with everything written so far.
+ */
+export interface StagedTrack {
+	/** The track's writer. */
+	readonly producer: Producer;
+	/** Resolve every waiting subscriber now. Throws, rejecting the request, if the info is refused. */
+	commit(): Producer;
+}
+
 /** Maximum buffered datagrams per subscriber; mirrors Rust's bounded send buffer. */
 export const MAX_DATAGRAMS = 64;
 
@@ -354,6 +365,13 @@ export const hooks: {
 	 * continues past it. Mirrors the Rust `close` on a `received()` copy; assigned by `track.ts`.
 	 */
 	releaseTrack: (producer: Producer) => void;
+	/**
+	 * Record that the publisher gave up the groups `[start, end)` with `error`: none will
+	 * arrive, so a reader waiting on one moves past it, a group still arriving in them ends with
+	 * `error`, the declared end does not wait for them, and a relay serving this track forwards
+	 * the drop. Mirrors the Rust `drop_groups`; assigned by `track.ts`.
+	 */
+	dropTrackGroups: (producer: Producer, start: number, end: number, error: Error) => void;
 	/** The producer's largest reported location and its subsequent changes. */
 	producerLargest: (producer: Producer) => Getter<Position | undefined>;
 
@@ -372,6 +390,11 @@ export const hooks: {
 	startTrack: (producer: Producer, served: { start?: number; requested?: number }) => void;
 	/** Mint a track {@link Request}; assigned by `track.ts`. */
 	makeRequest: (options: TrackRequestOptions) => Request;
+	/**
+	 * Take a request's producer without resolving its subscribers, so content written before
+	 * {@link StagedTrack.commit} is there when the first reader looks; assigned by `track.ts`.
+	 */
+	stageRequest: (request: Request, info: Partial<Info>) => StagedTrack;
 	/** The newest frame a producer's cache holds, read without subscribing; assigned by `track.ts`. */
 	cachedLargest: (producer: Producer) => Position | undefined;
 	/** The publisher priority from a demanded track's committed {@link Info}; assigned by `track.ts`. */
@@ -521,6 +544,9 @@ export const hooks: {
 	releaseTrack: () => {
 		throw new Error("track.ts not loaded");
 	},
+	dropTrackGroups: () => {
+		throw new Error("track.ts not loaded");
+	},
 	producerLargest: () => {
 		throw new Error("track.ts not loaded");
 	},
@@ -536,6 +562,9 @@ export const hooks: {
 	makeRequest: () => {
 		throw new Error("track.ts not loaded");
 	},
+	stageRequest: () => {
+		throw new Error("track.ts not loaded");
+	},
 	cachedLargest: () => {
 		throw new Error("track.ts not loaded");
 	},

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

@@ -879,7 +879,7 @@ for (const version of [Version.DRAFT_06, Version.DRAFT_07]) {
 			expect(await sub.servedSequence()).toBe(0);
 			expect("start" in (await decodeSubscribeResponse(sub.client.reader, version))).toBe(true);
 
-			sub.track.dropGroups(1, 3, new StreamError(StreamCode(70), { message: "cut" }));
+			hooks.dropTrackGroups(sub.track, 1, 3, new StreamError(StreamCode(70), { message: "cut" }));
 			const resp = await decodeSubscribeResponse(sub.client.reader, version);
 			if (!("drop" in resp)) throw new Error("expected SUBSCRIBE_DROP");
 			expect([resp.drop.start, resp.drop.end, resp.drop.error]).toEqual([
@@ -1201,7 +1201,7 @@ for (const skipped of ["failed", "stale"] as const) {
 			group.writeFrame(textFrame("old", Timestamp.fromMillis(5)));
 			sub.track.writeGroup(group);
 			group.close(skipped === "failed" ? new Error("cut") : undefined);
-			if (skipped === "failed") sub.track.dropGroups(7, 8, new Error("cut"));
+			if (skipped === "failed") hooks.dropTrackGroups(sub.track, 7, 8, new Error("cut"));
 			sub.serve(6);
 			const response = await decodeSubscribeResponse(sub.client.reader, Version.DRAFT_06);
 			if (!("start" in response)) throw new Error("expected SUBSCRIBE_START");

js/net/src/lite/subscriber.ts

@@ -1429,7 +1429,7 @@ export class Subscriber {
 				// The publisher will never send these groups, so they are accounted for without a
 				// stream, and a reader of one still arriving ends with the drop's error now.
 				entry.tail.account(resp.drop.start, resp.drop.end + 1);
-				entry.track.dropGroups(resp.drop.start, resp.drop.end + 1, resp.drop.cause(this.version));
+				hooks.dropTrackGroups(entry.track, resp.drop.start, resp.drop.end + 1, resp.drop.cause(this.version));
 			}
 		}
 	}

js/net/src/resume.test.ts

@@ -1023,7 +1023,7 @@ for (const fails of [true, false]) {
 		expect(tb.producer.demand().wanted.peek()).not.toBeUndefined();
 
 		const cut = new StreamError(StreamCode(70), { message: "cut" });
-		tb.producer.dropGroups(0, 1, cut);
+		hooks.dropTrackGroups(tb.producer, 0, 1, cut);
 		await expect(read).rejects.toBe(cut);
 
 		sub.close();
@@ -1053,9 +1053,9 @@ test("a resumed reader reports what its serving copy gives up, once, and never s
 	ta.write(0);
 	expect(recvDisposition(sub)).toBe(0);
 	// A route failing is no verdict: another route may still deliver the group.
-	ta.producer.dropGroups(1, 2, new StreamError(StreamCode.SessionClosed));
+	hooks.dropTrackGroups(ta.producer, 1, 2, new StreamError(StreamCode.SessionClosed));
 	expect(recvDisposition(sub)).toBe("idle");
-	ta.producer.dropGroups(2, 4, new StreamError(StreamCode(70), { message: "cut" }));
+	hooks.dropTrackGroups(ta.producer, 2, 4, new StreamError(StreamCode(70), { message: "cut" }));
 	expect(recvDisposition(sub)).toEqual([2, 4, 70]);
 	expect(recvDisposition(sub)).toBe("idle");
 	// Re-created under a run already reported dropped.
@@ -1110,7 +1110,7 @@ test("a drop waits while a copy the serving one replaced may still deliver it",
 		front.serve(b.consume());
 		await settle();
 		// Only the new route gave groups 1 and 2 up; the old one is not yet bounded.
-		tb.producer.dropGroups(1, 3, new StreamError(StreamCode(70), { message: "cut" }));
+		hooks.dropTrackGroups(tb.producer, 1, 3, new StreamError(StreamCode(70), { message: "cut" }));
 		expect(recvDisposition(sub)).toBe("idle");
 		ta.write(1);
 		expect(recvDisposition(sub)).toBe(1);
@@ -1148,7 +1148,7 @@ test("a drop a replaced route held back is no verdict on the route serving next"
 	front.serve(b.consume());
 	await settle();
 	// Held back: the copy b replaced may still deliver group 1.
-	tb.producer.dropGroups(1, 2, new StreamError(StreamCode(70), { message: "cut" }));
+	hooks.dropTrackGroups(tb.producer, 1, 2, new StreamError(StreamCode(70), { message: "cut" }));
 	expect(recvDisposition(sub)).toBe("idle");
 	front.serve(c.consume());
 	await settle();
@@ -1176,7 +1176,7 @@ test("a front serving another front reports the inner serving route's drop", asy
 	const sub = outer.consumer.track("seq").subscribe();
 	ta.write(0);
 	expect(recvDisposition(sub)).toBe(0);
-	ta.producer.dropGroups(1, 3, new StreamError(StreamCode(70), { message: "cut" }));
+	hooks.dropTrackGroups(ta.producer, 1, 3, new StreamError(StreamCode(70), { message: "cut" }));
 	expect(recvDisposition(sub)).toEqual([1, 3, 70]);
 	expect(recvDisposition(sub)).toBe("idle");
 	// Re-created under a run already reported dropped.
@@ -1960,7 +1960,7 @@ test.each(["lands", "dropped", "expired"])(
 			expect(drain(sub)).toEqual([]);
 			expect(ta.producer.demand().wanted.peek()).toBeDefined();
 
-			if (outcome === "dropped") ta.producer.dropGroups(1, 2, new NotFound("group"));
+			if (outcome === "dropped") hooks.dropTrackGroups(ta.producer, 1, 2, new NotFound("group"));
 			if (outcome === "expired") await clock.advance(Nano.fromMilli(SECOND));
 			if (outcome !== "lands") {
 				// The groups drained above are released once their close settles.

js/net/src/track.test.ts

@@ -2949,7 +2949,7 @@ const DROPPED = new StreamError(StreamCode.NotFound, { message: "dropped" });
 test("explicit and producer-cut drops survive outside the group cache", async () => {
 	const producer = new TrackProducer("video").accept({});
 	const reader = producer.subscribe({ maxDelay: Milli(60_000) });
-	producer.dropGroups(3, 4, DROPPED);
+	hooks.dropTrackGroups(producer, 3, 4, DROPPED);
 	expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "drop", start: 3, end: 4, error: DROPPED });
 
 	const group = new GroupProducer(4);
@@ -3000,7 +3000,7 @@ test("an upstream drop ends a group still arriving with its error", async () =>
 	expect(group?.sequence).toBe(3);
 	expect(await group?.readString()).toBe("3");
 	const cut = new StreamError(StreamCode(70), { message: "cut" });
-	producer.dropGroups(3, 5, cut);
+	hooks.dropTrackGroups(producer, 3, 5, cut);
 	// Settled by the drop itself, not a later reset; the finished group stays whole.
 	await expect(group?.readString()).rejects.toBe(cut);
 	const whole = await reader.recvGroup();
@@ -3014,7 +3014,7 @@ test("an upstream drop ends a group still arriving with its error", async () =>
 test("a drop below the window is reported once the window lowers to it", () => {
 	const producer = new TrackProducer("video").accept({});
 	const reader = producer.subscribe({ maxDelay: Milli(60_000), groups: { start: { included: 5 } } });
-	producer.dropGroups(2, 3, DROPPED);
+	hooks.dropTrackGroups(producer, 2, 3, DROPPED);
 	expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "idle" });
 
 	reader.withRange({ start: { included: 0 } });
@@ -3045,12 +3045,12 @@ test("a re-created cut group is served, not dropped", async () => {
 test("a wide drop is held and reported as one run", () => {
 	const producer = new TrackProducer("video").accept({});
 	const reader = producer.subscribe({ maxDelay: Milli(60_000) });
-	producer.dropGroups(2, 2 ** 40, DROPPED);
+	hooks.dropTrackGroups(producer, 2, 2 ** 40, DROPPED);
 	expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "drop", start: 2, end: 2 ** 40, error: DROPPED });
 
 	// A wider drop reports only what this cursor has not reported yet, with its own error.
 	const wider = new StreamError(StreamCode.Old, { message: "old" });
-	producer.dropGroups(1, 2 ** 40 + 2, wider);
+	hooks.dropTrackGroups(producer, 1, 2 ** 40 + 2, wider);
 	expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "drop", start: 1, end: 2, error: wider });
 	expect(hooks.tryRecvGroup(reader, true)).toEqual({
 		kind: "drop",
@@ -3069,7 +3069,12 @@ test("adjacent drops with the same code merge into one run", () => {
 	const reader = producer.subscribe({ maxDelay: Milli(60_000) });
 	// Each DROP decodes to its own error, so runs must merge by code, not identity.
 	for (let sequence = 2; sequence < 5; sequence++) {
-		producer.dropGroups(sequence, sequence + 1, new StreamError(StreamCode.NotFound, { message: "dropped" }));
+		hooks.dropTrackGroups(
+			producer,
+			sequence,
+			sequence + 1,
+			new StreamError(StreamCode.NotFound, { message: "dropped" }),
+		);
 	}
 	const recv = hooks.tryRecvGroup(reader, true);
 	expect(recv.kind === "drop" ? [recv.start, recv.end, recv.error] : recv).toEqual([2, 5, DROPPED]);
@@ -3603,7 +3608,7 @@ for (const abort of [false, true]) {
 		const next = reader.recvGroup();
 		expect(await settled(next)).toBe(false);
 
-		producer.dropGroups(0, 2, DROPPED);
+		hooks.dropTrackGroups(producer, 0, 2, DROPPED);
 		if (abort) producer.close(new Error("dropped"));
 		expect(await next).toBeUndefined();
 		if (abort) expect(await reader.closed).toBeNull();