fix(net): stream FETCH payloads and cancel abandoned frames

js/justfile

@@ -61,7 +61,7 @@ test-changed list:
 coverage package="":
     python3 ../sh/js/coverage.py '{{ package }}'
 
-# Run one net benchmark: broadcasts, forward, requests, fills, reader, frames, track, resume, observed, drift, ledger, bandwidth, replay, varint, lite-varint, priority, subscribe-drop, refresh, or compression.
+# Run one net benchmark: broadcasts, forward, requests, fills, fetch, reader, frames, track, resume, observed, drift, ledger, bandwidth, replay, varint, lite-varint, priority, subscribe-drop, refresh, or compression.
 bench name:
     bun install --frozen-lockfile
     bun net/bench/{{ name }}.ts

js/net/bench/fetch.ts

@@ -0,0 +1,124 @@
+/** Sweep cached groups, independent FETCH peers, and small or chunked frame payloads. */
+import { Milli } from "@moq/clock";
+import { Fetch, FetchHeader, FetchOk } from "../src/ietf/fetch.ts";
+import { decodeFetchObject } from "../src/ietf/object.ts";
+import { Publisher } from "../src/ietf/publisher.ts";
+import { NativeSession } from "../src/ietf/session.ts";
+import { ALPN, Version } from "../src/ietf/version.ts";
+import { createMockTransportPair } from "../src/mock.ts";
+import { Producer } from "../src/origin.ts";
+import * as Path from "../src/path.ts";
+import { Reader, Stream } from "../src/stream.ts";
+import { SEGMENT_BYTES } from "../src/util/limits.ts";
+
+console.debug = () => {};
+const version = Version.DRAFT_22;
+const frames = 4;
+console.log("groups,subscribers,frame_bytes,fetch_us");
+for (const groups of [2, 128]) {
+	for (const subscribers of [1, 8]) {
+		for (const size of [16, 1000, SEGMENT_BYTES * 2]) {
+			const origin = new Producer();
+			const publication = origin.createBroadcast(Path.from("fetch"));
+			publication.announce();
+			const track = publication.broadcast.createTrack("video", { maxAge: Milli(60_000) });
+			for (let sequence = 0; sequence < groups; sequence++) {
+				const group = track.appendGroup();
+				// Only the selected group needs payloads; the rest expose cache-table scans.
+				if (sequence === groups - 1) {
+					for (let index = 0; index < frames; index++) {
+						if (size <= SEGMENT_BYTES) group.writeFrame({ payload: new Uint8Array(size).fill(index) });
+						else {
+							const frame = group.createFrame({ size });
+							for (let offset = 0; offset < size; offset += SEGMENT_BYTES)
+								frame.write(new Uint8Array(Math.min(SEGMENT_BYTES, size - offset)).fill(index));
+							frame.finish();
+						}
+					}
+				}
+				group.finish();
+			}
+			const peers = Array.from({ length: subscribers }, () => {
+				const pair = createMockTransportPair(ALPN.DRAFT_22);
+				const publisher = new Publisher({
+					quic: pair.server,
+					session: new NativeSession(pair.server, version, false),
+					publish: origin.consume(),
+					requiresSolicitation: true,
+				});
+				return { pair, publisher, uni: pair.client.incomingUnidirectionalStreams.getReader() };
+			});
+			const samples: number[] = [];
+			try {
+				for (let round = 0; round < 7; round++) {
+					const start = performance.now();
+					await Promise.all(
+						peers.map(async ({ pair, publisher, uni }) => {
+							const client = await Stream.open(pair.client, { version });
+							const server = await Stream.accept(pair.server, version);
+							if (!server) throw new Error("missing FETCH request stream");
+							const requestId = BigInt(round);
+							const groupId = BigInt(groups - 1);
+							const running = publisher.runFetch(
+								new Fetch({
+									requestId,
+									type: {
+										kind: "filtered",
+										trackNamespace: Path.from("fetch"),
+										trackName: "video",
+										filter: {
+											kind: "absolute",
+											startGroup: groupId,
+											startObject: 0n,
+											endGroup: groupId,
+											endObject: BigInt(frames - 1),
+										},
+									},
+								}),
+								server,
+							);
+							try {
+								if ((await client.reader.u53()) !== FetchOk.id) throw new Error("missing FETCH_OK");
+								await FetchOk.decode(client.reader, version);
+								const stream = (await uni.read()).value;
+								if (!stream) throw new Error("missing FETCH data stream");
+								const reader = new Reader(stream, undefined, version);
+								if ((await reader.u53()) !== FetchHeader.type) throw new Error("wrong data stream");
+								if ((await FetchHeader.decode(reader, version)).requestId !== requestId)
+									throw new Error("wrong FETCH ID");
+								for (let index = 0; index < frames; index++) {
+									const object = await reader.decode((cursor) =>
+										decodeFetchObject(cursor, undefined),
+									);
+									if (object.kind !== "object" || object.size !== size)
+										throw new Error("wrong FETCH object");
+									if (object.payload) {
+										if (object.payload.some((byte) => byte !== index))
+											throw new Error("wrong FETCH payload");
+									} else {
+										for (let left = size; left > 0; ) {
+											const chunk = await reader.readChunk(Math.min(left, SEGMENT_BYTES));
+											if (chunk.some((byte) => byte !== index))
+												throw new Error("wrong FETCH chunk");
+											left -= chunk.byteLength;
+										}
+									}
+								}
+								if (!(await reader.done())) throw new Error("extra FETCH objects");
+							} finally {
+								client.close();
+								await running;
+							}
+						}),
+					);
+					if (round > 1) samples.push((performance.now() - start) * 1000);
+				}
+				samples.sort((a, b) => a - b);
+				console.log(`${groups},${subscribers},${size},${samples[2].toFixed(2)}`);
+			} finally {
+				for (const peer of peers) peer.uni.releaseLock();
+				origin.close();
+			}
+		}
+	}
+}

