diff --git a/js/justfile b/js/justfile index 0a08acd26c9..a73a5642a38 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.test.ts b/js/net/src/group.test.ts index 1c948d1d3f2..fe1e558ed50 100644 --- a/js/net/src/group.test.ts +++ b/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(); +}); diff --git a/js/net/src/group.ts b/js/net/src/group.ts index 7bf33c7b158..4cb6ad0b76a 100644 --- a/js/net/src/group.ts +++ b/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 { + 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; } } @@ -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 { + 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. @@ -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; diff --git a/js/net/src/ietf/fetch.test.ts b/js/net/src/ietf/fetch.test.ts index cd3ae2921ea..13c1e6c11cd 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"; @@ -747,3 +748,205 @@ test("draft-18: a joining FETCH whose subscription is refused is refused", async await task; client.close(); }); + +// 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([ + { 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(0n); + 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, 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, 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 <= prefix; index++) { + const length = index === prefix ? size : prefixSize; + const object = await reader.decode((cursor) => decodeFetchObject(cursor, Timescale.MILLI)); + expect(object).toMatchObject({ kind: "object", size: length, payload: undefined }); + if (object.kind !== "object") throw new Error("expected object"); + expect(object.timestamp?.value).toBe(index); + 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); + } + if (index === prefix) 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; + } + } +}); + +// 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, 0n, ["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", + async (failure) => { + for (const version of VERSIONS) { + using fx = serve(version); + const group = fx.track.createGroup(0n); + 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); + } + }, +); + +// 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 () => { + 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/object.ts b/js/net/src/ietf/object.ts index 48a33628352..4cebeb79dcd 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -421,6 +421,8 @@ 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; /** @@ -429,8 +431,14 @@ export class FetchFrame { */ priority: number; - constructor({ payload, timestamp, priority }: { payload: Uint8Array; timestamp?: Timestamp; priority: number }) { - this.payload = payload; + constructor({ + payload, + streamed, + timestamp, + priority, + }: { payload?: Uint8Array; streamed?: number; timestamp?: Timestamp; priority: number }) { + this.payload = payload ?? new Uint8Array(0); + this.streamed = streamed; this.timestamp = timestamp; this.priority = priority; } @@ -478,7 +486,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 43f59665270..02100a2d75a 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 { finalSequence, firstEligible, hiddenBelow, hooks, presented } from "../internal.ts"; +import { finalSequence, 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"; @@ -1212,16 +1212,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)); } @@ -1397,9 +1398,16 @@ export class Publisher { const out = new FetchWriter({ stream, version, timescale, priority: publisherPriority }); const group = hooks.cachedGroup(cache, fill.start.group); if (!group) throw new NotFound("fill group unavailable"); + const given = untilGivenUp(group, cancelled); try { - await this.#writeFillGroup({ 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(); } @@ -1421,12 +1429,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, @@ -1449,7 +1455,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; @@ -2287,8 +2296,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 = @@ -2303,7 +2312,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) { @@ -2332,8 +2341,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. @@ -2413,49 +2422,56 @@ export class Publisher { const selected = 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); + // 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 { - last = await readFetched(group, fetchSlice(range, selected), signal); - } finally { - group.close(); - } + 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 endOfTrack = last.complete && final !== undefined && finalSequence(final) === group.sequence + 1n; + const sequence = group.sequence; + const groupEnd = BigInt(Math.max(slice.skip, last.end)); + 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 && finalSequence(final) === last.sequence + 1n; - const sequence = 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, + publisherPriority: toWire(info.priority), + // 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, - publisherPriority: toWire(info.priority), - // Its own rank, so it competes with group streams by its priority alone. - sendOrder: sendOrder({ priority }), - }; - return answer; } finally { if (!answer) cache.close(); } @@ -2498,14 +2514,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; @@ -2524,24 +2539,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, @@ -2556,10 +2565,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, last, timescale, publisherPriority, ranking } = answer; + const { range, group, slice, timescale, publisherPriority, 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 @@ -2575,8 +2587,10 @@ export class Publisher { await untilAborted(new FetchHeader({ requestId: msg.requestId }).encode(stream, version), stop); const out = new FetchWriter({ stream, version, timescale, priority: publisherPriority }); - for (const frame of last.frames) await out.object(last.sequence, frame, stop); + 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) { @@ -2585,6 +2599,8 @@ 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) { const opened = stream; @@ -2705,8 +2721,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; @@ -2718,16 +2735,6 @@ interface FetchAnswer { ranking?: Priority; } -/** A group read for a FETCH, so a later eviction cannot truncate what FETCH_OK promised. */ -interface FetchedGroup { - sequence: bigint; - 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 object ID as a model frame index, saturating past what the model can name. */ function safe(object: bigint): number { return object > BigInt(Number.MAX_SAFE_INTEGER) ? Number.MAX_SAFE_INTEGER : Number(object); @@ -2760,33 +2767,6 @@ function fetchSlice(range: FetchRange, sequence: bigint): 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; @@ -2808,19 +2788,55 @@ class FetchWriter { this.#priority = priority; } - /** Write `frame` of group `sequence`, following the prior object of that group, if any. */ - async object(sequence: bigint, frame: { sequence: number } & group.Frame, stop: AbortSignal): Promise { - const obj = new FetchFrame({ payload: frame.payload, timestamp: frame.timestamp, priority: this.#priority }); + /** 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; + const obj = new FetchFrame({ + payload: chunks ? undefined : read.frame.payload, + streamed: chunks?.size || undefined, + timestamp: read.frame.timestamp, + priority: this.#priority, + }); 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); + // 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), + stop, + ); this.#written = true; } } +/** + * `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 diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 0f503cff677..0c6408ec830 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -671,6 +671,8 @@ export const hooks: { /** The next absolute frame this reader can serve. Assigned by `group.ts`. */ groupIndex: (group: GroupConsumer) => number; groupStarted: (group: GroupConsumer) => boolean; + /** 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. */ @@ -679,6 +681,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. */ @@ -930,6 +934,9 @@ export const hooks: { groupStarted: () => { throw new Error("group.ts not loaded"); }, + groupEnd: () => { + throw new Error("group.ts not loaded"); + }, groupFate: () => { throw new Error("group.ts not loaded"); }, @@ -939,6 +946,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 088b6fdcbed..962e19cdb55 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -24,7 +24,9 @@ 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 { Budget } from "../util/yield.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"; @@ -3033,3 +3035,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", 0n); + const b = subscriber.fetchGroup(Path.from("room"), "video", 0n); + 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 3815b0de301..60df226dece 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"; @@ -50,7 +49,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"; @@ -1699,49 +1698,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..ee9a6ab9139 --- /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(0n); + 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/README.md b/quest/m1/unbounded-groups/README.md index 65283e88e80..312bcb3a9af 100644 --- a/quest/m1/unbounded-groups/README.md +++ b/quest/m1/unbounded-groups/README.md @@ -160,4 +160,3 @@ covered only at the model layer (maintainer, 2026-10-30). - [Byte holes](/quest/m1/unbounded-groups/byte-holes.md) - segments inside a frame shed like front frames and flights start at `(frame, byte)` on the current lite wip; the frame cap goes - [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 - [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 3a0bfe47c8c..00000000000 --- a/quest/m1/unbounded-groups/fetch-stream.md +++ /dev/null @@ -1,35 +0,0 @@ -# [M] FETCH egress and ingest stream large frames without buffering - -## 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. - -## 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. diff --git a/sh/gh/nightly.sh b/sh/gh/nightly.sh index bb6f61333a7..22c2026ffc1 100755 --- a/sh/gh/nightly.sh +++ b/sh/gh/nightly.sh @@ -159,7 +159,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