refactor(net)!: public ingest

js/net/MIGRATE.md

@@ -4,6 +4,16 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### 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)
 
 `Track.Info.priority`, `Track.Subscription.priority`, and a fetch's `priority` take `-128..=127`, higher first, and 0 is the unset midpoint; anything outside throws a `RangeError`.

js/net/src/broadcast.test.ts

@@ -58,6 +58,48 @@ test("consumer dedupes repeat subscriptions onto one upstream request", async ()
 	expect((await pendingRequest(consumer))?.name).toBe("video");
 });
 
+test("a staged track resolves its subscribers at commit, with what was written", async () => {
+	const consumer = new TestConsumer();
+	const track = consumer.track("video");
+	const reader = track.subscribe().ordered();
+	const request = await pendingRequest(consumer);
+	if (!request) throw new Error("expected request");
+
+	const staged = request.stage({ timescale: Timescale.MILLI });
+	staged.producer.writeFrame(textFrame("hello"));
+	const pending = Symbol("pending");
+	expect(await Promise.race([track.info(), Promise.resolve(pending)])).toBe(pending);
+
+	staged.commit();
+	expect((await track.info()).timescale).toBe(Timescale.MILLI);
+	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 replacement continues the name's timeline, so its groups can't start before the highest one an
 // earlier producer checked, while a new broadcast starts over.
 test("a replacement producer refuses a rewind past its predecessor", () => {

js/net/src/group.ts

@@ -568,6 +568,7 @@ export class Consumer {
 			group.#state.unverified = true;
 		};
 		hooks.groupArrived = (group) => group.#state.arrived;
+		hooks.groupStarted = (group) => group.#state.total.peek() > 0 || group.#state.closed.peek() !== undefined;
 		hooks.arriveGroup = (group, at) => {
 			group.#state.arrived = at;
 		};

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

@@ -1288,15 +1288,15 @@ test.each(["acceptance", "timeout"])("returning demand preserves pending setup u
 	const subscriber = new Subscriber({ session });
 	const opening = spyOn(session, "openBi");
 	const broadcast = subscriber.consume(Path.from("room"));
-	const pending = hooks.pendingTrackProducer;
+	const make = hooks.makeRequest;
 	let returned: track.Subscriber | undefined;
 	const resumed = Promise.withResolvers<void>();
 	const waiting = Promise.withResolvers<void>();
 	const unwanted = demandUtil.unwanted;
 	const unwantedSpy = spyOn(demandUtil, "unwanted");
-	const captured = spyOn(hooks, "pendingTrackProducer").mockImplementationOnce((request) => {
-		const producer = pending(request);
-		const demand = producer.demand();
+	const captured = spyOn(hooks, "makeRequest").mockImplementationOnce((options) => {
+		const request = make(options);
+		const demand = request.demand();
 		let returnedOnce = false;
 		unwantedSpy.mockImplementation(async (watched) => {
 			if (watched !== demand) return unwanted(watched);
@@ -1319,7 +1319,7 @@ test.each(["acceptance", "timeout"])("returning demand preserves pending setup u
 				return used;
 			});
 		});
-		return producer;
+		return request;
 	});
 	const first = broadcast.track("video").subscribe();
 	// Restore real timers even if an injection promise outlives the test.
@@ -1366,13 +1366,13 @@ test("setup abandonment closes before a queued viewer can attach", async () => {
 	const session = new NativeSession(pair.server, VERSION, true);
 	const subscriber = new Subscriber({ session });
 	const broadcast = subscriber.consume(Path.from("room"));
-	const pending = hooks.pendingTrackProducer;
+	const make = hooks.makeRequest;
 	const resumed = Promise.withResolvers<track.Subscriber>();
 	const unwanted = demandUtil.unwanted;
 	const unwantedSpy = spyOn(demandUtil, "unwanted");
-	const captured = spyOn(hooks, "pendingTrackProducer").mockImplementationOnce((request) => {
-		const producer = pending(request);
-		const demand = producer.demand();
+	const captured = spyOn(hooks, "makeRequest").mockImplementationOnce((options) => {
+		const request = make(options);
+		const demand = request.demand();
 		unwantedSpy.mockImplementationOnce(async (watched) => {
 			await unwanted(watched);
 			const peek = demand.wanted.peek.bind(demand.wanted);
@@ -1388,7 +1388,7 @@ test("setup abandonment closes before a queued viewer can attach", async () => {
 				return used;
 			});
 		});
-		return producer;
+		return request;
 	});
 	const first = broadcast.track("video").subscribe();
 	let returned: track.Subscriber | undefined;
@@ -1433,13 +1433,19 @@ test("abandonment cancels once and a late acceptance cannot capture a reused ali
 	peer.reader.version = version;
 	peer.writer.version = version;
 	const broadcast = subscriber.consume(Path.from("room"));
-	const pending = hooks.pendingTrackProducer;
-	let producer: track.Producer | undefined;
+	const make = hooks.makeRequest;
 	let oldRequest: track.Request | undefined;
-	const captured = spyOn(hooks, "pendingTrackProducer").mockImplementationOnce((request) => {
+	// The error the abandoned request is rejected with.
+	const rejected = Promise.withResolvers<Error | undefined>();
+	const captured = spyOn(hooks, "makeRequest").mockImplementationOnce((options) => {
+		const request = make(options);
+		const reject = request.reject.bind(request);
+		spyOn(request, "reject").mockImplementation((err) => {
+			rejected.resolve(err);
+			reject(err);
+		});
 		oldRequest = request;
-		producer = pending(request);
-		return producer;
+		return request;
 	});
 	const decoded = Promise.withResolvers<void>();
 	const release = Promise.withResolvers<void>();
@@ -1455,13 +1461,13 @@ test("abandonment cancels once and a late acceptance cannot capture a reused ali
 	try {
 		expect(await peer.reader.u53()).toBe(Subscribe.id);
 		const request = await Subscribe.decode(peer.reader, version);
-		if (!producer || !oldRequest) throw new Error("missing pending track");
+		if (!oldRequest) throw new Error("missing pending track");
 		const accepted = spyOn(oldRequest, "accept");
 		await peer.writer.u53(SubscribeOk.id);
 		await new SubscribeOk({ requestId: request.requestId, trackAlias: ALIAS }).encode(peer.writer, version);
 		await decoded.promise;
 		first.close();
-		expect(await producer.closed).toBeInstanceOf(Error);
+		expect(await rejected.promise).toBeInstanceOf(Error);
 		expect(await peer.reader.u53()).toBe(Unsubscribe.id);
 		expect((await Unsubscribe.decode(peer.reader, version)).requestId).toBe(request.requestId);
 

js/net/src/ietf/subscriber.ts

@@ -101,6 +101,8 @@ type Subscription = {
 // before the setup promise settles.
 type SubscribeSetupState = {
 	stream?: RequestStream;
+	// The accepted subscription, which a registered alias names.
+	subscription?: Subscription;
 	registeredAlias?: bigint;
 	/**
 	 * Whether the caller has given up. Setup checks this once the stream exists so a request
@@ -812,26 +814,24 @@ export class Subscriber {
 
 		// Keep the request pending until SUBSCRIBE_OK supplies immutable track metadata.
 		// Group streams already wait on the alias, so early data stays behind this response.
-		const producer = hooks.pendingTrackProducer(request);
-		const subscription: Subscription = { track: producer, tail: new Tail() };
-
 		// Open the stream and wait for SUBSCRIBE_OK under a timeout. State
 		// flows back via `state` so the timeout path can clean up the stream
 		// and any registration if setup eventually finishes.
 		const state: SubscribeSetupState = {};
 		const deadline = new Deadline();
-		const setup = this.#openSubscribe(state, broadcast, request, subscription, requestId, deadline, epoch);
+		const setup = this.#openSubscribe(state, broadcast, request, requestId, deadline, epoch);
 
 		// The publisher can be serving before it answers, so waiting only on the response
 		// would miss the local side going away and leave it serving a track nobody reads.
 		// Demand returning before we commit is not abandonment, matching the serving loop.
-		const demand = producer.demand();
+		const demand = request.demand();
 		const waitAbandoned = async (): Promise<null> => {
 			await demandAbandoned(demand);
 			return null;
 		};
 
 		let stream: RequestStream;
+		let subscription: Subscription;
 		let trackAlias: bigint;
 		// The SUBSCRIBER_PRIORITY the request went out with.
 		let priority: number;
@@ -844,6 +844,7 @@ export class Subscriber {
 				const result = await untilAborted(race([accepted, waitAbandoned()]), valid);
 				if (result !== null) {
 					stream = result.stream;
+					subscription = result.subscription;
 					trackAlias = result.alias;
 					priority = result.priority;
 					maxAge = result.maxAge;
@@ -869,7 +870,11 @@ export class Subscriber {
 			const cleanup = async (afterSetup: boolean) => {
 				state.cancelled = true;
 
-				if (state.registeredAlias !== undefined && this.#aliases.retire(state.registeredAlias, subscription)) {
+				if (
+					state.registeredAlias !== undefined &&
+					state.subscription &&
+					this.#aliases.retire(state.registeredAlias, state.subscription)
+				) {
 					this.#timescales.delete(state.registeredAlias);
 				}
 
@@ -903,6 +908,7 @@ export class Subscriber {
 			return;
 		}
 
+		const producer = subscription.track;
 		let disposeGrant: Dispose | undefined;
 		try {
 			// Which terminal fired decides whether we owe the publisher a cancellation, so
@@ -1136,11 +1142,16 @@ export class Subscriber {
 		state: SubscribeSetupState,
 		broadcast: Path.Valid,
 		request: track.Request,
-		subscription: Subscription,
 		requestId: bigint,
 		deadline: Deadline,
 		epoch: Epoch.Valid | undefined,
-	): Promise<{ stream: RequestStream; alias: bigint; priority: number; maxAge?: number }> {
+	): Promise<{
+		stream: RequestStream;
+		subscription: Subscription;
+		alias: bigint;
+		priority: number;
+		maxAge?: number;
+	}> {
 		const version = this.#session.version;
 
 		const legacy = version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16;
@@ -1213,7 +1224,12 @@ export class Subscriber {
 		if (maxCacheDuration !== undefined && maxCacheDuration > BigInt(Number.MAX_SAFE_INTEGER)) {
 			throw new RangeError("max cache duration exceeds safe milliseconds");
 		}
-		request.accept({
+		if (ok.largest && ok.largest.groupId > BigInt(Number.MAX_SAFE_INTEGER)) {
+			throw new RangeError("largest group exceeds safe integers");
+		}
+		// Staged until the answer is written in below, so no reader sees the track before its
+		// Largest and start.
+		const staged = request.stage({
 			// 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,
@@ -1222,22 +1238,23 @@ export class Subscriber {
 			// A moq-transport peer may stamp arrival times, and nothing on the wire says it didn't.
 			unchecked: ok.properties.timescale != null,
 		});
-		if (ok.largest && ok.largest.groupId > BigInt(Number.MAX_SAFE_INTEGER)) {
-			throw new RangeError("largest group exceeds safe integers");
+		const producer = staged.producer;
+		const subscription: Subscription = { track: producer, tail: new Tail() };
+		state.subscription = subscription;
+		if (ok.largest) {
+			// Object IDs can start past zero or skip, while readers see frame ordinals, so only
+			// the group carries over: its first frame is the edge a reader can count on.
+			hooks.reportedLargest(producer, { group: Number(ok.largest.groupId), frame: 0 });
 		}
 		// The live join starts at the publisher's edge, whatever floor the demand asked for.
 		// With no Largest it skips nothing, so every group is owed: a lower one's stream may
 		// still arrive after a higher one's.
 		const groups = request.subscription.groups;
-		hooks.startTrack(subscription.track, {
+		hooks.startTrack(producer, {
 			start: ok.largest ? Number(ok.largest.groupId) : 0,
 			requested: groups?.start === undefined ? undefined : groupBounds(groups).start,
 		});
-		if (ok.largest) {
-			// Object IDs can start past zero or skip, while readers see frame ordinals, so only
-			// the group carries over: its first frame is the edge a reader can count on.
-			hooks.reportedLargest(subscription.track, { group: Number(ok.largest.groupId), frame: 0 });
-		}
+		staged.commit();
 
 		try {
 			this.#aliases.set(ok.trackAlias, subscription, { broadcast, name: request.name, epoch });
@@ -1253,7 +1270,13 @@ export class Subscriber {
 			throw err;
 		}
 		state.registeredAlias = ok.trackAlias;
-		return { stream: state.stream, alias: ok.trackAlias, priority: msg.subscriberPriority, maxAge: msg.maxAge };
+		return {
+			stream: state.stream,
+			subscription,
+			alias: ok.trackAlias,
+			priority: msg.subscriberPriority,
+			maxAge: msg.maxAge,
+		};
 	}
 
 	/**

js/net/src/internal.ts

@@ -400,7 +400,7 @@ export const hooks: {
 	 * end may refine the group boundary it falls just before. Throws `ProtocolViolation` on
 	 * any other second end. Readers then hold at a hole below it until the hole fills, folds
 	 * past the grace, or the track closes, since QUIC may deliver a lower group's stream after
-	 * a higher one. Mirrors the Rust `finish_at_pending`; assigned by `track.ts`.
+	 * a higher one. Mirrors the Rust `finish_at` on a `received()` copy; assigned by `track.ts`.
 	 */
 	endTrack: (producer: Producer, end: Location) => void;
 	// private: capability gap
@@ -424,9 +424,6 @@ 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;
-	// pending: /quest/m1/public-model/ingest.md
-	/** Access the existing producer while a request awaits immutable wire metadata. */
-	pendingTrackProducer: (request: Request) => Producer;
 	/** Have a producer write through a broadcast's sequence namespace for its name. */
 	bindSequence: (producer: Producer, sequence: TrackSequence) => void;
 	/** The newest frame a producer's cache holds, read without subscribing; assigned by `track.ts`. */
@@ -453,12 +450,6 @@ export const hooks: {
 	bufferedLargest: (subscriber: Subscriber) => Location | undefined;
 	/** Call `fn` with a group's frame count now and on every change; assigned by `group.ts`. */
 	groupProgress: (group: GroupProducer, fn: (total: number) => void) => Dispose;
-	// pending: /quest/m1/public-model/ingest.md
-	/**
-	 * Record sequences `[start, end)` an upstream publisher explicitly gave up with `error`,
-	 * ending the groups still arriving in them with it.
-	 */
-	dropTrackGroups: (producer: Producer, start: number, end: number, error: Error) => void;
 	// pending: /quest/m1/public-model/serve.md
 	/**
 	 * Take the next group the subscriber's cursor allows, without waiting; assigned by `track.ts`.
@@ -553,6 +544,8 @@ export const hooks: {
 	groupLatest: (group: GroupConsumer) => Timestamp | undefined;
 	/** Return when a group joined its track at this hop, on the installed clock. */
 	groupArrived: (group: GroupConsumer) => Instant;
+	/** Whether a group counts as content: its first frame landed, or it ended. */
+	groupStarted: (group: GroupConsumer) => boolean;
 	/** Record when a track's mirror of a group joined that track at this hop. */
 	arriveGroup: (group: GroupConsumer, at: Instant) => void;
 	/** Keep applying a subscription's drift policy after it hands a group out. */
@@ -601,9 +594,6 @@ export const hooks: {
 	makeRequest: () => {
 		throw new Error("track.ts not loaded");
 	},
-	pendingTrackProducer: () => {
-		throw new Error("track.ts not loaded");
-	},
 	bindSequence: () => {
 		throw new Error("track.ts not loaded");
 	},
@@ -628,9 +618,6 @@ export const hooks: {
 	groupProgress: () => {
 		throw new Error("group.ts not loaded");
 	},
-	dropTrackGroups: () => {
-		throw new Error("track.ts not loaded");
-	},
 	tryRecvGroup: () => {
 		throw new Error("track.ts not loaded");
 	},
@@ -730,6 +717,9 @@ export const hooks: {
 	groupArrived: () => {
 		throw new Error("group.ts not loaded");
 	},
+	groupStarted: () => {
+		throw new Error("group.ts not loaded");
+	},
 	expireGroup: () => {
 		throw new Error("group.ts not loaded");
 	},

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

@@ -1106,7 +1106,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") hooks.dropTrackGroups(sub.track, 7, 8, new Error("cut"));
+			if (skipped === "failed") sub.track.dropGroups(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");