diff --git a/doc/concept/standard.md b/doc/concept/standard.md index aea524e38e..06b6ef8002 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -28,14 +28,14 @@ subset drops. What a peer actually observes against this implementation: - **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. We never hold two overlapping requests at a peer that did not declare solicit: a broader one replaces the narrower ones once they are closed. The solicit `SETUP` option makes us wait to be asked. Without it, a peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We never send `PUBLISH`. - **We list no tracks by `PUBLISH`.** Rust and JavaScript refuse each `SUBSCRIBE_TRACKS` with `NOT_SUPPORTED`, and the session stays up. Our own peers list tracks over moq-lite instead. - **A Rust TRACKS listing sends `SUBSCRIBE_TRACKS` upstream, only on demand.** Listing a broadcast from a moq-transport peer sends `SUBSCRIBE_TRACKS` with `FORWARD=0` while someone lists it, never on connect, announce, or subscribe, so a peer that refuses, resets, or ignores it affects only that listing. Each listed track's paused `PUBLISH` is its upstream copy, which a reader resumes rather than subscribing again. Only a `PUBLISH_DONE` with `TRACK_ENDED` removes a listed track, and it lists as finished; any other end fails the listing, and readers list again. Our own moq-transport publishers refuse it, so a Rust listing of one fails. -- **History is served one group at a time.** Our moq-transport publishers answer a `FETCH` only within one group, a draft-18 joining `FETCH` for the subscription's current group head, and a draft 20+ fill for the largest group's head. Whole-group fetches wait for an open group to finish. Both implementations accept and ignore `FILL_TIMEOUT`, including zero. An explicit absolute subscription start is served from that group on every supported draft, so a reader keeps its prefix even when its answer arrives late. A `SUBSCRIBE` without a Location Filter joins the live group rather than replaying history. A subscriber that wants the track from group 0 sends exactly that filterless `SUBSCRIBE`, the same on every draft, and fetches the groups below the answer's Largest Object with one range `FETCH`, which our publishers refuse once it spans more than one group. A live JavaScript subscription joins at the current group: on draft 18 it fetches the group's head with a Relative Joining `FETCH`, and from draft 20 asks for it with a fill. Subscribers still request ranges upstream, one per run of missing groups; a third-party publisher can answer those ranges and mark their gaps. A Rust reader that only fetches learns the track from `TRACK_STATUS`, so a finished track stays fetchable; a refused `TRACK_STATUS` fails that reader's request, whatever the code, rather than falling back to a `SUBSCRIBE`. +- **History is served one group at a time.** Rust answers a `FETCH` only within one group, a draft-18 joining `FETCH` for the subscription's current group head, and a draft 20+ fill for the largest group's head. A JavaScript publisher refuses every `FETCH` with `NOT_SUPPORTED`, so on draft 18 its Next Object subscription carries the current group whole, from object 0; from draft 20 it answers the fill. Whole-group fetches wait for an open group to finish. Both implementations accept and ignore `FILL_TIMEOUT`, including zero. An explicit absolute subscription start is served from that group on every supported draft, so a reader keeps its prefix even when its answer arrives late. A `SUBSCRIBE` without a Location Filter joins the live group rather than replaying history. A subscriber that wants the track from group 0 sends exactly that filterless `SUBSCRIBE`, the same on every draft, and fetches the groups below the answer's Largest Object with one range `FETCH`, which Rust refuses once it spans more than one group. A live JavaScript subscription joins at the current group: on draft 18 it fetches the group's head with a Relative Joining `FETCH`, and from draft 20 asks for it with a fill. Subscribers still request ranges upstream, one per run of missing groups; a third-party publisher can answer those ranges and mark their gaps. A Rust reader that only fetches learns the track from `TRACK_STATUS`, so a finished track stays fetchable; a refused `TRACK_STATUS` fails that reader's request, whatever the code, rather than falling back to a `SUBSCRIBE`. - **A requester's FIN.** On draft 18 a requester finishing its stream cancels its request, except after `TRACK_STATUS`: nothing can follow it, so the FIN only ends the requester's input and the request is still answered. From draft 20 a FIN never cancels; a reset or `STOP_SENDING` does. - **Compression.** Rust peers exchange compressed tracks when both offer the [Flate extension](/draft/moq-flate). Other peers receive one shared inflated copy. A publisher that seals the compression declaration makes that fallback unavailable: an incompatible peer is refused, preserving the sealed metadata. JavaScript does not offer the IETF extension. - **Timing.** A track whose `SUBSCRIBE_OK` or `PUBLISH` declares no `TIMESCALE` is untimed. A `SUBSCRIBE` or `FETCH` with `INCLUDE_PROPERTIES=0` gets no `TIMESCALE` in its `SUBSCRIBE_OK` or `FETCH_OK`, so its objects go out unstamped. A Rust reader that never subscribed takes the units from `TRACK_STATUS`, and still asks for the properties whenever it subscribes to or fetches a timed track. - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. A `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session. An unknown message parameter, or one defined only for another message, closes the session. Other malformed control input closes the session with the draft's code too, such as a server `SETUP` carrying `PATH` or `AUTHORITY`, a namespace field ending in `/` (it would map to a path with an empty segment), a request cut short by FIN, or a second `GOAWAY` on one stream; a first `GOAWAY` on a request stream is legal. - **One `SETUP` credential**, forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second `SETUP` token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. Where the Rust side authorizes the peer or verifies tokens, each distinct request token is verified once while any open request on that session names it, and its grant authorizes only requests naming it, even where that authorization already covers it; it never widens the session. It is refused with `NOT_SUPPORTED` when nothing verifies it, when it is empty, or when its Token Type is anything but 0 (out of band). A side that does neither, JavaScript included, ignores the token. Rust refuses a request carrying more than one distinct token with `NOT_SUPPORTED`, since it verifies one; JavaScript, which verifies none, ignores them all. A token-authorized `SUBSCRIBE` still stays within the grant the peer gave us. A token-authorized `PUBLISH_NAMESPACE` needs a grant covering everything beneath the namespace, and the tracks we subscribe under it rely on that announce instead of our limit, ending with it. An update on a token-authorized request moves it to the token it carries; the old token ends with its last naming request. A request is admitted by the session or by a token for good, so an update naming a token on one the session admitted is refused with `NOT_SUPPORTED`. A token-less `PUBLISH_NAMESPACE` our limit does not cover is refused `UNAUTHORIZED`, and one the limit stops covering ends; the publisher re-announces once it widens. Expiry or revocation ends every request naming that token. The relay reports the lease end to its auth server, and a later request naming the token verifies it again. A refused renewal ends only that request. Rust and JavaScript send the token a route names (`Route::auth`) on its `PUBLISH_NAMESPACE`, and swapping it renews the request with `REQUEST_UPDATE`; the inline `NAMESPACE` that answers a solicited `SUBSCRIBE_NAMESPACE` carries no token. A token added to a reconnecting connection's auth handle rides every session it makes, so the announcement keeps it across a reconnect; one added to a single session's handle ends with that session. Rust honors a token on any supported request; discovery reveals only paths it grants, and a joining `FETCH` keeps following its subscription. A request repeating a token is refused with `MALFORMED_AUTH_TOKEN`. Additional session tokens in band need [moq-lite-07](/concept/moq-lite#authorization); moq-transport has no AUTH exchange. - **Pause delivery.** `FORWARD=0` on a `SUBSCRIBE` or its update drops the subscription, closing its data streams and holding nothing upstream, while the request and its slot stay. `FORWARD=1` subscribes again as a new `SUBSCRIBE` with the same filter would, so nothing carries across the pause: its `REQUEST_OK` names the live edge it joined at. A paused update may change only priority and max age. -- **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `SUBSCRIBE_TRACKS`, multi-group `FETCH`, and joining requests reaching before the joined group get `NOT_SUPPORTED`. Range Filters get `INVALID_FILTER` from draft 20 because the advertised filter limit is zero. From draft 20, namespaces whose first tuple field is exactly `.` or `.session` get `DOES_NOT_EXIST` before application lookup; other dot prefixes still reach the application. Subscription updates accept priority, negotiated subscriber max age, and delivery timeouts. Rust also carries range changes, keeps bounded subscriptions updatable, renews request tokens, and judges `NEW_GROUP_REQUEST` as a join (see dynamic groups below). From draft 20, a wider or historical fill fails on its separate fetch stream while its subscription stays open; a largest-group fill hands its head to the live subgroup tail. Other supported-but-unimplemented preferences end that subscription; a `TRACK_PROPERTY_FILTER` there closes the session with `PROTOCOL_VIOLATION`, since it belongs only to `SUBSCRIBE_TRACKS`. A namespace update may change its prefix; suffixes after the acknowledgement are relative to that new prefix. +- **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `SUBSCRIBE_TRACKS`, any `FETCH` to JavaScript, multi-group `FETCH`, and joining requests reaching before the joined group get `NOT_SUPPORTED`. Range Filters get `INVALID_FILTER` from draft 20 because the advertised filter limit is zero. From draft 20, namespaces whose first tuple field is exactly `.` or `.session` get `DOES_NOT_EXIST` before application lookup; other dot prefixes still reach the application. Subscription updates accept priority, negotiated subscriber max age, and delivery timeouts. Rust also carries range changes, keeps bounded subscriptions updatable, renews request tokens, and judges `NEW_GROUP_REQUEST` as a join (see dynamic groups below). From draft 20, a wider or historical fill fails on its separate fetch stream while its subscription stays open; a largest-group fill hands its head to the live subgroup tail. Other supported-but-unimplemented preferences end that subscription; a `TRACK_PROPERTY_FILTER` there closes the session with `PROTOCOL_VIOLATION`, since it belongs only to `SUBSCRIBE_TRACKS`. A namespace update may change its prefix; suffixes after the acknowledgement are relative to that new prefix. - **Objects and datagrams.** Only subgroup 0 is accepted: a stream for any other subgroup, whether its header gives the ID or takes it from the first Object ID, is stopped and the session stays up. Rust relays preserve Object IDs, gaps, object properties, FIRST\_OBJECT, and END\_OF\_GROUP status objects between moq-transport peers for subgroup 0, including across a moq-lite-07 hop between two relays that negotiate the [ietf](/draft/moq-ietf) extension. Plain Rust readers and moq-lite peers without the extension see consecutive frames. moq-transport maps a [datagram track](/concept/moq-lite#datagrams) to `SUBSCRIBE` and `OBJECT_DATAGRAM`, with its sequence as the group ID and object ID 0. One upstream subscription serves both kinds of a name. Stream consumers receive only stream objects, and datagram consumers receive only object 0 datagrams, including through relays. An undefined Object Status or malformed datagram framing closes the session. Unreadable object properties drop only that datagram. A relayed Timestamp leaves in the units the downstream response declares, and is dropped where it declares none, as for a plain frame; an immutable one that would change or vanish is refused. On a timed track the frame holds each object's time, so a mutable Timestamp that only repeats the immutable one is not restated; a status object has no frame, so it keeps its Timestamp among its properties. Only property 0x10 is read as a Timestamp. Property 0x06 stays opaque and keeps its value regardless of timestamp units; it is dropped between draft 18 and later drafts, or refused when immutable, because draft 20 assigns it to SUBGROUP\_DELIVERY\_TIMEOUT. A status object carrying properties is refused. A `FETCH` has no status objects, so its response ends at the last frame. A group cannot resume across route changes by frame ordinal once Object IDs skip ahead of the ordinals on either route; recovery refuses it instead of splicing different objects. - **Dynamic groups.** A track advertises `DYNAMIC_GROUPS` when its publisher sets a refresh ratio, and a track received with `DYNAMIC_GROUPS` gets a ratio of 0. A `NEW_GROUP_REQUEST` counts as one more subscriber joining, which a `SUBSCRIBE` already is: it asks for a new group only when the newest group's bytes after its first frame exceed the ratio times the first frame. Joins coalesce until the asked-for group arrives. A Rust relay sends `NEW_GROUP_REQUEST` upstream, on the `SUBSCRIBE` or in a `REQUEST_UPDATE`, once a join wants a group past the newest it holds; a relay holding nothing yet asks for none. On a track without a ratio, an update carrying one ends the subscription with `UPDATE_FAILED`. A moq-lite-07 hop (`moq-lite-07-wip`, opt-in) carries the same demand natively: the ratio in `TRACK_INFO` and the asked-for group in `SUBSCRIBE_UPDATE`. Earlier moq-lite versions carry neither, so a track received over one never refreshes. - **Priority.** Higher is served first. This stack's priorities run from -128 to 127 with 0 unset, and go out on the IETF wire, which ranks lower first, as `127 - priority`: an unset priority is IETF 127. An omitted subscriber or publisher priority reads as the drafts' default of IETF 128, one step below unset, so a relay forwards it unchanged. A relay's upstream subscription carries the highest priority its subscribers ask for and updates it as they come and go; its group order stays as the `SUBSCRIBE` asked, since no draft lets an update change it. diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index dadddf1f5e..9d2b3f5595 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -50,7 +50,7 @@ route.announce(); - **One receive estimate per connection.** `connection.viewers` divides it among watched broadcasts, max-min fair by each viewer's ceiling, so several players on one pooled connection don't each select against the whole downlink. Holding a viewer requests PROBE, and the viewers' ceilings, summed, are the PROBE target a relay advertising `Increase` probes the downlink up to. `setTarget(0)` opts out. `watch`, `update`, and `setTarget` throw a `RangeError` when the ceilings or the target would pass `Viewers.MAX_TARGET` (`Number.MAX_SAFE_INTEGER`), the most a PROBE target carries. - **List a broadcast's tracks.** `for await (const event of broadcast.tracks())` reports each track the broadcast lists (`added`, with its `Track.Info`), one `synced`, then `added` and `removed` as tracks come and go, until the broadcast ends; leaving the loop stops listing. Over a session it needs `moq-lite-07-wip`, and syncs when the peer does, so a track added after `synced` is new at the source. A track served only on request is never listed. See [discovery](/concept/moq-lite#discovery). - **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. The application owns sequence numbers: the broadcast keeps nothing of an ended track, so a track that replaces it, created or requested, starts at 0. A name always means the same content, so the replacement continues past its predecessor's groups by writing its first group with an explicit sequence. Sequences are `bigint`, so every ID the wire carries fits, past 2^53 included: `track.createGroup(5n)`, `{ absolute: 10n }`. Frame indices and relative anchors stay `number`. -- **Serve history on request.** `track.requests()` opens a handle that serves the groups readers miss: ones their ranges name below the newest group written that the cache no longer holds, or never did. Each run of misses arrives as a `Track.Fill` from `await requests.requestedFill()`, shared by every reader missing it. `accept(sequence)` it into the cache and write the group's frames as first published, `dropGroups(start, end)` what the source lacks (sequences are `bigint`), and serve `wanted()` until it is `undefined`: a reader that leaves or narrows stops wanting its groups. `close()` the fill when done; whatever is still wanted is named dropped. A `fetchGroup` of an uncached group asks the handle at once, even for a group not written yet, which a live write then answers; with no handle open it fails with `NotFound`, as in Rust, and a reader's miss is given up. A reader only asks for what its `delay.max` still wants, so a reader at the live edge never asks. This serves lite FETCH, single-group IETF FETCH, and lite-07 ranged subscriptions from the same track, as Rust's `requests()` does. +- **Serve history on request.** `track.requests()` opens a handle that serves the groups readers miss: ones their ranges name below the newest group written that the cache no longer holds, or never did. Each run of misses arrives as a `Track.Fill` from `await requests.requestedFill()`, shared by every reader missing it. `accept(sequence)` it into the cache and write the group's frames as first published, `dropGroups(start, end)` what the source lacks (sequences are `bigint`), and serve `wanted()` until it is `undefined`: a reader that leaves or narrows stops wanting its groups. `close()` the fill when done; whatever is still wanted is named dropped. A `fetchGroup` of an uncached group asks the handle at once, even for a group not written yet, which a live write then answers; with no handle open it fails with `NotFound`, as in Rust, and a reader's miss is given up. A reader only asks for what its `delay.max` still wants, so a reader at the live edge never asks. This serves lite FETCH and lite-07 ranged subscriptions from the same track, as Rust's `requests()` does. A moq-transport FETCH is refused, so a moq-transport reader never reaches it. - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Omitting it means untimed, as in Rust. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` producers. - **Durations and instants are `Clock` types.** Every duration (`maxAge`, `delay`, timeouts, RTT) is a `Clock.Duration` and every monotonic time a `Clock.Instant`, from [`@moq/clock`](./clock), which `@moq/net` re-exports as `Clock`: `maxAge: Moq.Clock.Duration.secs(30)`. Neither converts to a number implicitly; read one with an explicit unit, such as `asMillis()`. The wire still carries whole milliseconds, rounded up. Frame timestamps stay `Time.Timestamp`, a tick count at the track's timescale. - **A subscription's `delay` is a `Track.Delay { min, max }`, as in Rust.** Staleness is `delay.max`: it counts from a group's successor, by media time and by wall clock, whichever reaches it first. `delay.min` is the target latency, carried on moq-lite 07; a peer that cannot carry it leaves it `null` (not stated). Passing the old `maxAge` or `maxDelay` key throws a `TypeError` naming `delay.max`. The publisher's own staleness bound is the separate `Track.Info.maxAge`, which caps `delay.max` and fetches. diff --git a/js/justfile b/js/justfile index ea47805d79..6403176f98 100644 --- a/js/justfile +++ b/js/justfile @@ -70,7 +70,7 @@ test-changed list: coverage package="": python3 ../sh/js/coverage.py '{{ package }}' -# 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, compression, or datagrams. +# Run one net benchmark: broadcasts, forward, requests, fills, reader, frames, track, resume, observed, drift, ledger, bandwidth, replay, varint, lite-varint, priority, subscribe-drop, refresh, compression, or datagrams. bench name: uv run --no-project python ../sh/js/install.py frozen bun net/bench/{{ name }}.ts diff --git a/js/net/bench/fetch.ts b/js/net/bench/fetch.ts deleted file mode 100644 index e1b8800e8f..0000000000 --- a/js/net/bench/fetch.ts +++ /dev/null @@ -1,123 +0,0 @@ -/** Sweep cached groups, independent FETCH peers, and small or chunked frame payloads. */ -import { Duration } 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 route = origin.route(Path.from("fetch")); - const track = route.broadcast.createTrack("video", { maxAge: Duration.millis(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 18dd84b764..18a4c5e615 100644 --- a/js/net/src/group.test.ts +++ b/js/net/src/group.test.ts @@ -886,77 +886,6 @@ test("a drained unfinished group fails with Old when its budget gives up", async 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, new StreamError(StreamCode.Evicted)); - else expired.set(true); - await expect(end).rejects.toMatchObject({ - code: failure === "eviction" ? StreamCode.Evicted : StreamCode.Old, - }); - source.finish(); - reader.close(); -}); - // Rust's `GroupState::shed` keeps an open group's newest committed frame even while a newer // frame is still being written, since that partial frame is not readable yet. test("a debt shed keeps the newest committed frame behind one being written", async () => { diff --git a/js/net/src/group.ts b/js/net/src/group.ts index dc71f859bb..86b0489e59 100644 --- a/js/net/src/group.ts +++ b/js/net/src/group.ts @@ -1000,7 +1000,6 @@ 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 () => { @@ -1034,42 +1033,6 @@ 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 (;;) { - const verdict = this.#verdict.peek(); - if (verdict) { - this.#abandon(); - throw verdict; - } - 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. diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index 45863aacb6..a4a6e7b3e0 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -9,6 +9,7 @@ import * as Path from "../path.ts"; import { Stream, Writer } from "../stream.ts"; import { Timescale } from "../time.ts"; import { Connection } from "./connection.ts"; +import { Fetch, type FetchType } from "./fetch.ts"; import * as Message from "./message.ts"; import * as Namespace from "./namespace.ts"; import { Group } from "./object.ts"; @@ -120,13 +121,16 @@ for (const [version, alpn] of [ } } -/** We list no tracks by PUBLISH: each SUBSCRIBE_TRACKS gets NOT_SUPPORTED, and the session stays usable. */ +/** + * We serve neither FETCH (only a relay answers it, from its cache) nor SUBSCRIBE_TRACKS (we list + * no tracks by PUBLISH): each gets NOT_SUPPORTED on its own stream, and the session stays usable. + */ for (const [version, alpn] of [ [Version.DRAFT_18, ALPN.DRAFT_18], [Version.DRAFT_20, ALPN.DRAFT_20], [Version.DRAFT_22, ALPN.DRAFT_22], ] as const) { - test(`SUBSCRIBE_TRACKS is refused on its own stream on ${alpn}`, async () => { + test(`FETCH and SUBSCRIBE_TRACKS are refused on their own streams on ${alpn}`, async () => { const pair = createMockTransportPair(alpn); const control = await Stream.open(pair.server, { version }); const connection = new Connection({ @@ -137,16 +141,42 @@ for (const [version, alpn] of [ client: false, }); + const requests: ((writer: Writer, requestId: bigint) => Promise)[] = [ + async (writer, requestId) => { + await writer.u62(BigInt(SUBSCRIBE_TRACKS_ID)); + await Message.encode(writer, async (body) => { + await body.u62(requestId); + await Namespace.encode(body, Path.from("prefix")); + await body.u53(0); + }); + }, + async (writer, requestId) => { + await writer.u62(BigInt(Fetch.id)); + const type: FetchType = + version === Version.DRAFT_18 + ? { + kind: "standalone", + trackNamespace: Path.from("room"), + trackName: "video", + start: { group: 0n, object: 0n }, + end: { group: 1n, object: 0n }, + } + : { + kind: "filtered", + trackNamespace: Path.from("room"), + trackName: "video", + filter: { kind: "unfiltered" }, + }; + await new Fetch({ requestId, type }).encode(writer, version); + }, + ]; try { - // A second listing of the same prefix is refused the same way, each spending its Request ID. - for (let i = 0; i < 2; i++) { + // Each is refused again on a second request, spending its own Request ID. + let requestId = 0n; + for (const request of [...requests, ...requests]) { const stream = await Stream.open(pair.client, { version }); - await stream.writer.u62(BigInt(SUBSCRIBE_TRACKS_ID)); - await Message.encode(stream.writer, async (writer) => { - await writer.u62(BigInt(i * 2)); - await Namespace.encode(writer, Path.from("prefix")); - await writer.u53(0); - }); + await request(stream.writer, requestId); + requestId += 2n; expect(await stream.reader.u53()).toBe(RequestError.id); expect((await RequestError.decode(stream.reader, version)).errorCode).toBe(0x3); expect(await stream.reader.done()).toBe(true); diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 17ac1c5a54..b478adf752 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -17,6 +17,7 @@ import * as Cluster from "./cluster.ts"; import type { RequestKind } from "./error.ts"; import { Fetch, FetchHeader } from "./fetch.ts"; import { GoAway } from "./goaway.ts"; +import * as Message from "./message.ts"; import { Group } from "./object.ts"; import { Publish } from "./publish.ts"; import { PublishNamespace } from "./publish_namespace.ts"; @@ -25,7 +26,6 @@ import { malformed, refuse } from "./request_stream.ts"; import { NativeSession } from "./session.ts"; import { Subscribe, SubscribeUpdate } from "./subscribe.ts"; import { SUBSCRIBE_TRACKS_ID, SubscribeNamespace } from "./subscribe_namespace.ts"; -import { SubscribeTracks } from "./subscribe_tracks.ts"; import { Subscriber } from "./subscriber.ts"; import { MalformedAuthToken } from "./token.ts"; import { TrackStatusRequest } from "./track.ts"; @@ -370,9 +370,19 @@ export class Connection implements Session { break; } case BigInt(SUBSCRIBE_TRACKS_ID): { - const decoded = SubscribeTracks.decode(stream.reader, this.#session.version); - const msg = await this.#request(stream, "subscribe_namespace", decoded); - if (msg) await handed(this.#publisher.runSubscribeTracks(msg, stream)); + // We list no tracks by PUBLISH, so each listing is refused on its own. Only its + // Request ID is read; the rest of the body goes unread. + const requestId = await Message.decode(stream.reader, async (body) => { + const id = await body.u62(); + await body.discard(); + return id; + }).catch((err: unknown) => { + throw malformed(err, "subscribe_tracks"); + }); + this.#spend(requestId); + await handed( + this.#refuseUnsupported(stream, "subscribe_namespace", "SUBSCRIBE_TRACKS is not supported"), + ); break; } case BigInt(TrackStatusRequest.id): { @@ -384,7 +394,8 @@ export class Connection implements Session { case BigInt(Fetch.id): { const decoded = Fetch.decode(stream.reader, this.#session.version); const msg = await this.#request(stream, "fetch", decoded); - if (msg) await handed(this.#publisher.runFetch(msg, stream)); + // Only a relay answers a FETCH, from its cache. + if (msg) await handed(this.#refuseUnsupported(stream, "fetch", "FETCH is not supported")); break; } @@ -453,6 +464,11 @@ export class Connection implements Session { this.#session.checkRequestId(requestId); } + /** Refuse a request we never serve NOT_SUPPORTED, ending its stream. */ + async #refuseUnsupported(stream: Stream, kind: RequestKind, reasonPhrase: string) { + await refuse(stream, this.#session.version, kind, { condition: "not_supported", reasonPhrase }); + } + /** Refuse a request MALFORMED_AUTH_TOKEN, ending its stream. Mirrors Rust's `token::reject`. */ async #refuseToken(stream: Stream, kind: RequestKind) { await refuse(stream, this.#session.version, kind, { diff --git a/js/net/src/ietf/fetch.test.ts b/js/net/src/ietf/fetch.test.ts index 4987e553af..9815fe07f8 100644 --- a/js/net/src/ietf/fetch.test.ts +++ b/js/net/src/ietf/fetch.test.ts @@ -1,107 +1,16 @@ import { expect, test } from "bun:test"; -import { Duration, Manual } from "@moq/clock"; -import { Signal } from "@moq/signals"; -import type { Grant } from "../auth.ts"; -import { DEFAULT_EXPIRY } from "../cache.ts"; -import { accept as acceptSession, connect as connectSession } from "../connection/index.ts"; -import { error } from "../error.ts"; -import { hooks } from "../internal.ts"; -import * as Lite from "../lite/index.ts"; -import { createMockTransportPair } from "../mock.ts"; -import { Producer as OriginProducer } from "../origin.ts"; import * as Path from "../path.ts"; -import { Reader, Stream, Writer } from "../stream.ts"; -import { Timescale, Timestamp } from "../time.ts"; -import type { Fill, 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 { Reader, Writer } from "../stream.ts"; +import { Timescale } from "../time.ts"; +import { Fetch, FetchOk, type FetchType, type Location } from "./fetch.ts"; import type * as Filter from "./filter.ts"; -import { decodeFetchObject, Group as GroupMessage } from "./object.ts"; -import { Publisher } from "./publisher.ts"; -import { RequestError } from "./request.ts"; -import { NativeSession } from "./session.ts"; -import { Subscribe, SubscribeOk } from "./subscribe.ts"; -import { ALPN, type IetfVersion, Version } from "./version.ts"; +import { type IetfVersion, Version } from "./version.ts"; -const VERSIONS = [Version.DRAFT_18, Version.DRAFT_20, Version.DRAFT_22] as const; const D18 = Version.DRAFT_18; -const D20 = Version.DRAFT_20; const D22 = Version.DRAFT_22; -const ALPNS: Record = { [D18]: ALPN.DRAFT_18, [D20]: ALPN.DRAFT_20, [D22]: ALPN.DRAFT_22 }; -const SUBSCRIBE_ID = 7n; const FETCH_ID = 9n; -const TIMEOUT = 0x2; -const NOT_SUPPORTED = 0x3; -const UNAUTHORIZED = 0x1; -const DOES_NOT_EXIST = 0x10; -const INVALID_RANGE = 0x11; -const INVALID_JOINING_REQUEST_ID = 0x32; - -/** A publisher serving the `room` broadcast, with its `video` track and the client's uni streams. */ -// `maxAge` is the track's publisher max age: omitted is the authored zero, and a test that fills -// superseded history passes `null`, since the publisher fills only groups still fresh. -function serve(version: IetfVersion, grant?: Signal, priority?: number, maxAge?: Duration | null) { - const pair = createMockTransportPair(ALPNS[version]); - const origin = new OriginProducer(); - const pub = new Publisher({ - quic: pair.server, - session: new NativeSession(pair.server, version, false), - publish: origin.consume(), - requiresSolicitation: false, - grant, - }); - const publication = origin.route(Path.from("room")); - const track = publication.broadcast.createTrack("video", { timescale: Timescale.MILLI, priority, maxAge }); - const uni = pair.client.incomingUnidirectionalStreams.getReader() as ReadableStreamDefaultReader< - ReadableStream - >; - return { - pair, - pub, - origin, - broadcast: publication.broadcast, - track, - uni, - version, - [Symbol.dispose]() { - uni.releaseLock(); - origin.close(); - }, - }; -} - -type Served = ReturnType; - -/** Write `payloads` into group `sequence`, finishing it unless told otherwise. */ -function writeGroup(track: TrackProducer, sequence: bigint, payloads: string[], finish = true) { - const group = track.createGroup(sequence); - for (const [object, payload] of payloads.entries()) { - group.writeFrame({ - payload: new TextEncoder().encode(payload), - timestamp: new Timestamp(object, Timescale.MILLI), - }); - } - if (finish) group.finish(); - return group; -} - -/** Groups `0..count`, each holding `g-0` and `g-1`, skipping `hole`. */ -function publishPairs(track: TrackProducer, count: number, hole?: number) { - for (let sequence = 0; sequence < count; sequence++) { - if (sequence !== hole) writeGroup(track, BigInt(sequence), [`${sequence}-0`, `${sequence}-1`]); - } -} - -/** The `[group, object, payload]` a run of two-object groups holds. */ -function pairs(groups: number[]): [number, number, string][] { - return groups.flatMap((group) => - [0, 1].map((object): [number, number, string] => [group, object, `${group}-${object}`]), - ); -} - /** * The range `[start, end)` as each draft spells it: a standalone FETCH through draft-18, End * Object 0 for the whole End Group, and an inclusive LOCATION_FILTER from draft-22. @@ -123,134 +32,10 @@ function filtered(filter: Filter.Filter, trackName = "video"): FetchType { return { kind: "filtered", trackNamespace: Path.from("room"), trackName, filter }; } -/** A FETCH_OK End Location naming `last` as the last object: one past it before draft-22. */ -function endAt(version: IetfVersion, group: number, last: number): Location { - return { group: BigInt(group), object: BigInt(version === D18 ? last + 1 : last) }; -} - function at(group: number, object: number): Location { return { group: BigInt(group), object: BigInt(object) }; } -/** A manual clock installed until disposed, so no wait can lapse on its own. */ -function manual(): Manual & Disposable { - const clock = new Manual(); - const installed = clock.install(); - return Object.assign(clock, { [Symbol.dispose]: () => installed.close() }); -} - -function ms(value: number): Duration { - return Duration.millis(value); -} - -/** Let every pending promise chain run. */ -async function flush() { - for (let i = 0; i < 20; i++) await new Promise((resolve) => setImmediate(resolve)); -} - -/** What a FETCH was answered with. */ -type Answer = - | { refused: number } - | { - ok: FetchOk; - /** The fetch stream's own bytes, after its type. */ - raw: Uint8Array; - requestId: bigint; - objects: [number, number, string][]; - }; - -/** A request stream the test writes the requester's side of and reads the answer from. */ -function requestStream(version: IetfVersion) { - let input!: ReadableStreamDefaultController; - const output = new TransformStream(); - const stream = new Stream({ - readable: new ReadableStream({ - start(c) { - input = c; - }, - }), - writable: output.writable, - version, - }); - return { - stream, - answer: new Reader(output.readable, undefined, version), - /** End the request as `version` cancels one: by FIN through draft 18, and by a reset after. */ - withdraw: () => (version === D18 ? input.close() : input.error(new Error("request cancelled"))), - }; -} - -/** Send `msg` and read the answer, along with the fetch stream that follows a FETCH_OK. */ -async function fetch(fx: Served, msg: Fetch): Promise { - const request = requestStream(fx.version); - const task = fx.pub.runFetch(msg, request.stream); - try { - return await answer(fx, request.answer, msg.groupOrder ?? "ascending"); - } finally { - await task; - } -} - -async function answer(fx: Served, wire: Reader, order: Order): Promise { - const id = await wire.u53(); - if (id === RequestError.id) { - const refused = (await RequestError.decode(wire, fx.version)).errorCode; - expect(await wire.done()).toBe(true); - return { refused }; - } - expect(id).toBe(FetchOk.id); - const ok = await FetchOk.decode(wire, fx.version); - expect(await wire.done()).toBe(true); - - const next = await fx.uni.read(); - if (next.done) throw new Error("no fetch stream"); - const bytes = new Uint8Array(await new Response(next.value).arrayBuffer()); - const reader = new Reader(undefined, bytes, fx.version); - expect(await reader.u53()).toBe(FetchHeader.type); - const requestId = (await FetchHeader.decode(reader, fx.version)).requestId; - return { ok, raw: bytes, requestId, objects: await readObjects(reader, order) }; -} - -/** - * Decode a fetch stream's objects as `[group, object, payload]`, its groups in `order`, and - * each End of Range as `[group, object, "end "]`. - */ -async function readObjects(reader: Reader, order: Order): Promise<[number, number, string][]> { - const objects: [number, number, string][] = []; - let group = 0; - let object = 0; - while (!(await reader.done())) { - const flags = await reader.u53(); - // Absolute ids, and nothing after them. - if (flags >= 0x80) { - group = Number(await reader.u62()); - object = Number(await reader.u62()); - objects.push([group, object, `end ${flags.toString(16)}`]); - continue; - } - if (flags & 0x08) { - const value = await reader.u53(); - group = objects.length === 0 ? value : order === "ascending" ? group + value + 1 : group - value - 1; - } - object = flags & 0x04 ? await reader.u53() : object + 1; - if (flags & 0x10) await reader.u8(); - if (flags & 0x20) await reader.read(await reader.u53()); - const payload = new TextDecoder().decode(await reader.read(await reader.u53())); - objects.push([group, object, payload]); - } - return objects; -} - -function served(answer: Answer) { - if ("refused" in answer) throw new Error(`refused with ${answer.refused}`); - return answer; -} - -function refused(answer: Answer): number { - if (!("refused" in answer)) throw new Error("answered with FETCH_OK"); - return answer.refused; -} - test("FETCH and FETCH_OK round trip on every draft", async () => { const encode = async (msg: Fetch | FetchOk, version: IetfVersion) => { const chunks: Uint8Array[] = []; @@ -289,807 +74,3 @@ test("FETCH and FETCH_OK round trip on every draft", async () => { await expect(encode(new Fetch({ requestId: 1n, type: range(D18, at(0, 0), at(0, 0)) }), D22)).rejects.toThrow(); await expect(encode(new Fetch({ requestId: 1n, type: filtered({ kind: "unfiltered" }) }), D18)).rejects.toThrow(); }); - -test("a FETCH of one whole group", async () => { - for (const version of VERSIONS) { - using fx = serve(version); - publishPairs(fx.track, 5); - const answer = served( - await fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, at(2, 0), at(2, 0)) })), - ); - // A whole-group answer before draft-22 keeps Object 0 rather than claiming the next group. - expect(answer.ok.endLocation).toEqual(version === D18 ? at(2, 0) : at(2, 1)); - expect(answer.ok.endOfTrack).toBe(false); - expect(answer.ok.properties.timescale).toBe(Timescale.MILLI); - expect(answer.requestId).toBe(FETCH_ID); - expect(answer.objects).toEqual(pairs([2])); - } -}); - -test("a FETCH stops at its end object", async () => { - for (const version of VERSIONS) { - using fx = serve(version); - publishPairs(fx.track, 5); - const answer = served( - await fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, at(2, 0), at(2, 1)) })), - ); - expect(answer.ok.endLocation).toEqual(endAt(version, 2, 0)); - expect(answer.objects).toEqual(pairs([2]).slice(0, 1)); - } -}); - -test("a FETCH spanning groups is refused before FETCH_OK", async () => { - for (const version of VERSIONS) { - for (const order of ["ascending", "descending"] as const) { - using fx = serve(version); - publishPairs(fx.track, 5, 2); - for (const end of [3, 9]) { - const type = range(version, at(0, 1), at(end, 0)); - expect(refused(await fetch(fx, new Fetch({ requestId: FETCH_ID, type, groupOrder: order })))).toBe( - NOT_SUPPORTED, - ); - } - } - } -}); - -/** The Publisher Priority each object on a fetch stream states, absent where it inherits. */ -async function priorities(raw: Uint8Array, version: IetfVersion): Promise<(number | undefined)[]> { - const reader = new Reader(undefined, raw, version); - expect(await reader.u53()).toBe(FetchHeader.type); - await FetchHeader.decode(reader, version); - const out: (number | undefined)[] = []; - while (!(await reader.done())) { - const flags = await reader.u53(); - if (flags & 0x08) await reader.u53(); - if (flags & 0x04) await reader.u53(); - out.push(flags & 0x10 ? await reader.u8() : undefined); - if (flags & 0x20) await reader.read(await reader.u53()); - await reader.read(await reader.u53()); - } - return out; -} - -// A fetch stream's first object states the track's Publisher Priority, as a subgroup header -// does, and the rest inherit it. Model priority 72 is wire 55, lower first. -test("a FETCH states the track's priority on its first object", async () => { - for (const version of VERSIONS) { - using fx = serve(version, undefined, 72); - publishPairs(fx.track, 2); - const type = range(version, at(1, 0), at(1, 0)); - const answer = served(await fetch(fx, new Fetch({ requestId: FETCH_ID, type }))); - expect(answer.objects).toEqual(pairs([1])); - expect(answer.ok.properties.priority).toBe(55); - expect(await priorities(answer.raw, version)).toEqual([55, undefined]); - } -}); - -// Only a finished track's final object ends it, wherever the range starts or ends. -test("a FETCH through the final object reports the end of track", async () => { - for (const version of VERSIONS) { - using fx = serve(version); - publishPairs(fx.track, 5); - fx.track.finish(); - - const last = served( - await fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, at(4, 1), at(4, 0)) })), - ); - expect(last.ok.endLocation).toEqual(version === D18 ? at(4, 0) : at(4, 1)); - expect(last.ok.endOfTrack).toBe(true); - expect(last.objects).toEqual(pairs([4]).slice(1)); - - // A shorter prefix must not claim the track ends there. - const prefix = served( - await fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, at(4, 0), at(4, 1)) })), - ); - expect(prefix.ok.endOfTrack).toBe(false); - } -}); - -// The Largest Object's group is read whole, so an answer reaching it waits for it to finish, -// and never claims a live track ends. -test("a FETCH waits for the Largest Object's open group", async () => { - for (const version of VERSIONS) { - using fx = serve(version); - publishPairs(fx.track, 4); - const live = writeGroup(fx.track, 4n, ["4-0"], false); - let answered = false; - const answering = fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, at(4, 0), at(4, 0)) })); - void answering.finally(() => { - answered = true; - }); - await flush(); - expect(answered).toBe(false); - live.writeFrame({ payload: new TextEncoder().encode("4-1"), timestamp: new Timestamp(1, Timescale.MILLI) }); - live.finish(); - const answer = served(await answering); - expect(answer.ok.endLocation).toEqual(version === D18 ? at(4, 0) : at(4, 1)); - expect(answer.ok.endOfTrack).toBe(false); - expect(answer.objects).toEqual(pairs([4])); - } -}); - -test("a FETCH the cache cannot answer is refused before FETCH_OK", async () => { - for (const version of VERSIONS) { - using fx = serve(version); - publishPairs(fx.track, 4, 2); - const ask = async (start: Location, end: Location) => - refused(await fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, start, end) }))); - const missing = new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, 0), "missing") }); - expect(refused(await fetch(fx, missing))).toBe(DOES_NOT_EXIST); - // A group below the newest that does not exist, and one past every group. - expect(await ask(at(2, 0), at(2, 0))).toBe(DOES_NOT_EXIST); - expect(await ask(at(5, 0), at(5, 0))).toBe(DOES_NOT_EXIST); - // A start past Largest Object in a group the cache holds (draft-22 section 9.11). - expect(await ask(at(3, 2), at(3, 6))).toBe(INVALID_RANGE); - expect(await ask(at(3, 1), at(3, 1))).toBe(INVALID_RANGE); - // A newest group with no objects starts past Largest Object too. - writeGroup(fx.track, 4n, []); - expect(await ask(at(4, 0), at(4, 4))).toBe(INVALID_RANGE); - } -}); - -test("a FETCH fills never-cached and evicted history, then reuses the cache", async () => { - for (const version of VERSIONS) { - for (const evicted of [false, true]) { - using clock = manual(); - using fx = serve(version, undefined, undefined, null); - if (evicted) { - writeGroup(fx.track, 1n, ["1-0", "1-1", "1-2"]); - await clock.advance(DEFAULT_EXPIRY.mul(2)); - } - writeGroup(fx.track, 3n, ["live"]); - const cache = fx.track.subscribe({ ranges: [] }); - expect(hooks.holdsGroup(cache, 1n)).toBe(false); - cache.close(); - using requests = fx.track.requests(); - let fill: Fill | undefined; - const requested = requests.requestedFill().then((value) => { - fill = value; - }); - const msg = new Fetch({ - requestId: FETCH_ID, - type: range(version, at(1, 1), at(1, 3)), - subscriberPriority: 3, - groupOrder: "ascending", - }); - const fetching = fetch(fx, msg); - await flush(); - expect(fill).toBeDefined(); - if (!fill) throw new Error("no history fill"); - expect([fill.start, fill.end, fill.priority, fill.maxDelay]).toEqual([1n, 2n, 124, null]); - const group = fill.accept(1n); - for (let object = 0; object < 3; object++) - group.writeFrame({ - payload: new TextEncoder().encode(`1-${object}`), - timestamp: Timestamp.fromMillis(object), - }); - group.finish(); - fill.close(); - await requested; - const first = served(await fetching); - expect(first.objects).toEqual([ - [1, 1, "1-1"], - [1, 2, "1-2"], - ]); - expect(first.ok.endLocation).toEqual(endAt(version, 1, 2)); - expect(first.ok.endOfTrack).toBe(false); - // No handler remains: the same responder serves the filled group from cache. - requests.close(); - expect(served(await fetch(fx, msg)).objects).toEqual(first.objects); - } - } -}); - -test("a FETCH preserves relayed metadata and cached groups without upstream FETCH", async () => { - for (const version of VERSIONS) { - for (const warm of [false, true]) { - using fx = serve(version); - const upstream = new OriginProducer(); - const publication = upstream.route(Path.from("archive")); - const track = publication.broadcast.createTrack("video", { timescale: Timescale.MILLI }); - writeGroup(track, 3n, ["3-0", "3-1"]); - // lite-04 has no FETCH; lite-05 can fetch, but refuses empty subscription ranges. - const pair = createMockTransportPair(warm ? Lite.ALPN_04 : Lite.ALPN_05); - const url = new URL("https://localhost:4443/test"); - const [client, server] = await Promise.all([ - connectSession({ url, transport: pair.client }), - acceptSession({ url, transport: pair.server, publish: upstream.consume() }), - ]); - try { - const source = wireOf(client).consume(Path.from("archive")); - const cached = warm ? source.track("video").subscribe({ delay: { max: null } }) : undefined; - if (cached) { - const group = await cached.recvGroup(); - expect(group?.sequence).toBe(3n); - while (await group?.readFrame()) {} - group?.close(); - } - const requests = fx.origin.claim(Path.from("relayed")); - const cursor = requests.requested(); - const requested = cursor.next(); - const type = { ...range(version, at(3, 0), at(3, 0)), trackNamespace: Path.from("relayed") }; - const fetching = fetch(fx, new Fetch({ requestId: FETCH_ID, type })); - const next = await requested; - if (next.done) throw new Error("no relayed broadcast request"); - next.value.serve(source); - await cursor.return?.(); - expect(served(await fetching).objects).toEqual(pairs([3])); - cached?.close(); - } finally { - client.abort(); - server.abort(); - upstream.close(); - } - } - } -}); - -test("a FETCH refused by its history responder sends no FETCH_OK", async () => { - for (const version of VERSIONS) { - using fx = serve(version, undefined, undefined, null); - writeGroup(fx.track, 3n, ["live"]); - using requests = fx.track.requests(); - const responding = (async () => { - using fill = await requests.requestedFill(); - expect(fill?.wanted()).toBe(1n); - })(); - const fetching = fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, at(1, 0), at(1, 0)) })); - expect(refused(await fetching)).toBe(DOES_NOT_EXIST); - requests.close(); - await responding; - } -}); - -test("a cancelled or unauthorized FETCH withdraws its unanswered history fill", async () => { - for (const version of VERSIONS) { - for (const revoke of [false, true]) { - const grant = new Signal(undefined); - using fx = serve(version, grant, undefined, null); - writeGroup(fx.track, 3n, ["live"]); - using requests = fx.track.requests(); - const requested = requests.requestedFill(); - const request = requestStream(version); - const fetching = fx.pub.runFetch( - new Fetch({ requestId: FETCH_ID, type: range(version, at(1, 0), at(1, 0)) }), - request.stream, - ); - using fill = await requested; - expect(fill?.wanted()).toBe(1n); - if (revoke) grant.set({ publish: new Path.Patterns([]), subscribe: new Path.Patterns([]) }); - else request.withdraw(); - if (revoke) expect(refused(await answer(fx, request.answer, "ascending"))).toBe(UNAUTHORIZED); - await fetching; - expect(await fill?.wantedChanged()).toBeUndefined(); - const read = await request.answer.done().catch(error); - expect(read === true || read instanceof Error).toBe(true); - } - } -}); - -// A filter other than a bounded absolute one resolves against Largest Object. -test("draft-22: a FETCH filter resolves against Largest Object", async () => { - const cases: [Filter.Filter, number[] | number][] = [ - [{ kind: "unfiltered" }, NOT_SUPPORTED], - [{ kind: "nextObject" }, INVALID_RANGE], - [{ kind: "relative", groups: 2n }, NOT_SUPPORTED], - [{ kind: "relative", groups: 1n }, [2]], - [{ kind: "absolute", startGroup: 2n, startObject: 0n }, [2]], - ]; - for (const [filter, expected] of cases) { - using fx = serve(D22); - publishPairs(fx.track, 3); - const answer = await fetch(fx, new Fetch({ requestId: FETCH_ID, type: filtered(filter) })); - if (typeof expected === "number") { - expect(refused(answer)).toBe(expected); - continue; - } - expect(served(answer).ok.endLocation).toEqual(at(2, 1)); - expect(served(answer).objects).toEqual(pairs(expected)); - } - - // An empty track has no Largest Object to resolve against. - using fx = serve(D22); - expect(refused(await fetch(fx, new Fetch({ requestId: FETCH_ID, type: filtered({ kind: "unfiltered" }) })))).toBe( - NOT_SUPPORTED, - ); -}); - -test("a FETCH outside the grant is refused", async () => { - const nothing = (): Grant => ({ publish: new Path.Patterns([]), subscribe: new Path.Patterns([]) }); - for (const version of VERSIONS) { - using fx = serve(version, new Signal(nothing())); - publishPairs(fx.track, 2); - const answer = await fetch(fx, new Fetch({ requestId: FETCH_ID, type: range(version, at(0, 0), at(0, 0)) })); - expect(refused(answer)).toBe(UNAUTHORIZED); - } -}); - -// The requester leaving while the answer waits on an open group is owed nothing, budget or not. -test("a cancelled FETCH stops waiting", async () => { - for (const version of VERSIONS) { - for (const fillTimeout of [undefined, 10_000]) { - using _clock = manual(); - using fx = serve(version); - publishPairs(fx.track, 2); - writeGroup(fx.track, 2n, ["2-0"], false); - const request = requestStream(version); - const task = fx.pub.runFetch( - new Fetch({ requestId: FETCH_ID, type: range(version, at(2, 0), at(2, 0)), fillTimeout }), - request.stream, - ); - await flush(); - request.withdraw(); - await task; - const read = await request.answer.done().catch(error); - expect(read === true || read instanceof Error).toBe(true); - } - } -}); - -// FILL_TIMEOUT is accepted without truncating a one-group answer, including zero. -test("FILL_TIMEOUT is ignored while a group finishes", async () => { - for (const version of VERSIONS) { - for (const fillTimeout of [0, 100]) { - using clock = manual(); - using fx = serve(version); - const live = writeGroup(fx.track, 4n, ["4-0"], false); - let answered = false; - const answering = fetch( - fx, - new Fetch({ requestId: FETCH_ID, type: range(version, at(4, 0), at(4, 0)), fillTimeout }), - ); - void answering.then(() => { - answered = true; - }); - await clock.advance(ms(1000)); - expect(answered).toBe(false); - live.writeFrame({ payload: new TextEncoder().encode("4-1"), timestamp: new Timestamp(1, Timescale.MILLI) }); - live.finish(); - expect(served(await answering).objects).toEqual(pairs([4])); - } - } -}); - -/** Start a draft-18 Largest Object SUBSCRIBE of `room/video`, or as `fields` override, without reading its answer. */ -async function startSubscribe(fx: Served, fields: Partial[0]> = {}) { - const client = await Stream.open(fx.pair.client, { version: fx.version }); - const server = await Stream.accept(fx.pair.server, fx.version); - if (!server) throw new Error("publisher never accepted the subscribe stream"); - const msg = new Subscribe({ - requestId: SUBSCRIBE_ID, - trackNamespace: Path.from("room"), - trackName: "video", - subscriberPriority: 0, - filter: { kind: "nextObject" }, - ...fields, - }); - return { client, serving: fx.pub.runSubscribe(msg, server) }; -} - -/** Open a draft-18 SUBSCRIBE of `room/video` with `filter`, returning its SUBSCRIBE_OK. */ -async function subscribe(fx: Served, filter: Filter.Filter = { kind: "nextObject" }) { - const { client, serving } = await startSubscribe(fx, { filter }); - expect(await client.reader.u53()).toBe(SubscribeOk.id); - const ok = await SubscribeOk.decode(client.reader, fx.version); - return { client, ok, serving }; -} - -function joining(offset = 0n): Fetch { - return new Fetch({ requestId: FETCH_ID, type: { kind: "relativeJoining", subscription: SUBSCRIBE_ID, offset } }); -} - -// A Largest Object subscription starts right after the Largest Object SUBSCRIBE_OK names, -// and a joining FETCH answers the group's head up to it: together they cover the group with no -// gap or overlap, which is how the Rust subscriber joins mid-group on draft-18. -test("draft-18: a joining FETCH serves the head its subscription starts after", async () => { - using fx = serve(D18); - publishPairs(fx.track, 5); - const live = writeGroup(fx.track, 5n, ["5-0", "", "5-2"], false); - const { client, ok } = await subscribe(fx); - expect(ok.largest).toEqual({ groupId: 5n, objectId: 2n }); - - // Written after SUBSCRIBE_OK, so it belongs to the subscription, not the head. - live.writeFrame({ payload: new TextEncoder().encode("5-3"), timestamp: new Timestamp(3, Timescale.MILLI) }); - - const request = requestStream(D18); - const task = fx.pub.runFetch(joining(), request.stream); - // The subscription's stream for group 5 opened first, starting at the object after the head. - const next = await fx.uni.read(); - if (next.done) throw new Error("no subscription stream"); - const group = new Reader(next.value, undefined, D18); - const header = await GroupMessage.decode(group, D18); - expect(header.groupId).toBe(5n); - expect(header.flags.firstObject).toBe(false); - expect(await group.u53()).toBe(3); - - const head = served(await answer(fx, request.answer, "ascending")); - await task; - expect(head.ok).toEqual( - new FetchOk({ - endOfTrack: false, - endLocation: at(5, 3), - // The max age is never advertised as MAX_CACHE_DURATION. - properties: { timescale: Timescale.MILLI, priority: 127, groupOrder: 0x2 }, - }), - ); - expect(head.requestId).toBe(FETCH_ID); - expect(head.objects).toEqual([ - [5, 0, "5-0"], - [5, 1, ""], - [5, 2, "5-2"], - ]); - // The first object spells its group, object and priority, and the rest inherit them. - const reader = new Reader(undefined, head.raw, D18); - await reader.u53(); - await FetchHeader.decode(reader, D18); - expect(await reader.u53()).toBe(0x3c); - // The priority is the track's, as FETCH_OK and every subgroup header state it. - expect([await reader.u53(), await reader.u53(), await reader.u8()]).toEqual([5, 0, 127]); - await reader.read(await reader.u53()); - await reader.read(await reader.u53()); - expect(await reader.u53()).toBe(0x20); - client.close(); -}); - -// A subscription can end while its head is in flight, as when a track finishes right after -// SUBSCRIBE_OK, and the head is then all the subscriber gets of that group. -test("draft-18: a joining FETCH outlives a subscription that ends after FETCH_OK", async () => { - using fx = serve(D18); - const live = writeGroup(fx.track, 5n, ["5-0", "5-1"], false); - const { client, serving } = await subscribe(fx); - - const request = requestStream(D18); - const task = fx.pub.runFetch(joining(), request.stream); - expect(await request.answer.u53()).toBe(FetchOk.id); - client.close(); - await serving; - live.finish(); - - // The subscription's own stream for group 5 opened first, and its end reset it. - const subscription = await fx.uni.read(); - if (subscription.done) throw new Error("no subscription stream"); - const next = await fx.uni.read(); - if (next.done) throw new Error("no fetch stream"); - const reader = new Reader(next.value, undefined, D18); - expect(await reader.u53()).toBe(FetchHeader.type); - await FetchHeader.decode(reader, D18); - expect(await readObjects(reader, "ascending")).toEqual(pairs([5])); - await task; -}); - -test("draft-18: a joining FETCH its subscription cannot answer is refused", async () => { - { - using fx = serve(D18); - publishPairs(fx.track, 5); - // A subscription that never arrives, once the FETCH has waited as long as a request may. - const clock = new Manual(); - const installed = clock.install(); - try { - expect(refused(await clock.drive(fetch(fx, joining())))).toBe(INVALID_JOINING_REQUEST_ID); - } finally { - installed.close(); - } - - const { client, serving } = await subscribe(fx); - // Only the subscription's own group, never a run of groups before it. - expect(refused(await fetch(fx, joining(1n)))).toBe(NOT_SUPPORTED); - const absolute = new Fetch({ - requestId: FETCH_ID, - type: { kind: "absoluteJoining", subscription: SUBSCRIBE_ID, group: 5n }, - }); - expect(refused(await fetch(fx, absolute))).toBe(INVALID_RANGE); - - // A FETCH arriving after its subscription ended has no edge to end at. - client.close(); - await serving; - const later = new Manual(); - const reinstalled = later.install(); - try { - expect(refused(await later.drive(fetch(fx, joining())))).toBe(INVALID_JOINING_REQUEST_ID); - } finally { - reinstalled.close(); - } - } - { - // A subscription that does not start after Largest Object has no head to join. - using fx = serve(D18); - publishPairs(fx.track, 5); - const { client } = await subscribe(fx, { kind: "unfiltered" }); - expect(refused(await fetch(fx, joining()))).toBe(NOT_SUPPORTED); - client.close(); - } - { - // One that started on an empty track has no head at all. - using fx = serve(D18); - const { client } = await subscribe(fx); - expect(refused(await fetch(fx, joining()))).toBe(INVALID_RANGE); - client.close(); - } - { - // A head the cache lost is refused, rather than answered short of where the - // subscription starts. - using fx = serve(D18); - const live = writeGroup(fx.track, 5n, ["5-0", "5-1"], false); - const { client } = await subscribe(fx); - live.abort(new Error("evicted")); - expect(refused(await fetch(fx, joining()))).toBe(DOES_NOT_EXIST); - client.close(); - } - { - // An upstream can report the largest safe frame, making the head end one past it. No - // model group holds that many frames, so the head is unavailable rather than an error. - using fx = serve(D18); - writeGroup(fx.track, 5n, ["5-0", "5-1"], false); - hooks.reportedLargest(fx.track, { group: 5n, frame: Number.MAX_SAFE_INTEGER }); - const { client, ok } = await subscribe(fx); - expect(ok.largest).toEqual({ groupId: 5n, objectId: BigInt(Number.MAX_SAFE_INTEGER) }); - expect(refused(await fetch(fx, joining()))).toBe(DOES_NOT_EXIST); - client.close(); - } -}); - -// Request streams race, so a joining FETCH can be read before the SUBSCRIBE it names, as when -// the subscriber pipelines both: it waits for the subscription rather than being refused. -test("draft-18: a joining FETCH waits for the subscription it names", async () => { - using fx = serve(D18); - writeGroup(fx.track, 5n, ["5-0", "5-1"], false); - const request = requestStream(D18); - const task = fx.pub.runFetch(joining(), request.stream); - await flush(); - const { client } = await subscribe(fx); - - expect(await request.answer.u53()).toBe(FetchOk.id); - expect((await FetchOk.decode(request.answer, D18)).endLocation).toEqual(at(5, 2)); - // The subscription's own stream for group 5 may open on either side of the fetch stream. - for (;;) { - const next = await fx.uni.read(); - if (next.done) throw new Error("no fetch stream"); - const reader = new Reader(next.value, undefined, D18); - if ((await reader.u53()) !== FetchHeader.type) continue; - await FetchHeader.decode(reader, D18); - expect(await readObjects(reader, "ascending")).toEqual(pairs([5])); - break; - } - await task; - client.close(); -}); - -// A subscription still being set up when the FETCH's request budget runs out is refused TIMEOUT, -// rather than as one the publisher never saw. -test("draft-18: a joining FETCH times out on a subscription still being set up", async () => { - using fx = serve(D18); - // Serve tracks on demand and never answer the request, so the SUBSCRIBE never resolves. - const requested = wireOf(fx.broadcast).requested(); - const { client, serving } = await startSubscribe(fx, { trackName: "audio" }); - expect((await requested)?.name).toBe("audio"); - - const clock = new Manual(); - const installed = clock.install(); - try { - expect(refused(await clock.drive(fetch(fx, joining())))).toBe(TIMEOUT); - } finally { - installed.close(); - client.close(); - await serving; - } -}); - -// A subscription refused while a joining FETCH waits on it answers the FETCH at once. -test("draft-18: a joining FETCH whose subscription is refused is refused", async () => { - using fx = serve(D18); - publishPairs(fx.track, 5); - const request = requestStream(D18); - const task = fx.pub.runFetch(joining(), request.stream); - await flush(); - - const { client, serving } = await startSubscribe(fx, { rangeFilters: true }); - expect(await client.reader.u53()).toBe(RequestError.id); - await serving; - - expect(refused(await answer(fx, request.answer, "ascending"))).toBe(INVALID_JOINING_REQUEST_ID); - 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/malformed.test.ts b/js/net/src/ietf/malformed.test.ts index de11d7d619..338052db69 100644 --- a/js/net/src/ietf/malformed.test.ts +++ b/js/net/src/ietf/malformed.test.ts @@ -14,7 +14,7 @@ import { RequestOk } from "./request.ts"; import { Setup } from "./setup.ts"; import { SetupOption, SetupOptions } from "./setup_options.ts"; import { SubscribeOk } from "./subscribe.ts"; -import { SUBSCRIBE_TRACKS_ID, SubscribeNamespace } from "./subscribe_namespace.ts"; +import { SubscribeNamespace } from "./subscribe_namespace.ts"; import { ALPN, type IetfVersion, Version } from "./version.ts"; // Malformed control input the draft says ends the session must end it, with the draft's code. @@ -112,26 +112,6 @@ for (const [version, alpn] of DRAFTS) { ["a namespace over 4096 bytes", subscribe([field(4097)], 1), SessionCode.ProtocolViolation], ["a full track name over 4096 bytes", subscribe([field(4000)], 97), SessionCode.ProtocolViolation], ["a trailing byte", frame(0x03, [0, 1, ...field(4), ...field(5), 0, 0]), SessionCode.ProtocolViolation], - [ - "an invalid SUBSCRIBE_TRACKS FORWARD", - frame(SUBSCRIBE_TRACKS_ID, [0, 0, 1, 0x10, 2]), - SessionCode.ProtocolViolation, - ], - [ - "a duplicate SUBSCRIBE_TRACKS FORWARD", - frame(SUBSCRIBE_TRACKS_ID, [0, 0, 2, 0x10, 0, 0, 1]), - SessionCode.ProtocolViolation, - ], - [ - "a SUBSCRIBE_TRACKS prefix over 32 fields", - frame(SUBSCRIBE_TRACKS_ID, [0, 33, ...new Array(33).fill(field(1)).flat(), 0]), - SessionCode.ProtocolViolation, - ], - [ - "a SUBSCRIBE_TRACKS prefix over 4096 bytes", - frame(SUBSCRIBE_TRACKS_ID, [0, 1, ...field(4097), 0]), - SessionCode.ProtocolViolation, - ], ] as const) { test(`a request with ${name} closes ${alpn}`, async () => { using clock = manual(); diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 13f69a5970..3d9db33537 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -28,7 +28,7 @@ import { wireOf } from "../wire.ts"; import type * as Cluster from "./cluster.ts"; import { ObjectDatagram } from "./datagram.ts"; import { SubscribeRefused, toRequestCode } from "./error.ts"; -import { Fetch, FetchHeader } from "./fetch.ts"; +import { FetchHeader } from "./fetch.ts"; import * as Filter from "./filter.ts"; import { Frame, Group as GroupMessage } from "./object.ts"; import { Parameters } from "./parameters.ts"; @@ -682,21 +682,18 @@ test("Range Filters over MAX_FILTER_RANGES zero get INVALID_FILTER", async () => for (const version of [Version.DRAFT_22]) { const pair = createMockTransportPair(ALPNS[version]); const { pub, origin } = publisher(pair.server, { session: new NativeSession(pair.server, version, true) }); - for (const fetch of [false, true]) { - const request = requestStream(version); - try { - const name = [1, 4, ...new TextEncoder().encode("test"), 5, ...new TextEncoder().encode("video")]; - const fields = [7, ...name, 1, 0x25, 0]; - const encoded = new Reader(undefined, Uint8Array.from([0, fields.length, ...fields]), version); - if (fetch) await pub.runFetch(await Fetch.decode(encoded, version), request.stream); - else await pub.runSubscribe(await Subscribe.decode(encoded, version), request.stream); - await request.stream.writer.closed; - const wire = request.wire(); - expect(await wire.u53()).toBe(RequestError.id); - expect((await RequestError.decode(wire, version)).errorCode).toBe(0x36); - } finally { - request.cancel(); - } + const request = requestStream(version); + try { + const name = [1, 4, ...new TextEncoder().encode("test"), 5, ...new TextEncoder().encode("video")]; + const fields = [7, ...name, 1, 0x25, 0]; + const encoded = new Reader(undefined, Uint8Array.from([0, fields.length, ...fields]), version); + await pub.runSubscribe(await Subscribe.decode(encoded, version), request.stream); + await request.stream.writer.closed; + const wire = request.wire(); + expect(await wire.u53()).toBe(RequestError.id); + expect((await RequestError.decode(wire, version)).errorCode).toBe(0x36); + } finally { + request.cancel(); } origin.close(); } @@ -2851,8 +2848,8 @@ async function subscribeOkLargest(version: IetfVersion): Promise { for (const version of IETF_VERSIONS) { diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 6abbacca42..cc03ccdba5 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -22,15 +22,7 @@ import { } from "../error.ts"; import type * as group from "../group.ts"; import type { Hop, WireRoute } from "../hop.ts"; -import { - datagramTrackInfo, - finalSequence, - firstEligible, - hiddenBelow, - hooks, - presented, - type ReadGroupFrame, -} from "../internal.ts"; +import { datagramTrackInfo, 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"; @@ -51,14 +43,14 @@ import { toNumber } from "../util/bigint.ts"; import { nextChange } from "../util/change.ts"; import { Deadline, REQUEST_TIMEOUT_MS } from "../util/deadline.ts"; import { GivenUp } from "../util/given_up.ts"; -import { max, min, saturatingSub } from "../util/sequence.ts"; +import { max, saturatingSub } from "../util/sequence.ts"; import { TimeoutError, withTimeout } from "../util/timeout.ts"; import * as Varint from "../varint.ts"; import { type Advertised, type Advertisements, sameInstance, wireOf } from "../wire.ts"; import * as Cluster from "./cluster.ts"; import { encodeDatagram } from "./datagram.ts"; import { type RequestCondition, requestReason, SubscribeRefused } from "./error.ts"; -import { type Fetch, FetchHeader, FetchOk, type FetchType } from "./fetch.ts"; +import { FetchHeader } from "./fetch.ts"; import * as Filter from "./filter.ts"; import { MAX_AGE_PARAMETER } from "./max_age.ts"; import * as Message from "./message.ts"; @@ -86,7 +78,6 @@ import { SubscribeNamespaceEntry, SubscribeNamespaceEntryDone, } from "./subscribe_namespace.ts"; -import type { SubscribeTracks } from "./subscribe_tracks.ts"; import { MalformedAuthToken } from "./token.ts"; import { TrackStatusOk, type TrackStatusRequest } from "./track.ts"; import type { IetfVersion } from "./version.ts"; @@ -155,9 +146,6 @@ function sameHops(a: Hop[], b: Hop[]): boolean { const ADVERTISE_TIMEOUT_MS = REQUEST_TIMEOUT_MS; const ADVERTISE_TIMEOUT = Duration.millis(ADVERTISE_TIMEOUT_MS); -/** How long a joining FETCH waits for the subscription it names to arrive and resolve. */ -const JOIN_TIMEOUT = Duration.millis(REQUEST_TIMEOUT_MS); - /** Sleep `delay`, jittered, so a relay's namespaces don't all retry on the same tick. */ function retryAfter(delay: number): Promise { return sleep(Duration.millis(delay * (0.5 + Math.random() / 2))); @@ -370,12 +358,6 @@ export class Publisher { #datagramWriter?: ReturnType; #session: Session; #requiresSolicitation: boolean; - // What a joining FETCH needs from each draft-18 SUBSCRIBE, by its request ID, while it lasts: - // pending from its arrival until it resolves. Request streams race, so a joining FETCH can - // arrive before the subscription it names has. - #joins = new Map(); - // Bumped on every change to #joins, which a waiting joining FETCH rechecks. - #joinsChanged = new Signal(0); #hidden: boolean; // Whether a SUBSCRIBE_NAMESPACE's REQUEST_OK counts the NAMESPACEs that follow it (MoQ Active Count). @@ -475,19 +457,11 @@ export class Publisher { } async #runSubscribe(msg: Subscribe, request: RequestStream) { - // Draft-22 replaced joining FETCH with fills. - const joinable = !Filter.isDraft20(this.#session.version); - if (joinable) this.#setJoin(msg.requestId, "pending"); - try { - if (Namespace.reserved(msg.trackNamespace, this.#session.version)) { - await refuseReserved(request); - return; - } - await this.#serveSubscribe(msg, request); - } finally { - // A joining FETCH arriving after the subscription ended has no edge to end at. - if (joinable) this.#setJoin(msg.requestId, undefined); + if (Namespace.reserved(msg.trackNamespace, this.#session.version)) { + await refuseReserved(request); + return; } + await this.#serveSubscribe(msg, request); } async #serveSubscribe(msg: Subscribe, request: RequestStream) { @@ -535,7 +509,7 @@ export class Publisher { const subscribe = (source: broadcast.Consumer) => { // A range that resolves without the live edge has a fixed start, which is no join: // bound the subscription before subscribing so the refresh ratio never judges it. - const fixed = filterRange(msg.filter, {}); + const fixed = filterRange(msg.filter, {}, this.#session.version); // Parked past a bounded window's end: the peer reads this stream's end as the track's. return hooks.parked( source.track(msg.trackName).subscribe({ @@ -802,7 +776,7 @@ export class Publisher { msg.fill && msg.forward && Filter.isDraft20(version) ? fillRange(msg.fill, msg.filter, edge.largest) : undefined; - const requested = filterRange(msg.filter, edge); + const requested = filterRange(msg.filter, edge, version); // A copy, since an update moves the cursor in place and must not move the bounds too. const range = { ...afterFill(requested, fill, edge) }; // What the subscriber asked for, which only a narrowing changes: the fill's split @@ -838,26 +812,6 @@ export class Publisher { const ranked = new Priority(track, info.priority); ranking = ranked; const properties = trackProperties(info, msg.propertiesWanted); - // Draft-22 replaced joining FETCH with fills. Older drafts save the Largest Object the - // subscription starts after, again on every resume, so a joining FETCH ends exactly - // where the subscription begins. Saved before the answer names it. - if (!Filter.isDraft20(version)) { - this.#setJoin( - msg.requestId, - msg.filter.kind === "nextObject" - ? { - kind: "join", - track, - largest: edge.largest, - namespace: name, - properties, - timescale, - publisherPriority, - ranking: ranked, - } - : { kind: "unsupported" }, - ); - } if (answer) { // The resuming update learns where the subscription now joins. @@ -869,8 +823,8 @@ export class Publisher { const ok = new SubscribeOk({ trackAlias: msg.requestId, // Required once the track has content. A Next Object subscription starts right - // after it, and recovers the group's head with a FILL on draft-22 or a joining - // FETCH before it, both of which end here. + // after it and recovers the group's head with a FILL from draft 20, which ends + // here. On draft 18 it carries the group whole from object 0 instead. largest: edge.largest && { groupId: edge.largest.group, objectId: edge.largest.object }, properties, }); @@ -1002,7 +956,7 @@ export class Publisher { ? bounds.peek() : msg.filter.kind === "unfiltered" ? { start: { group: 0n, object: 0n } } - : filterRange(msg.filter, snapshot); + : filterRange(msg.filter, snapshot, version); const tail = afterFill(requested, selected, snapshot); range.start = tail.start; range.end = tail.end; @@ -2320,19 +2274,6 @@ export class Publisher { } } - /** Refuse a request to discover tracks under a namespace prefix: we list none by PUBLISH. @internal */ - runSubscribeTracks(msg: SubscribeTracks, stream: Stream): Promise { - const request = new RequestStream(stream, this.#session, this.#quic); - return request.serve( - Namespace.reserved(msg.prefix, this.#session.version) - ? refuseReserved(request) - : request.refuse("subscribe_namespace", { - condition: "not_supported", - reasonPhrase: "SUBSCRIBE_TRACKS is not supported", - }), - ); - } - async #subscriptionUpdate(framed: Framed): Promise<{ params: Parameters; requestId: bigint }> { const version = this.#session.version; const update = await Message.decode(framed.reader, async (body) => { @@ -2379,388 +2320,6 @@ export class Publisher { } } - /** - * Handles an incoming FETCH on a bidi stream: a range of the named track, or a joining - * FETCH's head of its subscription's group. A named range missing from the cache asks - * the producer's request handler; a joining head must already be cached. - * - * @internal - */ - runFetch(msg: Fetch, stream: Stream): Promise { - const request = new RequestStream(stream, this.#session, this.#quic); - return request.serve(this.#runFetch(msg, request)); - } - - async #runFetch(msg: Fetch, request: RequestStream) { - // FILL_TIMEOUT does not truncate a group or time out its application fill. Both remain - // wanted until answered, refused, or cancelled by the requester or its grant. - const version = this.#session.version; - const type = msg.type; - const named = type.kind === "standalone" || type.kind === "filtered" ? type : undefined; - const joining = type.kind === "relativeJoining" || type.kind === "absoluteJoining" ? type : undefined; - if (named && Namespace.reserved(named.trackNamespace, version)) { - await refuseReserved(request); - return; - } - // Serving a Range Filter unfiltered would deliver objects the requester excluded. - if (msg.rangeFilters) { - await request.refuse("fetch", { - condition: "invalid_filter", - reasonPhrase: "Range Filters exceed MAX_FILTER_RANGES=0", - }); - return; - } - // A joining FETCH was authorized with its subscription; a standalone one is checked - // here, so a denied request never reaches the origin. - const denied: Refusal = { condition: "unauthorized", reasonPhrase: "not granted" }; - if (named && !this.#gate.allows(named.trackNamespace)) { - await request.refuse("fetch", denied); - return; - } - - // Everything that ends the answer early: the requester leaving, our grant narrowing, - // or the subscription a joining FETCH named ending before it resolved. - const stop = new AbortController(); - const cancel = new StreamError(StreamCode.Cancel, { message: "cancel" }); - void request.ended.then(() => stop.abort(cancel)); - // Nothing may follow the request. - void request.next().then( - (framed) => framed && request.abort(new ProtocolViolation("unexpected message on a FETCH")), - // The request ended, which the stop already heard. - () => undefined, - ); - - let join: Joinable | undefined; - if (joining) { - let found: Joinable | Refusal; - try { - found = await this.#joinOf(joining.subscription, stop.signal); - } catch (err: unknown) { - if (stop.signal.reason !== cancel) throw err; - request.abort(cancel); - return; - } - if ("condition" in found) { - await request.refuse("fetch", found); - return; - } - join = found; - } - const namespace = named?.trackNamespace ?? (join?.kind === "join" ? join.namespace : undefined); - // 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 = - joining && - this.#joinsChanged.subscribe(() => { - if (!resolved && !this.#joins.has(joining.subscription)) stop.abort(lapsed); - }); - const watch = namespace && this.#gate.watch(namespace, () => stop.abort(unauthorized(namespace))); - if (namespace && watch?.revoked()) stop.abort(unauthorized(namespace)); - - let answer: FetchAnswer | Refusal | undefined; - try { - try { - if (named) answer = await this.#fetchRange(msg, named, 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) { - request.abort(cancel); - return; - } - answer = - stop.signal.reason === lapsed - ? { condition: "invalid_joining_request_id", reasonPhrase: "subscription ended" } - : stop.signal.aborted && namespace && !this.#gate.allows(namespace) - ? denied - : this.#failed(err); - } - if ("condition" in answer) { - await request.refuse("fetch", answer); - return; - } - resolved = true; - await this.#answerFetch(msg, request, answer, stop.signal); - } finally { - lapse?.(); - watch?.dispose(); - stop.abort(); - if (answer && "owned" in answer && answer.owned) answer.cache.close(); - } - } - - /** - * Resolve a standalone or draft-22 FETCH to its group reader and end metadata, filling - * uncached history through the producer's request handler, 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. - */ - async #fetchRange( - msg: Fetch, - named: Extract, - signal: AbortSignal, - ): Promise { - const version = this.#session.version; - // Every filter but a bounded absolute one ends at the Largest Object, which resolves - // once the track does. - let relative: Filter.Filter | undefined; - let requested: FetchRange; - if (named.kind === "standalone") { - requested = { start: named.start, end: named.end }; - } else if (named.filter.kind === "absolute" && named.filter.endGroup !== undefined) { - // The filter's End Object is inclusive, where a standalone one is the last object - // plus one. Without one, both include the whole End Group. - const { startGroup, startObject, endGroup, endObject } = named.filter; - requested = { - start: { group: startGroup, object: startObject }, - end: { group: endGroup, object: endObject === undefined ? 0n : endObject + 1n }, - }; - } else { - relative = named.filter; - // Placeholders until the filter resolves against the Largest Object. - requested = { start: { group: 0n, object: 0n }, end: { group: 0n, object: 0n } }; - } - // End Object 0 includes the entire End Group. - const { start, end } = requested; - if ( - !relative && - (start.group > end.group || (start.group === end.group && end.object !== 0n && start.object >= end.object)) - ) { - return { condition: "invalid_range", reasonPhrase: "empty range" }; - } - - if (!relative && start.group !== end.group) - return { condition: "not_supported", reasonPhrase: "FETCH spans more than one group" }; - - const broadcast = await this.#resolve(named.trackNamespace, signal); - const priority = fromWire(msg.subscriberPriority); - const track = broadcast.track(named.trackName); - const cache = track.subscribe({ priority, delay: { max: Duration.millis(Varint.MAX_U53), min: null } }); - let answer: FetchAnswer | undefined; - try { - const info = await untilAborted(cache.info(), signal); - const position = hooks.objectLargest(cache); - const largest = position && { group: position.group, object: BigInt(position.frame) }; - const range = { ...requested }; - if (relative) { - if (!largest) { - return { - condition: "not_supported", - reasonPhrase: "FETCH relative to Largest Object needs a known publisher edge", - }; - } - range.start = fetchStart(relative, largest); - range.end = { group: largest.group, object: 0n }; - } - if (range.start.group !== range.end.group) - return { condition: "not_supported", reasonPhrase: "FETCH spans more than one group" }; - - if (largest) { - // A start past Largest Object in a group the cache holds is refused INVALID_RANGE - // (draft-22 section 9.11), rather than waiting on the live group. A group past - // every one held does not exist yet. - if (after(range.start, largest)) { - return hooks.holdsGroup(cache, range.start.group) - ? { condition: "invalid_range", reasonPhrase: "start past Largest Object" } - : { condition: "does_not_exist", reasonPhrase: `group ${range.start.group}` }; - } - } - - const selected = range.start.group; - // A relayed track may hold the group even when its upstream cannot FETCH. Only a - // cache miss asks the producer's handler or fetches the group from its serving route. - const group = - hooks.cachedGroup(cache, selected) ?? (await track.fetchGroup(selected, { priority, signal })); - 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 { - 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; - } - - 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(); - } - } finally { - if (!answer) cache.close(); - } - } - - /** Record what a joining FETCH finds for subscription `id`, or remove it when absent. */ - #setJoin(id: bigint, join: Joinable | "pending" | undefined) { - if (join) this.#joins.set(id, join); - else this.#joins.delete(id); - this.#joinsChanged.update((n) => n + 1); - } - - /** - * Wait for the subscription a joining FETCH names to resolve, or the refusal to answer - * with. One not yet arrived or still resolving is waited on, bounded like any request, - * since the FETCH's stream can beat its SUBSCRIBE's. - */ - async #joinOf(id: bigint, signal: AbortSignal): Promise { - const missing: Refusal = { condition: "invalid_joining_request_id", reasonPhrase: "no such subscription" }; - const done = new AbortController(); - const expired = sleep(JOIN_TIMEOUT, { signal: done.signal }).then( - () => "expired" as const, - () => "expired" as const, - ); - let pending = false; - try { - for (;;) { - const join = this.#joins.get(id); - if (join === "pending") pending = true; - else if (join) return join; - // Seen pending and now gone: it ended, or was refused, before it resolved. - else if (pending) return missing; - if ((await untilAborted(race([Signal.race(this.#joinsChanged), expired]), signal)) === "expired") - return pending ? { condition: "timeout", reasonPhrase: "subscription not ready" } : missing; - } - } finally { - // Settles the deadline, which releases the change listener a cancelled wait left. - done.abort(); - } - } - - /** - * Resolve the head of a joining FETCH through its subscription's saved Largest Object, - * or the refusal to answer with. - */ - #fetchJoin( - type: Extract, - join: Joinable, - ): FetchAnswer | Refusal { - if (join.kind === "unsupported") - return { condition: "not_supported", reasonPhrase: "joining filter not supported" }; - const largest = join.largest; - if (!largest) return { condition: "invalid_range", reasonPhrase: "no objects at subscription start" }; - if (type.kind === "relativeJoining" && type.offset !== 0n) - return { condition: "not_supported", reasonPhrase: "joining FETCH reaches before the joined group" }; - const start = type.kind === "relativeJoining" ? largest.group : type.group; - if (start > largest.group) - return { condition: "invalid_range", reasonPhrase: "joining group past the subscription" }; - // The groups before the subscription's are a standalone FETCH's to ask for. - if (start !== largest.group) - return { condition: "not_supported", reasonPhrase: "joining FETCH spans more than one group" }; - - const end = { group: largest.group, object: largest.object + 1n }; - const sequence = end.group; - const unavailable: Refusal = { condition: "does_not_exist", reasonPhrase: "joining prefix unavailable" }; - const group = hooks.cachedGroup(join.track, sequence); - if (!group) return unavailable; - // 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 < frameBound(end.object)) { - group.close(); - return unavailable; - } - return { - cache: join.track, - owned: false, - range: { start: { group: end.group, object: 0n }, end }, - group, - slice: { skip: 0, until: frameBound(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, - publisherPriority: join.publisherPriority, - // Ranked as the group it heads, ahead of that group's live rest. - sendOrder: join.ranking.rank(sequence, 0), - ranking: join.ranking, - }; - } - - /** - * 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, cancel: AbortSignal) { - const version = this.#session.version; - 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 - // REQUEST_ERROR in answer to a FETCH. - await untilAborted(Message.write(request.writer, FetchOk.id, answer.ok, version), stop); - const stopped = new Promise((resolve) => { - if (stop.aborted) resolve(undefined); - stop.addEventListener("abort", () => resolve(undefined), { once: true }); - }); - stream = await Writer.tryOpen(this.#quic, { version, sendOrder: answer.sendOrder, cancel: stopped }); - if (!stream) throw stop.reason ?? new Error("fetch stream failed to open"); - ranking?.add(stream, range.start.group, frameBound(range.start.object)); - await untilAborted(new FetchHeader({ requestId: msg.requestId }).encode(stream, version), stop); - - const out = new FetchWriter({ stream, version, timescale, priority: publisherPriority }); - 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) { - const e = stop.aborted ? error(stop.reason) : error(err); - console.debug(`fetch failed: request=${msg.requestId} error=${reason(e)}`); - 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; - void opened.closed.catch(() => {}).then(() => ranking.remove(opened)); - } - } - } - /** * What a request that failed to resolve is refused with. A missing publisher a rendezvous * waited for is a timeout, which, like Rust's `rendezvous_error`, invites no retry. @@ -2838,57 +2397,6 @@ interface ServeRange { end?: EndLocation; } -/** What a draft-18 joining FETCH inherits from the SUBSCRIBE it names. */ -type Joinable = - /** The subscription's filter starts nowhere a joining FETCH could end. */ - | { kind: "unsupported" } - | { - kind: "join"; - /** The subscription's view of the track, whose cache holds the head. */ - track: TrackSubscriber; - /** The Largest Object the subscription starts right after, absent when there was none. */ - largest?: Location; - /** The subscription's namespace, which the fetch is held to the grant on. */ - namespace: Path.Valid; - /** The Track Properties the subscription was answered with. */ - properties: Properties.Properties; - /** The units the subscription stamps objects in, if any; see {@link RunGroup.timescale}. */ - timescale?: Timescale; - /** The track's Publisher Priority on the wire, which the fetch's first object states. */ - publisherPriority: number; - /** The subscription's ranking, which the fetch stream joins as the group it heads. */ - ranking: Priority; - }; - -/** The span a FETCH answers, as a standalone FETCH spells it. */ -interface FetchRange { - /** The first Location. */ - start: Location; - /** The exclusive end: End Object 0 includes the whole End Group. */ - end: Location; -} - -/** A FETCH resolved against its track, ready to answer. */ -interface FetchAnswer { - /** The track whose cache holds the answer's groups. */ - cache: TrackSubscriber; - /** Whether the answer subscribed `cache` itself, and so closes it. */ - owned: boolean; - range: FetchRange; - /** 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; - /** The track's Publisher Priority on the wire, which the stream's first object states. */ - publisherPriority: number; - /** The fetch stream's send order. */ - sendOrder: number; - /** The subscription ranking the fetch stream joins, if any. */ - ranking?: Priority; -} - // A filter keeps its peer-chosen Object IDs as bigint. Its derived frame selection // bound can stop at the model's count ceiling: no object exists at or above it, // so clamping either a skip or an exclusive end selects exactly the same frames. @@ -2896,33 +2404,6 @@ function frameBound(object: bigint): number { return object > BigInt(Number.MAX_SAFE_INTEGER) ? Number.MAX_SAFE_INTEGER : Number(object); } -/** Whether `a` is past `b`. */ -function after(a: Location, b: Location): boolean { - return a.group > b.group || (a.group === b.group && a.object > b.object); -} - -/** Where a draft-22 FETCH's filter starts, resolved against Largest Object. */ -function fetchStart(filter: Filter.Filter, largest: Location): Location { - switch (filter.kind) { - case "unfiltered": - return { group: 0n, object: 0n }; - case "nextObject": - return { group: largest.group, object: largest.object + 1n }; - case "relative": - return { group: saturatingSub(largest.group + 1n, filter.groups), object: 0n }; - case "absolute": - return { group: filter.startGroup, object: filter.startObject }; - } -} - -/** The slice of one group a {@link FetchRange} selects. */ -function fetchSlice(range: FetchRange, sequence: bigint): GroupSlice { - return { - skip: range.start.group === sequence ? frameBound(range.start.object) : 0, - until: range.end.group === sequence && range.end.object > 0n ? frameBound(range.end.object) : undefined, - }; -} - /** Write one group's fetch objects, inheriting the first object's priority. */ class FetchWriter { readonly stream: Writer; @@ -3094,7 +2575,7 @@ function liveEdge(track: TrackSubscriber): LiveEdge { } /** The Locations a single Location Filter selects, resolved against the live edge. */ -function filterRange(filter: Filter.Filter, edge: LiveEdge): ServeRange { +function filterRange(filter: Filter.Filter, edge: LiveEdge, version: IetfVersion): ServeRange { switch (filter.kind) { // No restriction. moq-lite starts at the beginning of the latest group, which is the // join point it is built around; a subscription passes objects as they are published, @@ -3104,10 +2585,14 @@ function filterRange(filter: Filter.Filter, edge: LiveEdge): ServeRange { return {}; // `{Largest.Group, Largest.Object + 1}`. Everything below it, including the already // published head of the current group, is outside the requested range, so the join is - // mid-group by construction. The draft pairs this with a fill, or a joining FETCH - // before draft-22, when the subscriber wants the head; see {@link Publisher.runFetch}. + // mid-group by construction. From draft-20 the draft pairs this with a fill when the + // subscriber wants the head. Before it, the head comes from a joining FETCH, which we + // refuse, so the subscription carries the current group whole instead: a subscriber + // starting at object 0 takes it as the whole group, and the head still reaches one that + // sends no joining FETCH. case "nextObject": - return { start: edge.next }; + if (!edge.next || Filter.isDraft20(version)) return { start: edge.next }; + return { start: { group: edge.next.group, object: 0n } }; // `{Largest.Group + 1 - groups, 0}`: 0 is the next group and 1 is the current one. // Counted from `Largest.Group`, which sits below the newest group while that group has // no objects yet; only with no largest at all does the newest group stand in for it. diff --git a/js/net/src/ietf/subscribe_tracks.ts b/js/net/src/ietf/subscribe_tracks.ts deleted file mode 100644 index 17a8346d28..0000000000 --- a/js/net/src/ietf/subscribe_tracks.ts +++ /dev/null @@ -1,40 +0,0 @@ -import type * as Path from "../path.ts"; -import type { Reader, Writer } from "../stream.ts"; -import * as Filter from "./filter.ts"; -import * as Message from "./message.ts"; -import * as Namespace from "./namespace.ts"; -import { Parameters } from "./parameters.ts"; -import { type IetfVersion, Version } from "./version.ts"; - -/** A request to list every track under a namespace prefix, which we refuse. @internal */ -export class SubscribeTracks { - static readonly id = 0x51; - readonly requestId: bigint; - readonly prefix: Path.Valid; - readonly parameters: Parameters; - constructor(requestId: bigint, prefix: Path.Valid, parameters = new Parameters()) { - this.requestId = requestId; - this.prefix = prefix; - this.parameters = parameters; - } - - async encode(writer: Writer, version: IetfVersion): Promise { - if (version < Version.DRAFT_18) throw new Error("SUBSCRIBE_TRACKS before draft-18"); - await Message.encode(writer, async (body) => { - await body.u62(this.requestId); - await Namespace.encode(body, this.prefix); - await this.parameters.encode(body, version); - }); - } - - static async decode(reader: Reader, version: IetfVersion): Promise { - if (version < Version.DRAFT_18) throw new Error("SUBSCRIBE_TRACKS before draft-18"); - return Message.decode(reader, async (body) => { - const requestId = await body.u62(); - const prefix = await Namespace.decode(body); - const parameters = await Parameters.decode(body, version, "subscribe-tracks"); - if (parameters.fillParameters) Filter.decodeFill(parameters.fillParameters, version); - return new SubscribeTracks(requestId, prefix, parameters); - }); - } -} diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 98a074c24f..39bec8a8c6 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -15,7 +15,6 @@ import * as Ietf from "./ietf/index.ts"; import { Publisher as IetfPublisher } from "./ietf/publisher.ts"; import { RequestError } from "./ietf/request.ts"; import { refuse as refuseRequest } from "./ietf/request_stream.ts"; -import { SubscribeTracks } from "./ietf/subscribe_tracks.ts"; import * as Instance from "./instance.ts"; import { hooks } from "./internal.ts"; import * as Lite from "./lite/index.ts"; @@ -1294,43 +1293,6 @@ test("integration: lite draft-07 fetch uses the republished track's timescale", server.abort(); }); -test("integration: ietf fetches a past group of a finished track", async () => { - for (const protocol of [Ietf.ALPN.DRAFT_18, Ietf.ALPN.DRAFT_20, Ietf.ALPN.DRAFT_22]) { - const pair = createMockTransportPair(protocol); - const origin = new OriginProducer(); - const publication = origin.route(Path.from("test")); - // No publisher limit, so the superseded group stays fresh to fetch, and no MAX_CACHE_DURATION. - const published = publication.broadcast.createTrack("video", { timescale: Timescale.MILLI, maxAge: null }); - for (let sequence = 0; sequence < 3; sequence++) { - const group = published.appendGroup(); - group.writeFrame({ - payload: new Uint8Array([sequence]), - timestamp: new Timestamp(sequence * 1000, Timescale.MILLI), - }); - group.finish(); - } - published.finish(); - const [client, server] = await Promise.all([ - connect(url, { transport: pair.client }), - accept(pair.server, url, { publish: origin.consume() }), - ]); - const remote = wireOf(client).consume(Path.from("test")); - const group = await remote.track("video").fetchGroup(1n); - expect(group.sequence).toBe(1n); - const frame = await group.readFrame(); - expect(frame?.payload).toEqual(new Uint8Array([1])); - expect(frame?.timestamp).toEqual(new Timestamp(1000, Timescale.MILLI)); - expect(await group.readFrame()).toBeUndefined(); - group.close(); - // The FETCH-only copy keeps the absent MAX_CACHE_DURATION as no limit, not the authored zero. - expect((await remote.track("video").info()).maxAge).toBeNull(); - remote.close(); - client.abort(); - server.abort(); - origin.close(); - } -}); - // We list no tracks by PUBLISH, so a SUBSCRIBE_TRACKS is refused NOT_SUPPORTED on its own stream, and // the subscription already on the session keeps going. test("integration: ietf SUBSCRIBE_TRACKS is refused per request", async () => { @@ -1365,8 +1327,9 @@ test("integration: ietf SUBSCRIBE_TRACKS is refused per request", async () => { const bidi = await pair.client.createBidirectionalStream(); const listing = new Stream({ ...bidi, version }); - await listing.writer.u53(SubscribeTracks.id); - await new SubscribeTracks(62n, Path.from("room")).encode(listing.writer, version); + // SUBSCRIBE_TRACKS (0x51): Request ID 62, the prefix "room", and no parameters. + await listing.writer.u53(0x51); + await listing.writer.write(Uint8Array.from([0, 8, 62, 1, 4, ...new TextEncoder().encode("room"), 0])); expect(await listing.reader.u53()).toBe(RequestError.id); expect((await RequestError.decode(listing.reader, version)).errorCode).toBe(0x3); diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index b774debbce..c4877bf6e2 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -827,8 +827,6 @@ 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. */ @@ -1170,9 +1168,6 @@ 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"); }, diff --git a/quest/m2/ietf-scope/README.md b/quest/m2/ietf-scope/README.md index 7a68470b75..f8ab7bfb5f 100644 --- a/quest/m2/ietf-scope/README.md +++ b/quest/m2/ietf-scope/README.md @@ -1,14 +1,16 @@ -# moq-transport is a compatibility layer again +# [S] Docs state the frozen moq-transport scope ## Goal -Our moq-transport support covers what a third-party implementation needs to +The docs that describe moq-transport support (`doc/concept/`, +`drafts/draft-lcurley-moq-ietf.md` if it lists the extensions, and the interop +matrix in [interop audit](/quest/m3/audit-interop.md)) state the frozen scope: +our moq-transport support covers what a third-party implementation needs to exchange live media with ours, and stops there: SETUP, the standard AUTHORIZATION TOKEN, live PUBLISH_NAMESPACE / SUBSCRIBE / PUBLISH, joining FETCH, and relay forwarding between moq-transport and moq-lite sessions, including across our relays, plus the extensions a third party speaks. The -IETF-only features nobody third-party uses are deleted from Rust, JS, and the -drafts. +code deletions are done; this is the docs work that remains. ## Plan @@ -30,6 +32,11 @@ Settled with this epic: [moq-dev/moq#7190](https://github.com/moq-dev/moq/pull/7190)), and the relay paths moq.pro's fleet smoke runs: cross-relay moq-transport subscribes and a draft-18 session to a third-party relay. +- Kept (decided 2026-11-10, maintainer, in + [moq-dev/moq#8445](https://github.com/moq-dev/moq/pull/8445)): moq-net's + listing of a moq-transport peer's tracks with SUBSCRIBE_TRACKS, since + Cloudflare's relay (cloudflare/moq-rs) answers it; without it we could not + list tracks on that relay. Receiving SUBSCRIBE_TRACKS stays refused. - Kept, as rejected in the same interview: lite-07's `ietf` object extension and moq-lite 01 to 04. - Not here: `ietf/flate.rs`; compression removal has its own work. @@ -45,11 +52,10 @@ them. AUTH cost about 2k dedicated lines plus wiring and no third party uses it; a third party authenticates with the standard AUTHORIZATION TOKEN, which stays. In-band grants and token renewal are moq-lite-07 only. -This epic's own work, after its children: the docs that describe -moq-transport support (`doc/concept/`, `drafts/draft-lcurley-moq-ietf.md` if -it lists the extensions, and the interop matrix in -[interop audit](/quest/m3/audit-interop.md)) state the frozen scope, and -AGENTS.md's IETF rule is reworded only if the maintainer asks. +Sized S as a standalone quest (decided 2026-11-11, maintainer, in +[moq-dev/moq#8445](https://github.com/moq-dev/moq/pull/8445), which deleted +the last child; rejected: finishing the docs there, XS). AGENTS.md's IETF rule +is reworded only if the maintainer asks. ## Related diff --git a/quest/m2/ietf-scope/serving.md b/quest/m2/ietf-scope/serving.md deleted file mode 100644 index 07691ef2cc..0000000000 --- a/quest/m2/ietf-scope/serving.md +++ /dev/null @@ -1,34 +0,0 @@ -# [M] moq-transport FETCH serving in JS and track listing are gone - -## Goal - -js/net no longer answers moq-transport FETCH as a publisher, and moq-net no -longer lists a moq-transport peer's tracks with SUBSCRIBE_TRACKS -(`ietf/subscribe_tracks.rs`, `ietf/subscriber/tracks.rs`, and the JS -`subscribe_tracks.ts`). A received SUBSCRIBE_TRACKS is refused, as it is -today. - -## Plan - -Part of the [IETF scope freeze](/quest/m2/ietf-scope/README.md), decided -2026-11-09. JS FETCH serving and its pacing landed in -[moq-dev/moq#6945](https://github.com/moq-dev/moq/pull/6945), -[moq-dev/moq#7793](https://github.com/moq-dev/moq/pull/7793), -[moq-dev/moq#7794](https://github.com/moq-dev/moq/pull/7794), -[moq-dev/moq#8117](https://github.com/moq-dev/moq/pull/8117), and -[moq-dev/moq#8231](https://github.com/moq-dev/moq/pull/8231); a browser -publisher serving a third-party FETCH has no user. Relays keep answering -FETCH from their cache, and a subscriber keeps its joining FETCH. - -Before deleting track listing, confirm no relay or moq.pro path needs it to -bridge a moq-transport publisher into moq-lite announcements; if one does, -keep that part, record why here, and delete the rest. A JS publisher that -receives a moq-transport FETCH answers it with the error moq-transport -defines for an unsupported request. - -[Subscribe ranges](/quest/m2/subscribe-ranges/README.md) fills relay misses -over moq-transport; this quest leaves relay FETCH alone. - -Public API: breaking only if js/net exposes a FETCH-serving option or Rust a -track-listing call publicly; name each in the PR's Impact section. Wire: -none; both are optional moq-transport behaviors. diff --git a/quest/m3/audit-interop.md b/quest/m3/audit-interop.md index dbd6c4b3c0..e46a7ac36c 100644 --- a/quest/m3/audit-interop.md +++ b/quest/m3/audit-interop.md @@ -72,7 +72,9 @@ Previous runs: [moq-dev/moq#6538](https://github.com/moq-dev/moq/pull/6538), Not yet filed: cloudflare/moq-rs `SubscribeRecv::error` reports a REQUEST_ERROR as `done` when the track has no reader. - Unsupported or down, not worth rechecking unless something changes: - PUBLISH and SUBSCRIBE_TRACKS cases, relays without namespace discovery, + PUBLISH and SUBSCRIBE_TRACKS cases, FETCH cases with a moq-dev-js + publisher (it refuses moq-transport FETCH, and on draft 18 a subscription + carries its current group whole instead), relays without namespace discovery, peers speaking only moq-transport drafts 14 to 17, raw-QUIC-only relays from JS, quiche-moq, and xquic's docker relay. diff --git a/quest/m3/ietf-js-joining-fetch.md b/quest/m3/ietf-js-joining-fetch.md deleted file mode 100644 index ca44c9eef4..0000000000 --- a/quest/m3/ietf-js-joining-fetch.md +++ /dev/null @@ -1,31 +0,0 @@ -# [S] The JS publisher answers a joining FETCH after a quick finish - -## Goal - -On draft 18, the JS publisher answers a joining FETCH that arrives after -its subscription finished cleanly, from the subscription's saved Largest -group, as the Rust publisher will. The `tail` interop cells with a native JS -or browser publisher pass on a track that finishes right after -SUBSCRIBE_OK. - -## Plan - -Decided 2026-10-25 (maintainer): honoring Largest Object and serving the -head moved into JavaScript FETCH in m1, which landed; only this part stays -in later work (m3). A FETCH that resolves before PUBLISH_DONE already finishes its head. - -In the tail test the track is finished before SUBSCRIBE_OK goes out, so -PUBLISH_DONE follows at once and the FETCH arrives to no subscription. Keep a -finished subscription's join answerable under the bound Rust settled in -[moq-dev/moq#7306](https://github.com/moq-dev/moq/pull/7306), and mirror it on the joinable-SUBSCRIBE registry js-fetch adds: -only a track that ended cleanly keeps its join, until the subscriber closes the -SUBSCRIBE's request stream, the session closes, or the request token's grant -or the session grant stops covering the track. - -Tests: a unit test for a FETCH that arrives after PUBLISH_DONE, the bound's -refusals mirrored from Rust, and the tail cells -`just test interop --tail --publishers js-native-node,js-native-bun --subscribers rust --versions ietf` -plus the browser publisher. - -Public API: none. Wire: none; a late joining FETCH is answered where it was -refused. diff --git a/sh/gh/nightly.sh b/sh/gh/nightly.sh index edd3ff95e5..acd3c71f3e 100755 --- a/sh/gh/nightly.sh +++ b/sh/gh/nightly.sh @@ -161,7 +161,7 @@ case "$1" in done ;; js) - 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 datagrams socket; do + for benchmark in broadcasts requests fills forward reader frames drift ledger bandwidth replay varint lite-varint track resume priority subscribe-drop refresh observed compression datagrams socket; do javascript "$benchmark" just js bench "$benchmark" done javascript catalog just js catalog-bench diff --git a/test/interop/README.md b/test/interop/README.md index 61e3b33d70..89d169d1c5 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -76,7 +76,9 @@ publisher and subscriber. Each reader fetches group 1 of a finished track, requi its three objects, each with its own bytes, and closes under the relay idle-out check. Each native JS runtime also serves that FETCH to all three subscribers from history (`CLI FETCH history`): its publisher writes only group 3 live and answers groups 0 through 2, which it never cached, from an in-memory responder on `track.requests()`. -This runs on moq-lite and every supported IETF draft. +Both lanes run on moq-lite and every supported IETF draft, except that a native JS +publisher's cells are skipped on moq-transport: the relay fills its miss with an upstream +FETCH, which a JS publisher refuses there by design. A late-join lane runs Rust-to-Rust and Rust/Node/Bun in both directions. The publisher writes the first two objects of group 0 before any subscriber exists and leaves the group diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 1d961072fa..7d49e4d51f 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -2147,6 +2147,13 @@ run_fetch_round() { local label="CLI FETCH${3:+ $3}" local started=$SECONDS why="" local cell=(case="$label" publisher="$pub" subscriber="$sub" log="$LOGS/$name-sub.log") + # The relay fills its miss with an upstream FETCH, which a JS publisher refuses on + # moq-transport by design: it serves FETCH only over moq-lite. + if [[ "$VERSION" == moq-transport-* && "$pub" == js-native-* ]]; then + echo " SKIP $label $pub -> $sub (a JS publisher refuses moq-transport FETCH)" + record skip "${cell[@]}" detail="a JS publisher refuses moq-transport FETCH" + return + fi if is_broken "$pub" || is_broken "$sub"; then echo " FAIL $label $pub -> $sub (client unavailable)" record fail "${cell[@]}" detail="client unavailable"