From d6eac8e2ebae818415ae864da09c63638b1de85a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 02:36:14 -0800 Subject: [PATCH 1/6] chore(quest): claim quest/m1/unbounded-groups/fetch-stream.md Quest-Claim: d210e297643ca3ac From b341a6bdb81416e30084eb63e97cb09681a758a4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 02:47:34 -0800 Subject: [PATCH 2/6] fix(net): stream FETCH payloads and cancel abandoned frames Co-Authored-By: GPT-6.1 Sol --- js/justfile | 2 +- js/net/bench/fetch.ts | 124 +++++++++++++ js/net/src/group.ts | 13 +- js/net/src/ietf/fetch.test.ts | 88 ++++++++- js/net/src/ietf/object.ts | 9 +- js/net/src/ietf/publisher.ts | 209 ++++++++++------------ js/net/src/internal.ts | 10 ++ js/net/src/lite/subscriber.test.ts | 30 ++++ js/net/src/lite/subscriber.ts | 52 +----- js/net/src/payload.test.ts | 47 +++++ js/net/src/payload.ts | 4 +- quest/m1/unbounded-groups/fetch-stream.md | 62 ++++--- sh/gh/nightly.sh | 2 +- 13 files changed, 453 insertions(+), 199 deletions(-) create mode 100644 js/net/bench/fetch.ts create mode 100644 js/net/src/payload.test.ts diff --git a/js/justfile b/js/justfile index fbc65ad98f9..cea97b27016 100644 --- a/js/justfile +++ b/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 diff --git a/js/net/bench/fetch.ts b/js/net/bench/fetch.ts new file mode 100644 index 00000000000..b16a9c50d66 --- /dev/null +++ b/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(); + } + } + } +} diff --git a/js/net/src/group.ts b/js/net/src/group.ts index abbc5e6547b..1265bc86f57 100644 --- a/js/net/src/group.ts +++ b/js/net/src/group.ts @@ -822,6 +822,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; @@ -853,11 +854,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 { + readChunk(): Promise { + return this.#readChunk(); + } + + async #readChunk(closed?: Promise): Promise { 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; } } @@ -1167,6 +1175,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.groupSealed = (group) => group.#state.closed; hooks.groupFate = (group) => { const state = group.#state; return () => { diff --git a/js/net/src/ietf/fetch.test.ts b/js/net/src/ietf/fetch.test.ts index fbcd95a6afa..4d14936dc5c 100644 --- a/js/net/src/ietf/fetch.test.ts +++ b/js/net/src/ietf/fetch.test.ts @@ -9,10 +9,11 @@ import * as Path from "../path.ts"; import { Reader, Stream, Writer } from "../stream.ts"; import { Timescale, Timestamp } from "../time.ts"; import type { Order, Producer as TrackProducer } from "../track.ts"; +import { SEGMENT_BYTES } from "../util/limits.ts"; import { wireOf } from "../wire.ts"; import { Fetch, FetchHeader, FetchOk, type FetchType, type Location } from "./fetch.ts"; import type * as Filter from "./filter.ts"; -import { Group as GroupMessage } from "./object.ts"; +import { decodeFetchObject, Group as GroupMessage } from "./object.ts"; import { Publisher } from "./publisher.ts"; import { RequestError } from "./request.ts"; import { NativeSession } from "./session.ts"; @@ -715,3 +716,88 @@ test("draft-18: a joining FETCH whose subscription is refused is refused", async await task; client.close(); }); + +// A bounded FETCH must deliver before its final frame is complete. Keep growing the +// requested slice past the group tail, so collecting either frames or payloads cannot pass. +test.each([ + { count: 40, size: SEGMENT_BYTES + 1 }, + { count: 1, size: SEGMENT_BYTES * 40 }, +])("a bounded FETCH streams $count frames of $size bytes past the tail", async ({ count, size }) => { + for (const version of VERSIONS) { + using fx = serve(version); + const group = fx.track.createGroup(0); + let frame = group.createFrame({ size, timestamp: new Timestamp(0, Timescale.MILLI) }); + const request = requestStream(version); + const task = fx.pub.runFetch( + new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, count)) }), + request.stream, + ); + try { + expect(await request.answer.u53()).toBe(FetchOk.id); + const ok = await FetchOk.decode(request.answer, version); + expect(ok.endLocation).toEqual(endAt(version, 0, count - 1)); + const next = await fx.uni.read(); + if (!next.value) throw new Error("missing fetch stream"); + const reader = new Reader(next.value, undefined, version); + expect(await reader.u53()).toBe(FetchHeader.type); + expect((await FetchHeader.decode(reader, version)).requestId).toBe(FETCH_ID); + for (let index = 0; index < count; index++) { + if (index > 0) frame = group.createFrame({ size, timestamp: new Timestamp(index, Timescale.MILLI) }); + const object = await reader.decode((cursor) => decodeFetchObject(cursor, Timescale.MILLI)); + expect(object).toMatchObject({ kind: "object", size, payload: undefined }); + if (object.kind !== "object") throw new Error("expected object"); + expect(object.timestamp?.value).toBe(index); + // Forward each segment before the rest exists; no whole-frame read can do this. + for (let offset = 0; offset < size; offset += SEGMENT_BYTES) { + const bytes = new Uint8Array(Math.min(SEGMENT_BYTES, size - offset)).fill(index); + frame.write(bytes); + expect(await reader.read(bytes.byteLength)).toEqual(bytes); + } + frame.finish(); + } + // The bound ends the response without waiting for this live group's FIN. + expect(await reader.done()).toBe(true); + await task; + } finally { + request.stream.abort(new Error("test ended")); + group.abort(new Error("test ended")); + await task; + } + } +}); + +// Both streams must fail together if any part of an accepted answer becomes undeliverable. +test.each(["source", "request", "output"] as const)( + "a FETCH stalled inside a frame stops on %s failure", + async (failure) => { + for (const version of VERSIONS) { + using fx = serve(version); + const group = fx.track.createGroup(0); + group.createFrame({ size: SEGMENT_BYTES * 2, timestamp: new Timestamp(0, Timescale.MILLI) }); + const request = requestStream(version); + const task = fx.pub.runFetch( + new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, 1)) }), + request.stream, + ); + expect(await request.answer.u53()).toBe(FetchOk.id); + await FetchOk.decode(request.answer, version); + const next = await fx.uni.read(); + if (!next.value) throw new Error("missing fetch stream"); + const reader = new Reader(next.value, undefined, version); + expect(await reader.u53()).toBe(FetchHeader.type); + await FetchHeader.decode(reader, version); + expect(await reader.decode((cursor) => decodeFetchObject(cursor, Timescale.MILLI))).toMatchObject({ + kind: "object", + size: SEGMENT_BYTES * 2, + payload: undefined, + }); + const cause = new Error("test failure"); + if (failure === "source") group.abort(cause); + else if (failure === "request") request.withdraw(); + else reader.stop(cause); + await task; + if (failure !== "output") expect(await reader.readChunk(1).catch(error)).toBeInstanceOf(Error); + group.abort(cause); + } + }, +); diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index a0c1dda874e..d615dc86780 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -421,11 +421,14 @@ export interface FetchPosition { export class FetchFrame { /** The object payload. */ payload: Uint8Array; + /** The payload length when the caller streams its bytes after the header. */ + streamed?: number; /** The presentation timestamp carried in object properties, when present. */ timestamp?: Timestamp; - constructor({ payload, timestamp }: { payload: Uint8Array; timestamp?: Timestamp }) { - this.payload = payload; + constructor({ payload, streamed, timestamp }: { payload?: Uint8Array; streamed?: number; timestamp?: Timestamp }) { + this.payload = payload ?? new Uint8Array(0); + this.streamed = streamed; this.timestamp = timestamp; } @@ -470,7 +473,7 @@ export class FetchFrame { header.push(encodeVarint(extensions.byteLength, version), extensions); } - header.push(encodeVarint(this.payload.byteLength, version)); + header.push(encodeVarint(this.streamed ?? this.payload.byteLength, version)); await w.write(concat(header)); if (this.payload.byteLength > 0) { await w.write(this.payload); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 1e8ad0f2b1c..823133d23c6 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -22,7 +22,7 @@ import { } from "../error.ts"; import type * as group from "../group.ts"; import type { Hop, Route } from "../hop.ts"; -import { firstEligible, hiddenBelow, hooks, presented } from "../internal.ts"; +import { firstEligible, hiddenBelow, hooks, presented, type ReadGroupFrame } from "../internal.ts"; import type { Consumer as OriginConsumer } from "../origin.ts"; import * as Path from "../path.ts"; import { sendPayload } from "../payload.ts"; @@ -1391,7 +1391,7 @@ export class Publisher { const group = hooks.cachedGroup(cache, Number(fill.start.group)); if (!group) throw new NotFound("fill group unavailable"); try { - await this.#writeFillGroup({ out, group, slice: groupSlice(fill, group.sequence), cancelled }); + await this.#writeFetchGroup({ out, group, slice: groupSlice(fill, group.sequence), cancelled }); } finally { group.close(); } @@ -1414,12 +1414,10 @@ export class Publisher { } /** - * Write the largest group's cached head as fetch objects. - * - * The cap is the Largest Object snapshot: the group may keep growing, but everything - * past the snapshot belongs to the subscription, not the fill. + * Stream one group's selected objects for a FETCH or subscription fill. + * A bound ends the response without waiting for the live group's end. */ - async #writeFillGroup({ + async #writeFetchGroup({ out, group, slice, @@ -1442,7 +1440,10 @@ export class Publisher { if (slice.until !== undefined && next >= slice.until) break; // Reading from the fill's start drops everything below it; see the same read in #runGroup. - const frame = await nextFrame(group, slice.skip, cancelled, out.stream.closed); + const frame = await untilAborted( + race([hooks.readGroupFrame(group, false, slice.skip), out.stream.closed]), + cancelled, + ); if (!frame) break; next = frame.sequence + 1; if (slice.until !== undefined && frame.sequence >= slice.until) break; @@ -2296,7 +2297,7 @@ export class Publisher { try { try { if (named) answer = await this.#fetchRange(msg, named, stop.signal); - else if (joining && join) answer = await this.#fetchJoin(joining, join, stop.signal); + else if (joining && join) answer = this.#fetchJoin(joining, join); else throw new Error("unknown FETCH type"); } catch (err: unknown) { if (stop.signal.reason === cancel) { @@ -2325,8 +2326,8 @@ export class Publisher { } /** - * Resolve a standalone or draft-22 FETCH against its track and read the highest group it - * holds, or the refusal to answer with. + * Resolve a standalone or draft-22 FETCH to its group reader and end metadata, + * or the refusal to answer with. * * A whole-group request waits for the group to finish, so a downstream cache never * mistakes a prefix for a complete group. @@ -2406,51 +2407,61 @@ export class Publisher { const selected = safe(range.start.group); const group = hooks.cachedGroup(cache, selected); if (!group) return { condition: "does_not_exist", reasonPhrase: `group ${range.start.group}` }; - let last: FetchedGroup; + const slice = fetchSlice(range, selected); + // An unbounded answer needs the group's final count for FETCH_OK. Waiting holds + // only its normal tail; a bounded answer already knows its end and can stream now. try { - last = await readFetched(group, fetchSlice(range, selected), signal); - } finally { - group.close(); - } + if (slice.until === undefined) await untilAborted(Promise.resolve(hooks.groupSealed(group)), signal); + const failed = hooks.groupFate(group)(); + if (failed) throw failed; + + // Only a finished track's final object ends it, never a live track's Largest Object. + const final = hooks.subscriberEnd(cache); + const complete = group.isClosed && (slice.until === undefined || group.frameCount <= slice.until); + const endOfTrack = + complete && + final !== undefined && + (final.frame === 0 ? final.group : final.group + 1) === group.sequence + 1; + const sequence = BigInt(group.sequence); + const groupEnd = BigInt(Math.max(slice.skip, group.frameCount)); + const until = slice.until; + let endLocation: Location; + if (Filter.isDraft20(version)) { + // Draft-22's End Location is inclusive. An empty answer still covers its start. + const object = + until === undefined + ? max(saturatingSub(groupEnd, 1n), BigInt(slice.skip)) + : endOfTrack + ? min(BigInt(until) - 1n, saturatingSub(groupEnd, 1n)) + : BigInt(until) - 1n; + endLocation = { group: sequence, object }; + } else if (endOfTrack && range.end.object > 0n) { + // A bounded response cannot promise objects past the track's end. + endLocation = { group: sequence, object: groupEnd }; + } else { + endLocation = range.end; + } - // Only a finished track's final object ends it, never a live track's Largest Object. - const final = hooks.subscriberEnd(cache); - const endOfTrack = - last.complete && - final !== undefined && - (final.frame === 0 ? final.group : final.group + 1) === last.sequence + 1; - const sequence = BigInt(last.sequence); - const groupEnd = BigInt(last.end); - const until = fetchSlice(range, last.sequence).until; - let endLocation: Location; - if (Filter.isDraft20(version)) { - // Draft-22's End Location is inclusive. An empty answer still covers its start. - const object = - until === undefined - ? max(saturatingSub(groupEnd, 1n), BigInt(fetchSlice(range, last.sequence).skip)) - : endOfTrack - ? min(BigInt(until) - 1n, saturatingSub(groupEnd, 1n)) - : BigInt(until) - 1n; - endLocation = { group: sequence, object }; - } else if (endOfTrack && range.end.object > 0n) { - // A bounded response cannot promise objects past the track's end. - endLocation = { group: sequence, object: groupEnd }; - } else { - endLocation = range.end; + answer = { + cache, + owned: true, + range, + group, + slice, + ok: new FetchOk({ + endOfTrack, + endLocation, + properties: trackProperties(info, msg.propertiesWanted), + }), + // A Timestamp goes out only when FETCH_OK actually carries TIMESCALE. + timescale: msg.propertiesWanted ? (info.timescale ?? undefined) : undefined, + // Its own rank, so it competes with group streams by its priority alone. + sendOrder: sendOrder({ priority }), + }; + return answer; + } finally { + if (!answer) group.close(); } - - answer = { - cache, - owned: true, - range, - last, - ok: new FetchOk({ endOfTrack, endLocation, properties: trackProperties(info, msg.propertiesWanted) }), - // A Timestamp goes out only when FETCH_OK actually carries TIMESCALE. - timescale: msg.propertiesWanted ? (info.timescale ?? undefined) : undefined, - // Its own rank, so it competes with group streams by its priority alone. - sendOrder: sendOrder({ priority }), - }; - return answer; } finally { if (!answer) cache.close(); } @@ -2493,14 +2504,13 @@ export class Publisher { } /** - * Read the head of the group a joining FETCH's subscription starts in, every object up to - * the Largest Object it saved, or the refusal to answer with. + * Resolve the head of a joining FETCH through its subscription's saved Largest Object, + * or the refusal to answer with. */ - async #fetchJoin( + #fetchJoin( type: Extract, join: Joinable, - signal: AbortSignal, - ): Promise { + ): FetchAnswer | Refusal { if (join.kind === "unsupported") return { condition: "not_supported", reasonPhrase: "joining filter not supported" }; const largest = join.largest; @@ -2519,24 +2529,18 @@ export class Publisher { const unavailable: Refusal = { condition: "does_not_exist", reasonPhrase: "joining prefix unavailable" }; const group = hooks.cachedGroup(join.track, sequence); if (!group) return unavailable; - let last: FetchedGroup; - try { - last = await readFetched(group, { skip: 0, until: Number(end.object) }, signal); - } catch (err) { - if (signal.aborted) throw err; - return unavailable; - } finally { + // The saved Largest Object proves this prefix's headers exist. Do not read their + // payloads ahead: eviction after FETCH_OK fails the stream, as for a live subgroup. + if (hooks.groupFate(group)() || hooks.groupIndex(group) !== 0 || group.frameCount < Number(end.object)) { group.close(); + return unavailable; } - // The subscription starts right after the saved Largest Object, so the head has to be - // here in full or the two leave a gap. Frames read are consecutive, so a full count - // starts at object 0. - if (last.frames.length !== Number(end.object)) return unavailable; return { cache: join.track, owned: false, range: { start: { group: end.group, object: 0n }, end }, - last, + group, + slice: { skip: 0, until: Number(end.object) }, // Where the subscription starts, never the end of the track. ok: new FetchOk({ endOfTrack: false, endLocation: end, properties: join.properties }), timescale: join.timescale, @@ -2552,7 +2556,7 @@ export class Publisher { */ async #answerFetch(msg: Fetch, request: RequestStream, answer: FetchAnswer, stop: AbortSignal) { const version = this.#session.version; - const { range, last, timescale, ranking } = answer; + const { range, group, slice, timescale, ranking } = answer; let stream: Writer | undefined; try { @@ -2569,7 +2573,7 @@ export class Publisher { await untilAborted(new FetchHeader({ requestId: msg.requestId }).encode(stream, version), stop); const out = new FetchWriter({ stream, version, timescale }); - for (const frame of last.frames) await out.object(last.sequence, frame, stop); + await this.#writeFetchGroup({ out, group, slice, cancelled: stop }); stream.close(); // The request stream ends once the answer is delivered. await untilAborted(stream.closed, stop); @@ -2579,6 +2583,7 @@ export class Publisher { stream?.reset(e); request.abort(e); } finally { + group.close(); // Ranked until it settles, like a group stream. if (stream && ranking) { const opened = stream; @@ -2697,8 +2702,9 @@ interface FetchAnswer { /** Whether the answer subscribed `cache` itself, and so closes it. */ owned: boolean; range: FetchRange; - /** The highest group the answer holds, read ahead so FETCH_OK can say where it ends. */ - last: FetchedGroup; + /** The reader held until the answer finishes; its normal tail bounds retained payloads. */ + group: group.Consumer; + slice: GroupSlice; ok: FetchOk; /** The units objects are stamped in, if any; see {@link RunGroup.timescale}. */ timescale?: Timescale; @@ -2708,16 +2714,6 @@ interface FetchAnswer { ranking?: Priority; } -/** A group read for a FETCH, so a later eviction cannot truncate what FETCH_OK promised. */ -interface FetchedGroup { - sequence: number; - frames: ({ sequence: number } & group.Frame)[]; - /** One past the last object read, or the slice's start when none was. */ - end: number; - /** Whether the read reached the group's own end rather than stopping at the slice's. */ - complete: boolean; -} - /** A wire sequence as a model one, saturating past what the model can name. */ function safe(sequence: bigint): number { return sequence > BigInt(Number.MAX_SAFE_INTEGER) ? Number.MAX_SAFE_INTEGER : Number(sequence); @@ -2759,33 +2755,6 @@ function fetchSlice(range: FetchRange, sequence: number): GroupSlice { }; } -/** Read a group's slice, waiting for its end when no object bound is set. */ -async function readFetched(group: group.Consumer, slice: GroupSlice, signal: AbortSignal): Promise { - const frames: FetchedGroup["frames"] = []; - let end = slice.skip; - const done = (complete: boolean) => ({ sequence: group.sequence, frames, end, complete }); - for (;;) { - if (slice.until !== undefined && end >= slice.until) - return done(group.isClosed && group.frameCount === end && !(group.closed.peek() instanceof Error)); - const frame = await nextFrame(group, slice.skip, signal); - if (!frame) return done(true); - if (slice.until !== undefined && frame.sequence >= slice.until) return done(false); - frames.push(frame); - end = frame.sequence + 1; - } -} - -/** Read the next selected frame until the requester leaves or the stream closes. */ -async function nextFrame( - group: group.Consumer, - from: number, - signal: AbortSignal, - closed?: Promise, -): Promise<({ sequence: number } & group.Frame) | undefined> { - const next = group.readFrameSequence({ from }); - return untilAborted(closed === undefined ? next : race([next, closed]).then((frame) => frame ?? undefined), signal); -} - /** Write one group's fetch objects, inheriting the first object's priority. */ class FetchWriter { readonly stream: Writer; @@ -2803,15 +2772,25 @@ class FetchWriter { this.#timescale = timescale; } - /** Write `frame` of group `sequence`, following the prior object of that group, if any. */ - async object(sequence: number, frame: { sequence: number } & group.Frame, stop: AbortSignal): Promise { - const obj = new FetchFrame({ payload: frame.payload, timestamp: frame.timestamp }); + /** Write `read` of group `sequence`, following the prior object of that group, if any. */ + async object(sequence: number, read: ReadGroupFrame, stop: AbortSignal): Promise { + const chunks = read.chunks?.size ? read.chunks : undefined; + const obj = new FetchFrame({ + payload: read.chunks ? undefined : read.frame.payload, + streamed: chunks?.size, + timestamp: read.frame.timestamp, + }); const position = { group: sequence, - object: frame.sequence, + object: read.sequence, first: !this.#written, }; await untilAborted(obj.encode(this.stream, position, this.#timescale, this.#version), stop); + if (chunks) + await untilAborted( + sendPayload(chunks, (chunk) => this.stream.write(chunk), this.stream.closed), + stop, + ); this.#written = true; } } diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index c2b21129ee1..ce1b8cea150 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -667,6 +667,8 @@ export const hooks: { /** The next absolute frame this reader can serve. Assigned by `group.ts`. */ groupIndex: (group: GroupConsumer) => number; groupStarted: (group: GroupConsumer) => boolean; + /** The copy's producer end, even while this reader still has buffered frames. */ + groupSealed: (group: GroupConsumer) => GetPromise; /** Watch the error a group's copy closes with, ignoring readers' verdicts, without holding the consumer. */ groupFate: (group: GroupConsumer) => () => Error | undefined; /** Record when a track's mirror of a group joined that track at this hop. */ @@ -675,6 +677,8 @@ export const hooks: { expireGroup: (group: GroupConsumer, expiry: Expiry) => void; /** Carry a handed-out group across its logical track's route changes. */ recoverGroup: (group: GroupConsumer, recover: GroupRecover) => void; + /** Read a payload chunk, releasing its wait when the output stream closes. */ + readFrameChunk: (frame: FrameConsumer, closed: Promise) => Promise; /** Read the next frame, as its stored DEFLATE slice when `compressed`, so a lite-07 publisher forwards it without inflating. */ readGroupFrame: (group: GroupConsumer, compressed: boolean, from?: number) => Promise; /** Notify the owning track when a reader reaches its shed front. */ @@ -928,6 +932,9 @@ export const hooks: { groupStarted: () => { throw new Error("group.ts not loaded"); }, + groupSealed: () => { + throw new Error("group.ts not loaded"); + }, groupFate: () => { throw new Error("group.ts not loaded"); }, @@ -937,6 +944,9 @@ export const hooks: { recoverGroup: () => { throw new Error("group.ts not loaded"); }, + readFrameChunk: () => { + throw new Error("group.ts not loaded"); + }, readGroupFrame: () => { throw new Error("group.ts not loaded"); }, diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index fbcd0c55a8d..30225ac6323 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -24,6 +24,8 @@ import * as Path from "../path.ts"; import { Reader, Writer } from "../stream.ts"; import { Timescale } from "../time.ts"; import type { Producer, Subscriber as TrackSubscriber } from "../track.ts"; +import { SEGMENT_BYTES } from "../util/limits.ts"; +import * as Varint from "../varint.ts"; import { type AnnounceBroadcast, AnnounceInit, AnnounceOk, encodeAnnounceBroadcast } from "./announce.ts"; import { Datagram as DatagramMessage } from "./datagram.ts"; import { Group } from "./group.ts"; @@ -3002,3 +3004,31 @@ test("an interleaved replacement mints a fresh instance", () => { minted.release(path, replacement, false); expect(minted.acquire(path)).toBe(replacement); }); + +// Reader demand applies inside a frame too, including when its peer never sends another byte. +test("a FETCH stops a stalled frame only after its last coalesced reader leaves", async () => { + const { quic, streams, opened } = fakeSession(); + const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); + try { + const a = subscriber.fetchGroup(Path.from("room"), "video", 0); + const b = subscriber.fetchGroup(Path.from("room"), "video", 0); + await opened(0); + await answerTrackInfo(streams[0]); + await opened(1); + await streams[1].reading; + streams[1].inbound.enqueue(new Uint8Array([0, ...Varint.encode(SEGMENT_BYTES * 2)])); + const first = await a; + const last = await b; + const firstFrame = await first.nextFrame(); + const lastFrame = await last.nextFrame(); + if (!firstFrame || !lastFrame) throw new Error("missing frame header"); + first.close(); + streams[1].inbound.enqueue(Uint8Array.of(7)); + expect(await lastFrame.readChunk()).toEqual(Uint8Array.of(7)); + last.close(); + const err = fromTransport(await streams[1].aborted) as StreamError; + expect(err.code).toBe(StreamCode.Cancel); + } finally { + subscriber.close(); + } +}); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 4ff89d4421d..d2c96a313a7 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -25,7 +25,6 @@ import * as Instance from "../instance.ts"; import { hiddenBelow, hooks, maxDelayOf, minDelayOf, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; import type { Lister } from "../listing.ts"; import * as Path from "../path.ts"; -import { receiveFrame } from "../payload.ts"; import { sendOrder } from "../priority.ts"; import { type OpenOptions, type Reader, Stream, Writer } from "../stream.ts"; import { Tail } from "../tail.ts"; @@ -49,7 +48,7 @@ import { import { Datagram as DatagramMessage } from "./datagram.ts"; import { closeViolation } from "./error.ts"; import { Fetch as FetchMessage } from "./fetch.ts"; -import { frameDecoder, type Group as GroupMessage, readFrames, type WireFrame } from "./group.ts"; +import { type Group as GroupMessage, readFrames } from "./group.ts"; import { Probe } from "./probe.ts"; import { ProbeLevel, type Setup } from "./setup.ts"; import { StreamId } from "./stream.ts"; @@ -1690,49 +1689,12 @@ export class Subscriber { // the gap. async #runFetchResponse(stream: Stream, group: netGroup.Producer, info: TrackInfo): Promise { try { - const decode = frameDecoder({ - scale: info.timescale ?? 0, - compress: info.compress, - }); - - // Serve until the stream FINs, the group closes, or every reader leaves. A group can - // stay open indefinitely (a catalog or JSON stream), so an abandoned fetch is stopped by - // demand, not by the stream ending. `idling` is watched across frames as one stable - // promise; the check is level-triggered, so a coalesced fetch that arrives before we - // cancel re-arms and resumes. - const idle: unique symbol = Symbol("idle"); - let idling = groupUnwanted(group).then((): typeof idle => idle); - // A decode consumes its frame whenever it lands, so one outstanding across a re-arm is - // kept and awaited again rather than abandoned with its frame. - let pending: Promise | undefined; - for (;;) { - // Buffered frames are written without an await, as in a group stream. - let frame = pending === undefined ? stream.reader.tryDecode(decode) : undefined; - if (!frame) { - pending ??= stream.reader.decodeMaybe(decode); - const next = await race([pending, group.closed, idling]); - if (next === idle) { - if (group.isClosed) break; - if (group.demand().wanted.peek()) { - idling = groupUnwanted(group).then((): typeof idle => idle); - continue; - } - // Abandoned mid-group: the truncated group must never end clean. - throw new StreamError(StreamCode.Cancel, { message: "cancel" }); - } - pending = undefined; - // Reset the stream with the group's error, such as Unauthorized on revocation. - if (next instanceof Error) throw next; - if (!next) break; - frame = next; - } - if (frame.payload) hooks.writeGroupFrame(group, frame as netGroup.Frame, frame.inflated); - else if (!(await receiveFrame(stream.reader, group, frame))) { - const closed = group.closed.peek(); - if (closed instanceof Error) throw closed; - break; - } - } + // One guard covers both header waits and a stalled payload. Abandonment aborts + // the group before resetting the stream, releasing the shared frame pump too. + await untilAbandoned( + group, + readFrames(stream.reader, group, { scale: info.timescale ?? 0, compress: info.compress }), + ); group.finish(); stream.close(); diff --git a/js/net/src/payload.test.ts b/js/net/src/payload.test.ts new file mode 100644 index 00000000000..c8d00229094 --- /dev/null +++ b/js/net/src/payload.test.ts @@ -0,0 +1,47 @@ +import { expect, spyOn, test } from "bun:test"; +import { type Dispose, Signal, type Subscriber } from "@moq/signals"; +import { Producer } from "./group.ts"; +import { sendPayload } from "./payload.ts"; + +// A stopped sender must release its wait even when the source frame stays open forever. +// Otherwise each cancelled FETCH leaves a reader and its listeners on the shared frame. +test("stopping a payload releases its pending chunk wait", async () => { + const producer = new Producer(0); + producer.createFrame({ size: 2 }); + const group = producer.consume(); + const frame = await group.nextFrame(); + if (!frame) throw new Error("missing frame"); + let stop!: () => void; + const closed = new Promise((resolve) => (stop = resolve)); + let listeners = 0; + const changed = Signal.prototype.changed; + function tracked(this: Signal): Promise; + function tracked(this: Signal, fn: Subscriber): Dispose; + function tracked(this: Signal, fn?: Subscriber): Promise | Dispose { + if (!fn) return new Promise((resolve) => this.changed(resolve)); + const dispose = changed.call(this, fn); + listeners++; + return () => { + listeners--; + dispose(); + }; + } + const watch = spyOn(Signal.prototype, "changed").mockImplementation(tracked); + try { + const sending = sendPayload( + frame, + async () => { + throw new Error("no bytes were written"); + }, + closed, + ); + expect(listeners).toBeGreaterThan(0); + stop(); + await expect(sending).rejects.toThrow("stream closed mid-frame"); + expect(listeners).toBe(0); + } finally { + watch.mockRestore(); + producer.abort(new Error("test ended")); + group.close(); + } +}); diff --git a/js/net/src/payload.ts b/js/net/src/payload.ts index 53c5a103521..be73a0ad7b7 100644 --- a/js/net/src/payload.ts +++ b/js/net/src/payload.ts @@ -63,10 +63,10 @@ export function readSize(read: ReadGroupFrame): number { export async function sendPayload( chunks: group.FrameConsumer, write: (chunk: Uint8Array) => Promise, - closed: Promise, + closed: Promise, ): Promise { for (let sent = 0; ; ) { - const chunk = await race([chunks.readChunk(), closed]); + const chunk = await hooks.readFrameChunk(chunks, closed); if (!(chunk instanceof Uint8Array)) { if (sent < chunks.size) throw new Error("stream closed mid-frame"); return; diff --git a/quest/m1/unbounded-groups/fetch-stream.md b/quest/m1/unbounded-groups/fetch-stream.md index 3a0bfe47c8c..ac09bbe0860 100644 --- a/quest/m1/unbounded-groups/fetch-stream.md +++ b/quest/m1/unbounded-groups/fetch-stream.md @@ -1,35 +1,39 @@ -# [M] FETCH egress and ingest stream large frames without buffering +# [M] Verify FETCH streaming and group tails on a quiet Chromium host ## Goal -`@moq/net` serves a moq-transport FETCH object by object as it reads them, -never collecting a requested group slice whole, so a FETCH of a group with -large frames holds no more than the group tail. A lite FETCH whose readers -all leave in the middle of a frame stops reading that frame's payload at -once. The Chromium group-tail memory measurement and the frame benches are -rerun on a quiet host after [moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411) and after this quest. +Rerun the Chromium group-tail memory measurement and frame benches on a quiet +host against the pre-[moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411) +base, the landed chunked storage, and the FETCH streaming head. Record the +comparison in [moq-dev/moq#7721](https://github.com/moq-dev/moq/pull/7721), and +fix any small-frame browser regression it establishes. ## Plan -Decided 2026-10-31 (user, /quest-plan of the follow-ups from the 2026-10-31 -/quest-complete run), from -[moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411). - -- moq-transport FETCH egress: the user: "absolutely don't buffer." - `readFetched` (`js/net/src/ietf/publisher.ts:2760` at be39188000, called at - `:2408` and `:2521`) collects each requested group slice before writing it. Stream it - the way [moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411)'s subgroup egress (the `serveGroup` closure, `:811`) does, with `nextFrame`-style - reads and chunked writes for a payload past one segment. -- lite FETCH whose readers all leave mid-frame: the user: "fix that I - think." Today it keeps reading the frame's payload before it notices. Watch - the reader count while streaming a frame and stop the stream (reset or - `STOP_SENDING`) as soon as none is left. -- Benches: [moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411)'s small-frame hot-path results were inconclusive (paired - head/base medians 0.91 to 1.11 at load 24 to 29), and the Chromium - group-tail memory numbers predate chunked storage. Rerun both on a quiet - host (`js/net/bench/frames.ts` and the Chromium group-tail measurement) - against the pre-[moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411) base and record them in the PR. A small-frame - regression found here is fixed here. -- Milestone: ✅ m1. Not chosen: m2. - -Public API: none. Wire: none. +The implementation in [moq-dev/moq#7721](https://github.com/moq-dev/moq/pull/7721) +streams moq-transport FETCH and subscription fill objects through the existing +chunked group reader, without collecting a requested slice. Bounded FETCHes +answer before payloads finish; whole-group FETCHes wait for the final count +without retaining payloads beyond the ordinary group tail. An evicted answer +fails rather than pinning data for a slow reader. Lite FETCH cancellation watches +reader demand throughout the frame pump, including a stalled payload. A stopped +egress releases its chunk wait. Public API: none. Wire format: none. + +The functional coverage includes a requested slice and an individual frame +larger than the tail, mid-frame source/request/output failures, coalesced lite +readers leaving mid-frame, and disposal of a stopped egress's chunk listener. +`js/net/bench/fetch.ts` sweeps retained groups, independent readers, and small or +chunked frames; Nightly runs it alongside `js/net/bench/frames.ts`. + +Decided 2026-10-31 (user): do not buffer moq-transport FETCH slices; cancel a +lite FETCH promptly when its last reader leaves mid-frame. The requested quiet +comparison remains unfinished. This implementation ran alongside the user's +`/quest-spawn m0/m1` work, so shared-host timings cannot establish the browser +performance outcome. Do not substitute them for the quiet-host comparison. + +## Waiting + +- Maintainer: provide a quiet host or a reserved measurement window for the + Chromium group-tail memory and paired frame benches. Recommend comparing the + pre-7411 base, landed chunked storage, and FETCH streaming head there, then + fixing any demonstrated small-frame regression before completing this quest. diff --git a/sh/gh/nightly.sh b/sh/gh/nightly.sh index afdc1746b93..62341fd3fb8 100755 --- a/sh/gh/nightly.sh +++ b/sh/gh/nightly.sh @@ -158,7 +158,7 @@ case "$1" in done ;; js) - for benchmark in broadcasts requests fills forward reader frames drift ledger bandwidth replay varint lite-varint track resume priority subscribe-drop refresh observed compression; do + for benchmark in broadcasts requests fills fetch forward reader frames drift ledger bandwidth replay varint lite-varint track resume priority subscribe-drop refresh observed compression; do javascript "$benchmark" just js bench "$benchmark" done javascript catalog just js catalog-bench From 66375ef9c6b34dca56493fe1cc24e12b2ce530e7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 03:02:03 -0800 Subject: [PATCH 3/6] fix(net): resolve FETCH ends through reader recovery Co-Authored-By: GPT-6.1 Sol --- js/net/src/group.test.ts | 53 ++++++++++++++++++++ js/net/src/group.ts | 41 ++++++++++++++-- js/net/src/ietf/fetch.test.ts | 59 +++++++++++++++++------ js/net/src/ietf/publisher.ts | 17 +++---- js/net/src/internal.ts | 6 +-- quest/m1/unbounded-groups/fetch-stream.md | 5 +- 6 files changed, 147 insertions(+), 34 deletions(-) diff --git a/js/net/src/group.test.ts b/js/net/src/group.test.ts index 8beadaeb5d5..11c0459af01 100644 --- a/js/net/src/group.test.ts +++ b/js/net/src/group.test.ts @@ -886,3 +886,56 @@ 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(0); + source.writeFrame(textFrame("zero")); + const reader = source.consume(); + const replacement = new Producer(0); + 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(); + }, +); + +test.each(["eviction", "expiry"] as const)("a group end wait observes reader %s", async (failure) => { + const source = new Producer(0); + 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(); +}); diff --git a/js/net/src/group.ts b/js/net/src/group.ts index 1265bc86f57..a67e63718fa 100644 --- a/js/net/src/group.ts +++ b/js/net/src/group.ts @@ -1175,7 +1175,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.groupSealed = (group) => group.#state.closed; + hooks.groupEnd = (group, until) => group.#end(until); hooks.groupFate = (group) => { const state = group.#state; return () => { @@ -1203,6 +1203,38 @@ 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(); + if (closed instanceof Error) { + if (await this.#resume(closed)) continue; + this.#abandon(); + throw closed; + } + const count = this.frameCount; + if (closed !== undefined || (until !== undefined && count >= until)) { + this.#abandon(); + return { + end: Math.min(count, until ?? count), + complete: closed !== undefined && (until === undefined || count <= until), + }; + } + 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. @@ -1258,7 +1290,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 { + async #resume(failed: Error | undefined, preferBuffered = true): Promise { 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. @@ -1294,7 +1326,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; diff --git a/js/net/src/ietf/fetch.test.ts b/js/net/src/ietf/fetch.test.ts index 4d14936dc5c..b66fec85411 100644 --- a/js/net/src/ietf/fetch.test.ts +++ b/js/net/src/ietf/fetch.test.ts @@ -717,43 +717,50 @@ test("draft-18: a joining FETCH whose subscription is refused is refused", async client.close(); }); -// A bounded FETCH must deliver before its final frame is complete. Keep growing the -// requested slice past the group tail, so collecting either frames or payloads cannot pass. +// Every selected header is known, but the last payload grows the slice past the group +// tail. Neither collecting the slice nor reading its last frame whole can deliver this. test.each([ - { count: 40, size: SEGMENT_BYTES + 1 }, - { count: 1, size: SEGMENT_BYTES * 40 }, -])("a bounded FETCH streams $count frames of $size bytes past the tail", async ({ count, size }) => { + { prefix: 3, size: SEGMENT_BYTES * 16 }, + { prefix: 0, size: SEGMENT_BYTES * 40 }, +])("a FETCH streams $prefix cached frames and a $size byte final frame past the tail", async ({ prefix, size }) => { for (const version of VERSIONS) { using fx = serve(version); const group = fx.track.createGroup(0); - let frame = group.createFrame({ size, timestamp: new Timestamp(0, Timescale.MILLI) }); + const prefixSize = SEGMENT_BYTES * 8; + for (let index = 0; index < prefix; index++) { + const frame = group.createFrame({ size: prefixSize, timestamp: new Timestamp(index, Timescale.MILLI) }); + for (let offset = 0; offset < prefixSize; offset += SEGMENT_BYTES) + frame.write(new Uint8Array(SEGMENT_BYTES).fill(index)); + frame.finish(); + } + const frame = group.createFrame({ size, timestamp: new Timestamp(prefix, Timescale.MILLI) }); const request = requestStream(version); const task = fx.pub.runFetch( - new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, count)) }), + new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, prefix + 1)) }), request.stream, ); try { expect(await request.answer.u53()).toBe(FetchOk.id); const ok = await FetchOk.decode(request.answer, version); - expect(ok.endLocation).toEqual(endAt(version, 0, count - 1)); + expect(ok.endLocation).toEqual(endAt(version, 0, prefix)); const next = await fx.uni.read(); if (!next.value) throw new Error("missing fetch stream"); const reader = new Reader(next.value, undefined, version); expect(await reader.u53()).toBe(FetchHeader.type); expect((await FetchHeader.decode(reader, version)).requestId).toBe(FETCH_ID); - for (let index = 0; index < count; index++) { - if (index > 0) frame = group.createFrame({ size, timestamp: new Timestamp(index, Timescale.MILLI) }); + for (let index = 0; index <= prefix; index++) { + const length = index === prefix ? size : prefixSize; const object = await reader.decode((cursor) => decodeFetchObject(cursor, Timescale.MILLI)); - expect(object).toMatchObject({ kind: "object", size, payload: undefined }); + expect(object).toMatchObject({ kind: "object", size: length, payload: undefined }); if (object.kind !== "object") throw new Error("expected object"); expect(object.timestamp?.value).toBe(index); - // Forward each segment before the rest exists; no whole-frame read can do this. - for (let offset = 0; offset < size; offset += SEGMENT_BYTES) { - const bytes = new Uint8Array(Math.min(SEGMENT_BYTES, size - offset)).fill(index); - frame.write(bytes); + for (let offset = 0; offset < length; offset += SEGMENT_BYTES) { + const bytes = new Uint8Array(Math.min(SEGMENT_BYTES, length - offset)).fill(index); + // Forward each segment before the rest exists; no whole-frame read can do this. + if (index === prefix) frame.write(bytes); expect(await reader.read(bytes.byteLength)).toEqual(bytes); } - frame.finish(); + if (index === prefix) frame.finish(); } // The bound ends the response without waiting for this live group's FIN. expect(await reader.done()).toBe(true); @@ -766,6 +773,26 @@ test.each([ } }); +// The group ending before a requested bound determines both the true end and its track flag. +test("a bounded FETCH waits for headers or the final group before promising its end", async () => { + for (const version of VERSIONS) { + using fx = serve(version); + const live = writeGroup(fx.track, 0, ["0-0", "0-1", "0-2"], false); + const answering = fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, 10)) })); + await flush(); + live.finish(); + fx.track.finish(); + const answer = served(await answering); + expect(answer.ok.endOfTrack).toBe(true); + expect(answer.ok.endLocation).toEqual(endAt(version, 0, 2)); + expect(answer.objects).toEqual([ + [0, 0, "0-0"], + [0, 1, "0-1"], + [0, 2, "0-2"], + ]); + } +}); + // Both streams must fail together if any part of an accepted answer becomes undeliverable. test.each(["source", "request", "output"] as const)( "a FETCH stalled inside a frame stops on %s failure", diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 823133d23c6..1eec071199f 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -2281,8 +2281,8 @@ export class Publisher { join = found; } const namespace = named?.trackNamespace ?? (join?.kind === "join" ? join.namespace : undefined); - // A resolved answer holds its head in memory, so a subscription that ends after it, as - // a track finishing right after SUBSCRIBE_OK does, still gets the head it joined. + // A resolved answer owns its group reader independently of the subscription. Ending + // that subscription alone does not cancel the head; source failure or eviction still can. let resolved = false; const lapsed = new Error("subscription ended"); const lapse = @@ -2408,22 +2408,19 @@ export class Publisher { const group = hooks.cachedGroup(cache, selected); if (!group) return { condition: "does_not_exist", reasonPhrase: `group ${range.start.group}` }; const slice = fetchSlice(range, selected); - // An unbounded answer needs the group's final count for FETCH_OK. Waiting holds - // only its normal tail; a bounded answer already knows its end and can stream now. + // Wait for the selected headers or the group's end, through this reader's recovery + // and verdict. The reader holds only the normal tail, not a copied slice. try { - if (slice.until === undefined) await untilAborted(Promise.resolve(hooks.groupSealed(group)), signal); - const failed = hooks.groupFate(group)(); - if (failed) throw failed; + const last = await untilAborted(hooks.groupEnd(group, slice.until), signal); // Only a finished track's final object ends it, never a live track's Largest Object. const final = hooks.subscriberEnd(cache); - const complete = group.isClosed && (slice.until === undefined || group.frameCount <= slice.until); const endOfTrack = - complete && + last.complete && final !== undefined && (final.frame === 0 ? final.group : final.group + 1) === group.sequence + 1; const sequence = BigInt(group.sequence); - const groupEnd = BigInt(Math.max(slice.skip, group.frameCount)); + const groupEnd = BigInt(Math.max(slice.skip, last.end)); const until = slice.until; let endLocation: Location; if (Filter.isDraft20(version)) { diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index ce1b8cea150..9aa93640dab 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -667,8 +667,8 @@ export const hooks: { /** The next absolute frame this reader can serve. Assigned by `group.ts`. */ groupIndex: (group: GroupConsumer) => number; groupStarted: (group: GroupConsumer) => boolean; - /** The copy's producer end, even while this reader still has buffered frames. */ - groupSealed: (group: GroupConsumer) => GetPromise; + /** Wait for the selected headers or this reader's group end, without retaining payloads. */ + groupEnd: (group: GroupConsumer, until?: number) => Promise<{ end: number; complete: boolean }>; /** Watch the error a group's copy closes with, ignoring readers' verdicts, without holding the consumer. */ groupFate: (group: GroupConsumer) => () => Error | undefined; /** Record when a track's mirror of a group joined that track at this hop. */ @@ -932,7 +932,7 @@ export const hooks: { groupStarted: () => { throw new Error("group.ts not loaded"); }, - groupSealed: () => { + groupEnd: () => { throw new Error("group.ts not loaded"); }, groupFate: () => { diff --git a/quest/m1/unbounded-groups/fetch-stream.md b/quest/m1/unbounded-groups/fetch-stream.md index ac09bbe0860..3712970703b 100644 --- a/quest/m1/unbounded-groups/fetch-stream.md +++ b/quest/m1/unbounded-groups/fetch-stream.md @@ -13,8 +13,9 @@ fix any small-frame browser regression it establishes. The implementation in [moq-dev/moq#7721](https://github.com/moq-dev/moq/pull/7721) streams moq-transport FETCH and subscription fill objects through the existing chunked group reader, without collecting a requested slice. Bounded FETCHes -answer before payloads finish; whole-group FETCHes wait for the final count -without retaining payloads beyond the ordinary group tail. An evicted answer +wait for their selected headers, then answer before payloads finish; whole-group +FETCHes wait for the final count. These metadata waits preserve reader recovery, +eviction, and expiry without retaining payloads beyond the ordinary group tail. An evicted answer fails rather than pinning data for a slow reader. Lite FETCH cancellation watches reader demand throughout the frame pump, including a stalled payload. A stopped egress releases its chunk wait. Public API: none. Wire format: none. From 74d02c03c8757304ae257dd2080c58ed1f311b31 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 13:46:19 -0800 Subject: [PATCH 4/6] fix(net): stop a FETCH parked on flow control when its group fails A FETCH or subscription fill write blocked on flow control never read the group again, so a source failure, eviction, or expiry left the response and its group reader held until the request ended. Derive the answer's stop signal from the group's GivenUp verdict, as subgroup egress does, and release the watch once the stream holds every object. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/fetch.test.ts | 48 +++++++++++++++++++++++++++++++++++ js/net/src/ietf/publisher.ts | 41 ++++++++++++++++++++++++++++-- 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/js/net/src/ietf/fetch.test.ts b/js/net/src/ietf/fetch.test.ts index f6a9adee08f..659d27e894a 100644 --- a/js/net/src/ietf/fetch.test.ts +++ b/js/net/src/ietf/fetch.test.ts @@ -828,3 +828,51 @@ test.each(["source", "request", "output"] as const)( } }, ); + +// A write parked on flow control never reads the group, so the group failing has to reach it +// another way, or the answer holds its reader until the request ends. +test("a FETCH parked on flow control stops when its group fails", async () => { + for (const version of VERSIONS) { + using fx = serve(version); + const group = fx.track.createGroup(0n); + const frames = 2048; + for (let index = 0; index < frames; index++) + group.writeFrame({ + payload: new Uint8Array(64).fill(index), + timestamp: new Timestamp(index, Timescale.MILLI), + }); + const request = requestStream(version); + let settled = false; + const task = fx.pub + .runFetch(new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, frames)) }), request.stream) + .finally(() => { + settled = true; + }); + try { + expect(await request.answer.u53()).toBe(FetchOk.id); + await FetchOk.decode(request.answer, version); + const next = await fx.uni.read(); + if (!next.value) throw new Error("missing fetch stream"); + // Nothing reads the fetch stream, so it fills its window and the publisher parks on a write. + await flush(); + expect(settled).toBe(false); + + group.abort(new Error("source failed")); + await flush(); + expect(settled).toBe(true); + // Both streams are reset rather than ending as if the answer were complete. + expect(await request.answer.u53().catch(error)).toBeInstanceOf(Error); + const data = next.value.getReader(); + let failure: unknown; + try { + while (!(await data.read()).done); + } catch (err: unknown) { + failure = err; + } + expect(failure).toBeInstanceOf(Error); + } finally { + request.stream.abort(new Error("test ended")); + await task; + } + } +}); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 8f561923cc3..3be9cc556a4 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -1391,9 +1391,16 @@ export class Publisher { const out = new FetchWriter({ stream, version, timescale }); const group = hooks.cachedGroup(cache, fill.start.group); if (!group) throw new NotFound("fill group unavailable"); + const given = untilGivenUp(group, cancelled); try { - await this.#writeFetchGroup({ out, group, slice: groupSlice(fill, group.sequence), cancelled }); + await this.#writeFetchGroup({ + out, + group, + slice: groupSlice(fill, group.sequence), + cancelled: given.signal, + }); } finally { + given.release(); group.close(); } @@ -2549,10 +2556,13 @@ export class Publisher { * Write FETCH_OK, then the group on a fetch stream named by the FETCH's own request ID. Nothing past FETCH_OK can be refused, so a failure * resets both streams instead. */ - async #answerFetch(msg: Fetch, request: RequestStream, answer: FetchAnswer, stop: AbortSignal) { + async #answerFetch(msg: Fetch, request: RequestStream, answer: FetchAnswer, cancel: AbortSignal) { const version = this.#session.version; const { range, group, slice, timescale, ranking } = answer; + // The group failing for good ends the answer too, even while a write is parked on flow control. + const given = untilGivenUp(group, cancel); + const stop = given.signal; let stream: Writer | undefined; try { // FETCH_OK on every draft, never REQUEST_OK: section 5.2 allows exactly one FETCH_OK or @@ -2570,6 +2580,8 @@ export class Publisher { const out = new FetchWriter({ stream, version, timescale }); await this.#writeFetchGroup({ out, group, slice, cancelled: stop }); stream.close(); + // The stream holds every object it owes, so the group failing later no longer resets it. + given.release(); // The request stream ends once the answer is delivered. await untilAborted(stream.closed, stop); } catch (err: unknown) { @@ -2578,6 +2590,7 @@ export class Publisher { stream?.reset(e); request.abort(e); } finally { + given.release(); group.close(); // Ranked until it settles, like a group stream. if (stream && ranking) { @@ -2781,6 +2794,30 @@ class FetchWriter { } } +/** + * `signal`, also aborted with why `group` was given up: its copy failed with no route to continue + * it, it was evicted, or it expired. A write parked on flow control never reads the group, so + * without this a FETCH would hold its reader until the request ends. Release once nothing is + * left to write. + */ +function untilGivenUp(group: group.Consumer, signal: AbortSignal): { signal: AbortSignal; release(): void } { + const givenUp = new GivenUp(group); + const failed = new AbortController(); + const released = Promise.withResolvers(); + void givenUp + .write(() => released.promise) + .then((reason) => { + if (reason) failed.abort(reason); + }); + return { + signal: AbortSignal.any([signal, failed.signal]), + release: () => { + givenUp.close(); + released.resolve(); + }, + }; +} + /** The Track Properties a SUBSCRIBE_OK, TRACK_STATUS, or FETCH_OK answers with. */ function trackProperties(info: Info, wanted: boolean): Properties.Properties { // INCLUDE_PROPERTIES=0. The block stays present but empty, which also means the track opts From fa2d5bf61cf6311f160a357e7f4d02d1a074af26 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 14:00:13 -0800 Subject: [PATCH 5/6] quest: finish FETCH streaming verification The quiet Chromium comparison of the pre-#7411 base, the landed chunked storage, main, and this head found no small-frame regression from either change, and group tails hold near 32 MiB on every post-#7411 head. Co-Authored-By: Claude Opus 5.5 --- quest/m1/unbounded-groups/README.md | 1 - quest/m1/unbounded-groups/fetch-stream.md | 40 ----------------------- 2 files changed, 41 deletions(-) delete mode 100644 quest/m1/unbounded-groups/fetch-stream.md diff --git a/quest/m1/unbounded-groups/README.md b/quest/m1/unbounded-groups/README.md index 003c394c37e..05319403b84 100644 --- a/quest/m1/unbounded-groups/README.md +++ b/quest/m1/unbounded-groups/README.md @@ -155,4 +155,3 @@ covered only at the model layer (maintainer, 2026-10-30). - [JS byte holes](/quest/m1/unbounded-groups/js-byte-holes.md) - `@moq/net` mirrors the byte-offset wire and continues a frame cut off mid-way - [Log lock profile](/quest/m1/unbounded-groups/log-lock-profile.md) - a lock profile on a perf host confirms the epoch-based `kio::log` cut `GroupState` contention - [Compressed frames have no size cap and inflate as a bounded stream](/quest/m1/unbounded-groups/compressed-stream.md) - A compressed frame of any size publishes and is received in Rust and `@moq/net`, like a plain one -- [FETCH egress and ingest stream large frames without buffering](/quest/m1/unbounded-groups/fetch-stream.md) - `@moq/net` serves a moq-transport FETCH object by object as it reads them, never collecting a requested group slice whole, so a FETCH of a group with large frames holds no more than the group tail diff --git a/quest/m1/unbounded-groups/fetch-stream.md b/quest/m1/unbounded-groups/fetch-stream.md deleted file mode 100644 index 3712970703b..00000000000 --- a/quest/m1/unbounded-groups/fetch-stream.md +++ /dev/null @@ -1,40 +0,0 @@ -# [M] Verify FETCH streaming and group tails on a quiet Chromium host - -## Goal - -Rerun the Chromium group-tail memory measurement and frame benches on a quiet -host against the pre-[moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411) -base, the landed chunked storage, and the FETCH streaming head. Record the -comparison in [moq-dev/moq#7721](https://github.com/moq-dev/moq/pull/7721), and -fix any small-frame browser regression it establishes. - -## Plan - -The implementation in [moq-dev/moq#7721](https://github.com/moq-dev/moq/pull/7721) -streams moq-transport FETCH and subscription fill objects through the existing -chunked group reader, without collecting a requested slice. Bounded FETCHes -wait for their selected headers, then answer before payloads finish; whole-group -FETCHes wait for the final count. These metadata waits preserve reader recovery, -eviction, and expiry without retaining payloads beyond the ordinary group tail. An evicted answer -fails rather than pinning data for a slow reader. Lite FETCH cancellation watches -reader demand throughout the frame pump, including a stalled payload. A stopped -egress releases its chunk wait. Public API: none. Wire format: none. - -The functional coverage includes a requested slice and an individual frame -larger than the tail, mid-frame source/request/output failures, coalesced lite -readers leaving mid-frame, and disposal of a stopped egress's chunk listener. -`js/net/bench/fetch.ts` sweeps retained groups, independent readers, and small or -chunked frames; Nightly runs it alongside `js/net/bench/frames.ts`. - -Decided 2026-10-31 (user): do not buffer moq-transport FETCH slices; cancel a -lite FETCH promptly when its last reader leaves mid-frame. The requested quiet -comparison remains unfinished. This implementation ran alongside the user's -`/quest-spawn m0/m1` work, so shared-host timings cannot establish the browser -performance outcome. Do not substitute them for the quiet-host comparison. - -## Waiting - -- Maintainer: provide a quiet host or a reserved measurement window for the - Chromium group-tail memory and paired frame benches. Recommend comparing the - pre-7411 base, landed chunked storage, and FETCH streaming head there, then - fixing any demonstrated small-frame regression before completing this quest. From 1959b4d2de0b7423ef23cd15cd063d96ad202a96 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 15:12:48 -0800 Subject: [PATCH 6/6] fix(net): serve buffered FETCH bounds and wait for empty frames to commit A bounded FETCH end wait now returns once its requested headers are buffered, even if the copy fails before the wait resumes: the payload reads drain buffered frames first, so the answer is served or reset there instead of being refused. An errored copy never claims the group completed. A streamed frame with a zero declared size is an object only once committed. FETCH and subgroup egress now wait for its terminal verdict after writing its header, so a frame aborted later resets the answer instead of completing it. Co-Authored-By: Claude Opus 5.5 --- js/net/src/group.test.ts | 18 +++++++++++++++ js/net/src/group.ts | 15 ++++++++----- js/net/src/ietf/fetch.test.ts | 42 +++++++++++++++++++++++++++++++++++ js/net/src/ietf/publisher.ts | 14 +++++++----- 4 files changed, 77 insertions(+), 12 deletions(-) diff --git a/js/net/src/group.test.ts b/js/net/src/group.test.ts index 6073a45be3f..fe1e558ed50 100644 --- a/js/net/src/group.test.ts +++ b/js/net/src/group.test.ts @@ -924,6 +924,24 @@ test.each([undefined, 2].flatMap((until) => [false, true].map((stalled) => ({ un }, ); +// 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")); diff --git a/js/net/src/group.ts b/js/net/src/group.ts index c2955a6438d..4cb6ad0b76a 100644 --- a/js/net/src/group.ts +++ b/js/net/src/group.ts @@ -1218,18 +1218,21 @@ export class Consumer { 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; } - const count = this.frameCount; - if (closed !== undefined || (until !== undefined && count >= until)) { + if (closed !== undefined) { this.#abandon(); - return { - end: Math.min(count, until ?? count), - complete: closed !== undefined && (until === undefined || count <= until), - }; + return { end: count, complete: true }; } if (this.#expire()) continue; if (this.#recover?.wants(undefined)) { diff --git a/js/net/src/ietf/fetch.test.ts b/js/net/src/ietf/fetch.test.ts index 659d27e894a..0477a0260d8 100644 --- a/js/net/src/ietf/fetch.test.ts +++ b/js/net/src/ietf/fetch.test.ts @@ -829,6 +829,48 @@ test.each(["source", "request", "output"] as const)( }, ); +// An empty frame is an object only once committed: its header alone cannot complete the answer, +// or a frame aborted after it would be served as if it existed. +test.each(["finish", "abort"] as const)("a FETCH of an unfinished empty frame waits for its %s", async (outcome) => { + for (const version of VERSIONS) { + using fx = serve(version); + const group = fx.track.createGroup(0n); + const frame = group.createFrame({ size: 0, timestamp: new Timestamp(0, Timescale.MILLI) }); + const request = requestStream(version); + let settled = false; + const task = fx.pub + .runFetch(new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, 1)) }), request.stream) + .finally(() => { + settled = true; + }); + try { + expect(await request.answer.u53()).toBe(FetchOk.id); + await FetchOk.decode(request.answer, version); + const next = await fx.uni.read(); + if (!next.value) throw new Error("missing fetch stream"); + const reader = new Reader(next.value, undefined, version); + expect(await reader.u53()).toBe(FetchHeader.type); + await FetchHeader.decode(reader, version); + expect(await reader.decode((cursor) => decodeFetchObject(cursor, Timescale.MILLI))).toMatchObject({ + kind: "object", + size: 0, + }); + await flush(); + expect(settled).toBe(false); + + if (outcome === "finish") frame.finish(); + else group.abort(new Error("source failed")); + await task; + if (outcome === "finish") expect(await reader.done()).toBe(true); + else expect(await reader.done().catch(error)).toBeInstanceOf(Error); + } finally { + request.stream.abort(new Error("test ended")); + group.abort(new Error("test ended")); + await task; + } + } +}); + // A write parked on flow control never reads the group, so the group failing has to reach it // another way, or the answer holds its reader until the request ends. test("a FETCH parked on flow control stops when its group fails", async () => { diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 3be9cc556a4..c4eb676b1f8 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -1207,16 +1207,17 @@ export class Publisher { if (slice.until !== undefined && read.sequence >= slice.until) break; // A frame written in pieces streams its payload after the header. - const chunks = read.chunks?.size ? read.chunks : undefined; - const obj = chunks + const chunks = read.chunks; + const obj = chunks?.size ? new Frame({ streamed: chunks.size, timestamp: chunks.timestamp }) : new Frame({ - payload: read.chunks ? new Uint8Array(0) : read.frame.payload, + payload: chunks ? new Uint8Array(0) : read.frame.payload, timestamp: read.frame.timestamp, }); const delta = first ? read.sequence : 0; first = false; await send(() => obj.encode(stream, header.flags, timescale, this.#session.version, delta)); + // Even an empty one: it is an object only once committed, and may still be aborted. if (chunks) await send(() => sendPayload(chunks, (chunk) => stream.write(chunk), stream.closed)); } @@ -2773,10 +2774,10 @@ class FetchWriter { /** Write `read` of group `sequence`, following the prior object of that group, if any. */ async object(sequence: bigint, read: ReadGroupFrame, stop: AbortSignal): Promise { - const chunks = read.chunks?.size ? read.chunks : undefined; + const chunks = read.chunks; const obj = new FetchFrame({ - payload: read.chunks ? undefined : read.frame.payload, - streamed: chunks?.size, + payload: chunks ? undefined : read.frame.payload, + streamed: chunks?.size || undefined, timestamp: read.frame.timestamp, }); const position = { @@ -2785,6 +2786,7 @@ class FetchWriter { first: !this.#written, }; await untilAborted(obj.encode(this.stream, position, this.#timescale, this.#version), stop); + // Even an empty payload: the object exists only once committed, and may still be aborted. if (chunks) await untilAborted( sendPayload(chunks, (chunk) => this.stream.write(chunk), this.stream.closed),