js/net/src/group.test.ts

@@ -886,3 +886,74 @@ test("a drained unfinished group fails with Old when its budget gives up", async
 	expect(consumer.done).toBe(true);
 	expect(await consumer.closed).toBe(err as StreamError);
 });
+
+// Metadata waits use the same recovery and reader verdict as payload reads, without
+// consuming the frames a later FETCH must still deliver.
+test.each([undefined, 2].flatMap((until) => [false, true].map((stalled) => ({ until, stalled }))))(
+	"a group end wait recovers before selecting its end (cap $until, stalled $stalled)",
+	async ({ until, stalled }) => {
+		const source = new Producer(0n);
+		source.writeFrame(textFrame("zero"));
+		const reader = source.consume();
+		const replacement = new Producer(0n);
+		replacement.writeFrame(textFrame("zero"));
+		replacement.writeFrame(textFrame("one"));
+		replacement.finish();
+		const newer = new Signal(false);
+		hooks.recoverGroup(reader, {
+			changed: newer,
+			wants: (failed) => failed !== undefined || newer.peek(),
+			replace: async (index) => {
+				expect(index).toBe(0);
+				return {
+					group: replacement.consume(),
+					expiry: { expired: () => false, changed: [] },
+					adopt: () => newer.set(false),
+					discard: () => {},
+				};
+			},
+		});
+		const end = hooks.groupEnd(reader, until);
+		if (stalled) newer.set(true);
+		else source.abort(new Error("route failed"));
+		expect(await end).toEqual({ end: 2, complete: true });
+		expect(await reader.readString()).toBe("zero");
+		expect(await reader.readString()).toBe("one");
+		reader.close();
+		source.finish();
+	},
+);
+
+// The requested headers are already buffered: a copy failing afterwards reaches the payload
+// reads, which drain them first, rather than refusing an answer that can still be served.
+test("a bounded group end wait is met by buffered headers despite a failed copy", async () => {
+	const source = new Producer(0n);
+	source.writeFrame(textFrame("zero"));
+	const reader = source.consume();
+	const cause = new Error("route failed");
+	const end = hooks.groupEnd(reader, 2);
+	await yieldNow();
+	// Both land before the parked wait looks again.
+	source.writeFrame(textFrame("one"));
+	source.abort(cause);
+	expect(await end).toEqual({ end: 2, complete: false });
+	expect(await reader.readString()).toBe("zero");
+	expect(await reader.readString()).toBe("one");
+	await expect(reader.readFrame()).rejects.toBe(cause);
+});
+
+test.each(["eviction", "expiry"] as const)("a group end wait observes reader %s", async (failure) => {
+	const source = new Producer(0n);
+	source.writeFrame(textFrame("zero"));
+	const reader = source.consume();
+	const expired = new Signal(false);
+	hooks.expireGroup(reader, { expired: () => expired.peek(), changed: [expired] });
+	const end = hooks.groupEnd(reader);
+	if (failure === "eviction") hooks.evictGroup(reader);
+	else expired.set(true);
+	await expect(end).rejects.toMatchObject({
+		code: failure === "eviction" ? StreamCode.TooFarBehind : StreamCode.Old,
+	});
+	source.finish();
+	reader.close();
+});

js/net/src/group.ts

@@ -828,6 +828,7 @@ export class FrameConsumer {
 
 	static {
 		makeFrameConsumer = (sequence, frame) => new FrameConsumer(sequence, frame);
+		hooks.readFrameChunk = (consumer, closed) => consumer.#readChunk(closed);
 		pollChunk = (consumer) => consumer.#poll();
 		chunkChanged = (consumer) => {
 			const frame = consumer.#frame;
@@ -859,11 +860,18 @@ export class FrameConsumer {
 	}
 
 	/** The next chunk of the payload, waiting for it to be written; `undefined` once every byte is read. */
-	async readChunk(): Promise<Uint8Array | undefined> {
+	readChunk(): Promise<Uint8Array | undefined> {
+		return this.#readChunk();
+	}
+
+	async #readChunk(closed?: Promise<void>): Promise<Uint8Array | undefined> {
 		for (;;) {
 			const chunk = this.#poll();
 			if (chunk !== WAIT) return chunk;
-			await Signal.race(...chunkChanged(this));
+			const changed = Signal.race(...chunkChanged(this));
+			if (closed === undefined) await changed;
+			// A frame changes with a byte count or a verdict; only the stream resolves void.
+			else if ((await race([changed, closed])) === undefined) return undefined;
 		}
 	}
 
@@ -1173,6 +1181,7 @@ export class Consumer {
 		hooks.groupIndex = (group) => group.#index();
 		hooks.groupStarted = (group) =>
 			group.#state.total.peek() > group.#state.offset || group.#state.closed.peek() !== undefined;
+		hooks.groupEnd = (group, until) => group.#end(until);
 		hooks.groupFate = (group) => {
 			const state = group.#state;
 			return () => {
@@ -1200,6 +1209,41 @@ export class Consumer {
 		};
 	}
 
+	// Resolve a FETCH's end from headers without consuming or pinning payloads. This is a
+	// reader operation: its verdict and replacement copy, not a raw producer end, decide it.
+	async #end(until?: number): Promise<{ end: number; complete: boolean }> {
+		for (;;) {
+			if (this.#terminal) {
+				this.#abandon();
+				throw this.#terminal;
+			}
+			const closed = this.#state.closed.peek();
+			const count = this.frameCount;
+			// The bound's headers are in hand: a failure after them reaches the payload reads,
+			// which drain what is buffered first, rather than refusing the answer.
+			if (until !== undefined && count >= until) {
+				this.#abandon();
+				return { end: until, complete: closed === null && count === until };
+			}
+			if (closed instanceof Error) {
+				if (await this.#resume(closed)) continue;
+				this.#abandon();
+				throw closed;
+			}
+			if (closed !== undefined) {
+				this.#abandon();
+				return { end: count, complete: true };
+			}
+			if (this.#expire()) continue;
+			if (this.#recover?.wants(undefined)) {
+				// Buffered payloads do not satisfy an end wait, so a newer copy may win here.
+				await this.#resume(undefined, false);
+				continue;
+			}
+			await this.#changed();
+		}
+	}
+
 	// Evaluate the drift budget for a read that has nothing buffered and is about to
 	// wait. A group with frames in hand is always drained: the budget bounds a group
 	// that has stalled while the live edge moved on, not a reader slower than the wire.
@@ -1255,7 +1299,7 @@ export class Consumer {
 	// Continue from the serving route's copy once the one being read failed with `failed`, or
 	// stalled (no `failed`) while a newer route serves. Returns whether the read should look
 	// again; throws the error that ends the group once nothing can continue it.
-	async #resume(failed: Error | undefined): Promise<boolean> {
+	async #resume(failed: Error | undefined, preferBuffered = true): Promise<boolean> {
 		const recover = this.#recover;
 		if (this.#torn() || !recover?.wants(failed)) return false;
 		// A stall's attempt learns nothing of a failure since: ask again with it.
@@ -1291,7 +1335,10 @@ export class Consumer {
 		const replacement = attempt.replacement;
 		attempt.replacement = undefined;
 		if (!replacement) return true;
-		if (!failed && (this.#bufferedFrames().length > 0 || this.#state.closed.peek() !== undefined)) {
+		if (
+			!failed &&
+			((preferBuffered && this.#bufferedFrames().length > 0) || this.#state.closed.peek() !== undefined)
+		) {
 			// The copy being read delivered or ended meanwhile; keep reading it.
 			replacement.discard();
 			return true;