diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index 868c8362119..2c0c75759d4 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -38,15 +38,9 @@ Path normalization, relative references, and subscription paths retain their sem Drop the `await`. Every caller reads the same value: the old shared cursor let one caller take an update and leave another waiting forever. To wait for each newer one, keep a reader and follow its stream: `const reader = producer.subscribe();`, then `await reader.largest()`. -### Stage a requested track before subscribers see it (#6111) - -`Track.Request.stage(info)` returns a `Track.Staged` whose `producer` writes while subscribers waiting on the request stay pending; `commit()` publishes it with everything written so far and returns the `Producer`. -Disposing a `Staged` uncommitted (`using`, or `[Symbol.dispose]()`) rejects the request. `accept(info)` is unchanged and equals `stage(info).commit()`. - ### A frameless group is not a successor (#6111) A group followed by a frameless one is judged stale against the next group that has a frame, where the frameless group used to count. -`Track.Producer.dropGroups(start, end, error)` marks groups `[start, end)` the peer said it skipped; a reader holding one of them rejects with `error`. ### Priorities are signed (#6142) diff --git a/js/net/api/index.api.md b/js/net/api/index.api.md index 3fad27f882a..21bb1deb916 100644 --- a/js/net/api/index.api.md +++ b/js/net/api/index.api.md @@ -632,7 +632,6 @@ class Producer { close(abort?: Error): void; get closed(): GetPromise; demand(): Demand; - dropGroups(start: number, end: number, error: Error): void; finishAt(end: Position_2): void; info(): Promise; insertDatagram(sequence: number, timestamp: Timestamp | undefined, payload: Uint8Array): void; @@ -737,7 +736,6 @@ class Request_2 { readonly name: string; get priority(): number; reject(err?: Error): void; - stage(info?: Partial): Staged; get subscription(): Readonly; } @@ -825,12 +823,6 @@ export { Signals } function size(v: number): number; -class Staged { - [Symbol.dispose](): void; - commit(): Producer; - readonly producer: Producer; -} - interface Stats { bytesLost?: number; bytesReceived?: number; @@ -972,7 +964,6 @@ declare namespace Track { Subscription, Order, Request_2 as Request, - Staged, FetchGroupOptions, Consumer_3 as Consumer, Demand, @@ -993,7 +984,6 @@ declare namespace track { Subscription, Order, Request_2 as Request, - Staged, FetchGroupOptions, Consumer_3 as Consumer, Demand, diff --git a/js/net/api/zod.api.md b/js/net/api/zod.api.md index fbc1c617171..36aa2a49c23 100644 --- a/js/net/api/zod.api.md +++ b/js/net/api/zod.api.md @@ -141,7 +141,6 @@ class Producer { close(abort?: Error): void; get closed(): GetPromise; demand(): Demand; - dropGroups(start: number, end: number, error: Error): void; finishAt(end: Position_2): void; info(): Promise; insertDatagram(sequence: number, timestamp: Timestamp | undefined, payload: Uint8Array): void; @@ -176,16 +175,9 @@ class Request_2 { readonly name: string; get priority(): number; reject(err?: Error): void; - stage(info?: Partial): Staged; get subscription(): Readonly; } -class Staged { - [Symbol.dispose](): void; - commit(): Producer; - readonly producer: Producer; -} - class Subscriber { close(abort?: Error): void; get closed(): GetPromise; @@ -245,7 +237,6 @@ declare namespace track { Subscription, Order, Request_2 as Request, - Staged, FetchGroupOptions, Consumer_2 as Consumer, Demand, diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts index 8f669afa130..98a6e8e886a 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -3,6 +3,7 @@ import { Milli } from "@moq/clock"; import { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import { GroupTooLarge, NotFound } from "./error.ts"; import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "./group.ts"; +import { hooks } from "./internal.ts"; import { textFrame } from "./mock.ts"; import { Timescale, Timestamp } from "./time.ts"; import type { Request as TrackRequest } from "./track.ts"; @@ -62,7 +63,7 @@ test("a staged track resolves its subscribers at commit, with what was written", const request = await pendingRequest(consumer); if (!request) throw new Error("expected request"); - const staged = request.stage({ timescale: Timescale.MILLI }); + const staged = hooks.stageRequest(request, { timescale: Timescale.MILLI }); staged.producer.writeFrame(textFrame("hello")); const pending = Symbol("pending"); expect(await Promise.race([track.info(), Promise.resolve(pending)])).toBe(pending); @@ -72,31 +73,6 @@ test("a staged track resolves its subscribers at commit, with what was written", expect(await reader.readString()).toBe("hello"); }); -test("disposing an uncommitted staged track rejects the request", async () => { - const consumer = new TestConsumer(); - const info = consumer.track("video").info(); - const request = await pendingRequest(consumer); - if (!request) throw new Error("expected request"); - - { - using staged = request.stage({ timescale: Timescale.MILLI }); - staged.producer.writeFrame(textFrame("hello")); - } - await expect(info).rejects.toThrow(); -}); - -test("a disposed staged track refuses a later commit", async () => { - const consumer = new TestConsumer(); - const info = consumer.track("video").info(); - const request = await pendingRequest(consumer); - if (!request) throw new Error("expected request"); - - const staged = request.stage({ timescale: Timescale.MILLI }); - staged[Symbol.dispose](); - expect(() => staged.commit()).toThrow("staged track already committed or dropped"); - await expect(info).rejects.toThrow("staged track dropped"); -}); - // A broadcast keeps nothing of a name once its track is gone: a replacement numbers and times its // groups from scratch, and the application owns keeping them past its predecessor's. test("a replacement producer starts over", async () => { diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index fb659a19222..e5b60530c3a 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -1282,7 +1282,7 @@ export class Subscriber { } // Staged until the answer is written in below, so no reader sees the track before its // Largest and start. - const staged = request.stage({ + const staged = hooks.stageRequest(request, { // No TIMESCALE (always so on drafts 14-16, which can't carry it) means no timeline, // and the track must not claim one when served onward. timescale: ok.properties.timescale, diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 37994f7a00b..5a67fd88245 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -330,6 +330,17 @@ export interface TrackRequestOptions { pending: Set; } +/** + * A requested track written before any subscriber sees it; see `hooks.stageRequest`. + * Write through {@link producer}; {@link commit} publishes it with everything written so far. + */ +export interface StagedTrack { + /** The track's writer. */ + readonly producer: Producer; + /** Resolve every waiting subscriber now. Throws, rejecting the request, if the info is refused. */ + commit(): Producer; +} + /** Maximum buffered datagrams per subscriber; mirrors Rust's bounded send buffer. */ export const MAX_DATAGRAMS = 64; @@ -354,6 +365,13 @@ export const hooks: { * continues past it. Mirrors the Rust `close` on a `received()` copy; assigned by `track.ts`. */ releaseTrack: (producer: Producer) => void; + /** + * Record that the publisher gave up the groups `[start, end)` with `error`: none will + * arrive, so a reader waiting on one moves past it, a group still arriving in them ends with + * `error`, the declared end does not wait for them, and a relay serving this track forwards + * the drop. Mirrors the Rust `drop_groups`; assigned by `track.ts`. + */ + dropTrackGroups: (producer: Producer, start: number, end: number, error: Error) => void; /** The producer's largest reported location and its subsequent changes. */ producerLargest: (producer: Producer) => Getter; @@ -372,6 +390,11 @@ export const hooks: { startTrack: (producer: Producer, served: { start?: number; requested?: number }) => void; /** Mint a track {@link Request}; assigned by `track.ts`. */ makeRequest: (options: TrackRequestOptions) => Request; + /** + * Take a request's producer without resolving its subscribers, so content written before + * {@link StagedTrack.commit} is there when the first reader looks; assigned by `track.ts`. + */ + stageRequest: (request: Request, info: Partial) => StagedTrack; /** The newest frame a producer's cache holds, read without subscribing; assigned by `track.ts`. */ cachedLargest: (producer: Producer) => Position | undefined; /** The publisher priority from a demanded track's committed {@link Info}; assigned by `track.ts`. */ @@ -521,6 +544,9 @@ export const hooks: { releaseTrack: () => { throw new Error("track.ts not loaded"); }, + dropTrackGroups: () => { + throw new Error("track.ts not loaded"); + }, producerLargest: () => { throw new Error("track.ts not loaded"); }, @@ -536,6 +562,9 @@ export const hooks: { makeRequest: () => { throw new Error("track.ts not loaded"); }, + stageRequest: () => { + throw new Error("track.ts not loaded"); + }, cachedLargest: () => { throw new Error("track.ts not loaded"); }, diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index ba4a0675e58..8f6d25862ab 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -879,7 +879,7 @@ for (const version of [Version.DRAFT_06, Version.DRAFT_07]) { expect(await sub.servedSequence()).toBe(0); expect("start" in (await decodeSubscribeResponse(sub.client.reader, version))).toBe(true); - sub.track.dropGroups(1, 3, new StreamError(StreamCode(70), { message: "cut" })); + hooks.dropTrackGroups(sub.track, 1, 3, new StreamError(StreamCode(70), { message: "cut" })); const resp = await decodeSubscribeResponse(sub.client.reader, version); if (!("drop" in resp)) throw new Error("expected SUBSCRIBE_DROP"); expect([resp.drop.start, resp.drop.end, resp.drop.error]).toEqual([ @@ -1201,7 +1201,7 @@ for (const skipped of ["failed", "stale"] as const) { group.writeFrame(textFrame("old", Timestamp.fromMillis(5))); sub.track.writeGroup(group); group.close(skipped === "failed" ? new Error("cut") : undefined); - if (skipped === "failed") sub.track.dropGroups(7, 8, new Error("cut")); + if (skipped === "failed") hooks.dropTrackGroups(sub.track, 7, 8, new Error("cut")); sub.serve(6); const response = await decodeSubscribeResponse(sub.client.reader, Version.DRAFT_06); if (!("start" in response)) throw new Error("expected SUBSCRIBE_START"); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index c7a1cb8e1e7..245b45d4e59 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -1429,7 +1429,7 @@ export class Subscriber { // The publisher will never send these groups, so they are accounted for without a // stream, and a reader of one still arriving ends with the drop's error now. entry.tail.account(resp.drop.start, resp.drop.end + 1); - entry.track.dropGroups(resp.drop.start, resp.drop.end + 1, resp.drop.cause(this.version)); + hooks.dropTrackGroups(entry.track, resp.drop.start, resp.drop.end + 1, resp.drop.cause(this.version)); } } } diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index 021f2509f9b..2f346bd457f 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -1023,7 +1023,7 @@ for (const fails of [true, false]) { expect(tb.producer.demand().wanted.peek()).not.toBeUndefined(); const cut = new StreamError(StreamCode(70), { message: "cut" }); - tb.producer.dropGroups(0, 1, cut); + hooks.dropTrackGroups(tb.producer, 0, 1, cut); await expect(read).rejects.toBe(cut); sub.close(); @@ -1053,9 +1053,9 @@ test("a resumed reader reports what its serving copy gives up, once, and never s ta.write(0); expect(recvDisposition(sub)).toBe(0); // A route failing is no verdict: another route may still deliver the group. - ta.producer.dropGroups(1, 2, new StreamError(StreamCode.SessionClosed)); + hooks.dropTrackGroups(ta.producer, 1, 2, new StreamError(StreamCode.SessionClosed)); expect(recvDisposition(sub)).toBe("idle"); - ta.producer.dropGroups(2, 4, new StreamError(StreamCode(70), { message: "cut" })); + hooks.dropTrackGroups(ta.producer, 2, 4, new StreamError(StreamCode(70), { message: "cut" })); expect(recvDisposition(sub)).toEqual([2, 4, 70]); expect(recvDisposition(sub)).toBe("idle"); // Re-created under a run already reported dropped. @@ -1110,7 +1110,7 @@ test("a drop waits while a copy the serving one replaced may still deliver it", front.serve(b.consume()); await settle(); // Only the new route gave groups 1 and 2 up; the old one is not yet bounded. - tb.producer.dropGroups(1, 3, new StreamError(StreamCode(70), { message: "cut" })); + hooks.dropTrackGroups(tb.producer, 1, 3, new StreamError(StreamCode(70), { message: "cut" })); expect(recvDisposition(sub)).toBe("idle"); ta.write(1); expect(recvDisposition(sub)).toBe(1); @@ -1148,7 +1148,7 @@ test("a drop a replaced route held back is no verdict on the route serving next" front.serve(b.consume()); await settle(); // Held back: the copy b replaced may still deliver group 1. - tb.producer.dropGroups(1, 2, new StreamError(StreamCode(70), { message: "cut" })); + hooks.dropTrackGroups(tb.producer, 1, 2, new StreamError(StreamCode(70), { message: "cut" })); expect(recvDisposition(sub)).toBe("idle"); front.serve(c.consume()); await settle(); @@ -1176,7 +1176,7 @@ test("a front serving another front reports the inner serving route's drop", asy const sub = outer.consumer.track("seq").subscribe(); ta.write(0); expect(recvDisposition(sub)).toBe(0); - ta.producer.dropGroups(1, 3, new StreamError(StreamCode(70), { message: "cut" })); + hooks.dropTrackGroups(ta.producer, 1, 3, new StreamError(StreamCode(70), { message: "cut" })); expect(recvDisposition(sub)).toEqual([1, 3, 70]); expect(recvDisposition(sub)).toBe("idle"); // Re-created under a run already reported dropped. @@ -1960,7 +1960,7 @@ test.each(["lands", "dropped", "expired"])( expect(drain(sub)).toEqual([]); expect(ta.producer.demand().wanted.peek()).toBeDefined(); - if (outcome === "dropped") ta.producer.dropGroups(1, 2, new NotFound("group")); + if (outcome === "dropped") hooks.dropTrackGroups(ta.producer, 1, 2, new NotFound("group")); if (outcome === "expired") await clock.advance(Nano.fromMilli(SECOND)); if (outcome !== "lands") { // The groups drained above are released once their close settles. diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 749e3b8659d..98de6cc6ca4 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -2949,7 +2949,7 @@ const DROPPED = new StreamError(StreamCode.NotFound, { message: "dropped" }); test("explicit and producer-cut drops survive outside the group cache", async () => { const producer = new TrackProducer("video").accept({}); const reader = producer.subscribe({ maxDelay: Milli(60_000) }); - producer.dropGroups(3, 4, DROPPED); + hooks.dropTrackGroups(producer, 3, 4, DROPPED); expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "drop", start: 3, end: 4, error: DROPPED }); const group = new GroupProducer(4); @@ -3000,7 +3000,7 @@ test("an upstream drop ends a group still arriving with its error", async () => expect(group?.sequence).toBe(3); expect(await group?.readString()).toBe("3"); const cut = new StreamError(StreamCode(70), { message: "cut" }); - producer.dropGroups(3, 5, cut); + hooks.dropTrackGroups(producer, 3, 5, cut); // Settled by the drop itself, not a later reset; the finished group stays whole. await expect(group?.readString()).rejects.toBe(cut); const whole = await reader.recvGroup(); @@ -3014,7 +3014,7 @@ test("an upstream drop ends a group still arriving with its error", async () => test("a drop below the window is reported once the window lowers to it", () => { const producer = new TrackProducer("video").accept({}); const reader = producer.subscribe({ maxDelay: Milli(60_000), groups: { start: { included: 5 } } }); - producer.dropGroups(2, 3, DROPPED); + hooks.dropTrackGroups(producer, 2, 3, DROPPED); expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "idle" }); reader.withRange({ start: { included: 0 } }); @@ -3045,12 +3045,12 @@ test("a re-created cut group is served, not dropped", async () => { test("a wide drop is held and reported as one run", () => { const producer = new TrackProducer("video").accept({}); const reader = producer.subscribe({ maxDelay: Milli(60_000) }); - producer.dropGroups(2, 2 ** 40, DROPPED); + hooks.dropTrackGroups(producer, 2, 2 ** 40, DROPPED); expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "drop", start: 2, end: 2 ** 40, error: DROPPED }); // A wider drop reports only what this cursor has not reported yet, with its own error. const wider = new StreamError(StreamCode.Old, { message: "old" }); - producer.dropGroups(1, 2 ** 40 + 2, wider); + hooks.dropTrackGroups(producer, 1, 2 ** 40 + 2, wider); expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "drop", start: 1, end: 2, error: wider }); expect(hooks.tryRecvGroup(reader, true)).toEqual({ kind: "drop", @@ -3069,7 +3069,12 @@ test("adjacent drops with the same code merge into one run", () => { const reader = producer.subscribe({ maxDelay: Milli(60_000) }); // Each DROP decodes to its own error, so runs must merge by code, not identity. for (let sequence = 2; sequence < 5; sequence++) { - producer.dropGroups(sequence, sequence + 1, new StreamError(StreamCode.NotFound, { message: "dropped" })); + hooks.dropTrackGroups( + producer, + sequence, + sequence + 1, + new StreamError(StreamCode.NotFound, { message: "dropped" }), + ); } const recv = hooks.tryRecvGroup(reader, true); expect(recv.kind === "drop" ? [recv.start, recv.end, recv.error] : recv).toEqual([2, 5, DROPPED]); @@ -3603,7 +3608,7 @@ for (const abort of [false, true]) { const next = reader.recvGroup(); expect(await settled(next)).toBe(false); - producer.dropGroups(0, 2, DROPPED); + hooks.dropTrackGroups(producer, 0, 2, DROPPED); if (abort) producer.close(new Error("dropped")); expect(await next).toBeUndefined(); if (abort) expect(await reader.closed).toBeNull(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 1f96273c161..f9b79707614 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -36,6 +36,7 @@ import { type Run, removeRun, runGaps, + type StagedTrack, type TrackRequestOptions, } from "./internal.ts"; import { grace, Ledger } from "./ledger.ts"; @@ -342,6 +343,7 @@ export class Request { static { hooks.makeRequest = (options) => new Request(options); + hooks.stageRequest = (request, info) => request.#stage(info); } /** Watch the subscribers waiting for this request. */ @@ -361,21 +363,27 @@ export class Request { /** * Accept the request, committing the track's immutable {@link Info}. Throws, rejecting the - * request, if `info` is invalid. The same as {@link stage} followed by {@link Staged.commit}. + * request, if `info` is invalid. */ accept(info: Partial = {}): Producer { - return this.stage(info).commit(); + return this.#stage(info).commit(); } - /** - * Take the track's {@link Producer} without resolving any subscriber yet, so content written - * through {@link Staged.producer} is there when the first reader looks. For a relay that - * already holds groups of the track: write them, then {@link Staged.commit}. Disposing the - * {@link Staged} uncommitted rejects the request. - */ - stage(info: Partial = {}): Staged { + #stage(info: Partial): StagedTrack { this.#pending.delete(this); - return new Staged(this.#producer, info); + const producer = this.#producer; + return { + producer, + commit: () => { + try { + return producer.accept(info); + } catch (err) { + // Refused, so its subscribers see the error instead of waiting on an open request. + producer.close(err instanceof Error ? err : new Error(String(err))); + throw err; + } + }, + }; } /** Reject the request, closing the track optionally with an error. */ @@ -385,55 +393,6 @@ export class Request { } } -/** - * A requested track being written before any subscriber sees it; see {@link Request.stage}. - * - * Write through {@link producer}. Subscribers waiting on the request resolve only at - * {@link commit}, with everything written by then. Disposing it uncommitted rejects the request, - * so hold it with `using` or commit it. - * - * @public - */ -export class Staged { - /** The track's writer. */ - readonly producer: Producer; - - #info: Partial; - #done = false; - - /** @internal */ - constructor(producer: Producer, info: Partial) { - this.producer = producer; - this.#info = info; - } - - /** - * Publish the track: every waiting subscriber resolves now, with what was written. Throws, - * rejecting the request, if the info is invalid or something written does not match it. - * Throws without publishing once committed or disposed. - */ - commit(): Producer { - // Rust's `commit(self)` consumes the handle; here a disposed one has already rejected - // its subscribers, and accepting a closed producer would silently do nothing. - if (this.#done) throw new Error("staged track already committed or dropped"); - this.#done = true; - try { - return this.producer.accept(this.#info); - } catch (err) { - // Refused, so its subscribers see the error instead of waiting on an open request. - this.producer.close(err instanceof Error ? err : new Error(String(err))); - throw err; - } - } - - /** Reject the request unless it was committed. */ - [Symbol.dispose](): void { - if (this.#done) return; - this.#done = true; - this.producer.close(new Error("staged track dropped")); - } -} - /** Options for {@link Consumer.fetchGroup}. */ export interface FetchGroupOptions { /** Delivery priority for the fetch stream. Defaults to `0`. */ @@ -803,6 +762,7 @@ export class Producer { hooks.joinProducer = (producer) => producer.#join(); hooks.endTrack = (producer, end) => producer.#endPending(end); hooks.releaseTrack = (producer) => producer.#close(undefined, false); + hooks.dropTrackGroups = (producer, start, end, error) => producer.#drop(start, end, error, true); hooks.startTrack = (producer, served) => { producer.#served = served; producer.#owe(); @@ -832,16 +792,6 @@ export class Producer { mayLand = (producer, land) => producer.#mayLand(land); } - /** - * Record that the publisher gave up the groups `[start, end)` with `error`, mirroring the - * Rust `drop_groups`: none will arrive, so a reader waiting on one moves past it, a group - * still arriving in them ends with `error`, the declared end does not wait for them, and a - * relay serving this track forwards the drop. - */ - dropGroups(start: number, end: number, error: Error): void { - this.#drop(start, end, error, true); - } - // Whether a group may still land from the live feed, by the rule a pending tail is judged // by; see {@link Land}. Mirrors the Rust `TrackState::may_land`. #mayLand(land: Land): boolean { diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index aef6f473f00..f1e5976e555 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -76,7 +76,6 @@ Rejected in planning: - [Two broadcast resolutions](/quest/m1/api-audit/resolve.md) - Rust, JS, and the bindings resolve a broadcast with `request(path, epoch)` or `follow(path, epoch)`, and every racy two-step caller is gone - [Binding publications](/quest/m1/api-audit/binding-publication.md) - moq-ffi, the C ABI, and every wrapper hand out a publication handle that announces, replacing the private adapters - [Bindings read gaps and ends](/quest/m1/api-audit/decode-read-bindings.md) - moq-ffi and every wrapper surface `Gap` and `End` instead of discarding them -- [Session-only items stay private](/quest/m1/api-audit/model-private.md) - model items only the built-in sessions use are private - [Encoded consumers read gaps and ends](/quest/m1/api-audit/encoded-read-bindings.md) - moq-ffi, moq-c, and every wrapper read encoded media with the tagged-record shape the decoded consumers use - [One lifecycle vocabulary](/quest/m1/api-audit/lifecycle.md) - finish, abort, closed, and close mean one thing on every handle, and Rust and JS observers fire alike - [moq-net names say one thing](/quest/m1/api-audit/net-names.md) - demand says `wanted`, colliding stats and transport names are renamed, and Rust and JS info lookups mirror diff --git a/quest/m1/api-audit/lifecycle.md b/quest/m1/api-audit/lifecycle.md index 60e1acd57e2..cd8d5c08c17 100644 --- a/quest/m1/api-audit/lifecycle.md +++ b/quest/m1/api-audit/lifecycle.md @@ -33,8 +33,8 @@ handle's writer verbs and observers: Changes: -- Rust `track::Producer::close()`, crate-private after - [session-only items stay private](/quest/m1/api-audit/model-private.md), +- Rust `track::Producer::close()`, crate-private since + [moq-dev/moq#6747](https://github.com/moq-dev/moq/pull/6747), folds into `abort` with the clean-close error ([moq-dev/moq#6482](https://github.com/moq-dev/moq/pull/6482) makes it `Stream(Session(Cancel))`). Readers that drained a closed copy to a clean diff --git a/quest/m1/api-audit/model-private.md b/quest/m1/api-audit/model-private.md deleted file mode 100644 index 1d6dfa258ec..00000000000 --- a/quest/m1/api-audit/model-private.md +++ /dev/null @@ -1,34 +0,0 @@ -# [M] Session-only model items stay private - -## Goal - -moq-net and `@moq/net` model items that only the built-in sessions use are -private. The public model serves applications: publish, subscribe, and answer -on demand. - -## Plan - -Decided 2026-10-23 (maintainer): the old public-model goal, "a third party -builds its own publisher, subscriber, or forwarding relay from the public -model alone", is dropped. It made every item a built-in session reaches -public, with no consumer outside the sessions. Items go public again only -for a named consumer. - -- Revert the session-only items - [moq-dev/moq#6111](https://github.com/moq-dev/moq/pull/6111) made public: - `track::Producer::{received, drop_groups, close}`, - `track::Request::{request_start, stage}`, `track::Staged`, and - `group::Producer::info` (it duplicates `Deref`) become - `pub(crate)`. JS `Track.Request.stage`, `Track.Staged`, and - `Track.Producer.dropGroups` become internal again (`dropGroups` returns to - `hooks.dropTrackGroups`). Keep `create_frame(self)`: hang and moq-mux use it. -- The `// pending:` and `// private:` marks and their lint are already gone. - The moq-transport passthrough fields stay crate-private. - -Rejected: the toy-transport integration tests proving a third party could -build from the public model; the goal they proved is dropped. - -Public API: moq-net and `@moq/net` removals, all main-only (none reached -`release`), so no migration note; the Impact bullet flags the unreleased #6111 entry -for the cut to drop. -Wire: none. diff --git a/quest/m2/release/api-review.md b/quest/m2/release/api-review.md index 7a023705dcf..157bb6b4338 100644 --- a/quest/m2/release/api-review.md +++ b/quest/m2/release/api-review.md @@ -25,8 +25,8 @@ against AGENTS.md's Public API rules: names mirrored across languages, no callbacks or 4+ args, misuse unrepresentable, private until a consumer needs it. Propose 2-3 shapes with a recommendation, ask the maintainer, and file each approved change as a quest in m1, which holds the cut until it lands. -Related: [session-only items stay private](/quest/m1/api-audit/model-private.md), which decides what -moq-net exposes; this review judges the shape of what ships. +Model items only the built-in sessions use stay private +([moq-dev/moq#6747](https://github.com/moq-dev/moq/pull/6747)); this review judges the shape of what ships. ## Required diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index f842b0e3a24..6fd59fcf89b 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -149,7 +149,7 @@ harness = false [[bench]] name = "tail" harness = false -required-features = ["async"] +required-features = ["fuzz"] [[bench]] name = "priority" diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index b0d41ad16ac..73e9011880d 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -101,18 +101,6 @@ lease.release(); `transport::CloseKind` gained `Evicted`, which `Error::from_transport` maps to `Error::Evicted`. `transport::Stats` gained `streams_evicted()`, and `session::Stats` the matching `streams_evicted` field. -### Stage a requested track before subscribers see it (#6111) - -`track::Request::stage(info)` returns a `track::Staged` that derefs to the track's `Producer` while subscribers waiting on the request stay pending; `commit()` publishes it with everything written so far and returns the `Producer`. -A relay that holds groups of a lazily served track writes them before committing, so the first reader (and an IETF `SUBSCRIBE_OK`'s Largest Location) sees them. -Dropping a `Staged` uncommitted rejects the request with `Error::Dropped`. `accept(info)` is unchanged and equals `stage(info).commit()`. - -```rust -let mut staged = request.stage(info); -for group in retained { /* staged.create_group(..), write, finish */ } -let producer = staged.commit(); -``` - ### Frame writers own their group (#6111) `group::Producer::create_frame` now takes the writer by value and returns an owned `frame::Producer` with no lifetime. @@ -132,14 +120,11 @@ group = frame.finish()?; A caller that holds the group behind `&mut` can write a whole frame with `write_frame` instead. -### A received track's end waits for groups in flight (#6111) +### A track's end tolerates a repeat, and skipped groups can be dropped (#6111) -Call `track::Producer::received()` before writing what a peer sends into a track, then declare the peer's end with `finish_at`. -On a received copy, readers wait at a hole below that end for the subscription's max delay (or one second) in case its group is still in flight; `drop_groups(range, err)` marks groups the peer said it skipped, so readers stop waiting for them and a group still arriving there ends with `err`. +`track::Producer::drop_groups(range, err)` marks groups the publisher skipped, so readers stop waiting for them and a group still arriving there ends with `err`. `finish_at` now accepts a repeat of the same end and a refinement within the last group; any other second end fails with `Error::ProtocolViolation`. -`track::Request::request_start(floor)` leaves an accepted copy's start pending at `floor` until `track::Producer::start_at` resolves it, for a subscriber whose peer reports the start after accepting. - ### A frameless group is not the live edge (#6111) Every group reaches readers as soon as it is created, but the live edge, `largest()`, and the judgment of whether a group was succeeded count a group only once its first frame opens or it finishes. @@ -148,8 +133,7 @@ A group followed by a frameless one is judged against the next group that has a ### Ending a track aborts its open groups (#6111) Aborting a track (`abort` or `abort_unused`) now aborts its unfinished groups with the same error, so their readers end with it and their writers fail. -Dropping the last `track::Producer` without a finish aborts them with `Error::Dropped`, and `close` with `Error::Stream(StreamError::Session(SessionError::Cancel))`. -`track::Producer::close`, `track::Producer::received`, and `group::Producer::info` are now public. +Dropping the last `track::Producer` without a finish aborts them with `Error::Dropped`. ### Priorities are signed, with 0 unset (#6142) diff --git a/rs/moq-net/benches/tail.rs b/rs/moq-net/benches/tail.rs index 3ca1d3ca13a..6f196ee81d2 100644 --- a/rs/moq-net/benches/tail.rs +++ b/rs/moq-net/benches/tail.rs @@ -5,7 +5,7 @@ //! check is judged by the track's record of arrivals, so a cost that grows with the groups //! cached or the tracks held shows up as a slope over either axis. //! -//! Run with `cargo bench -p moq-net --bench tail`. +//! Run with `cargo bench -p moq-net --features fuzz --bench tail`. use std::{hint::black_box, task::Poll, time::Duration}; @@ -38,7 +38,7 @@ impl Held { track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); track.start_at(0).unwrap(); let end = track::Position::group(groups as u64 + 1); - track.received(); + moq_net::fuzz::received(&track); track.finish_at(end).unwrap(); // Group 0 never arrives: the hole that holds the reader. for sequence in 1..=groups as u64 { diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index bb348e6dfb8..6139ad8bffb 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -290,6 +290,12 @@ impl AnnounceWriter { } } +/// Mark `track` as a copy received from a remote publisher, as a subscriber session does, for +/// the tail bench. +pub fn received(track: &track::Producer) { + track.received(); +} + /// Fill the `groups` of `track` its cache misses, a run per FETCH, as a publisher answers a /// range FETCH, for the session bench. The returned guard keeps the filled groups cached. pub async fn fill_groups(track: &track::Consumer, groups: std::ops::Range) -> crate::Result { diff --git a/rs/moq-net/src/model/frame.rs b/rs/moq-net/src/model/frame.rs index 0f08d87d5e2..e2df69a6f67 100644 --- a/rs/moq-net/src/model/frame.rs +++ b/rs/moq-net/src/model/frame.rs @@ -636,7 +636,7 @@ impl Drop for Producer { }; if !group.is_aborted() && !self.cancel_on_drop { tracing::warn!( - group = group.info().sequence, + group = group.sequence, "frame::Producer dropped before writing all bytes" ); } diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index a8229ddf461..661af18b9c1 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -1157,11 +1157,6 @@ impl Producer { self } - /// The group's header, as it was created. - pub fn info(&self) -> Info { - self.handle.info - } - /// The parent track's timescale, or `None` when it declared no timeline. pub fn timescale(&self) -> Option { self.handle.track.timescale @@ -4607,13 +4602,6 @@ mod test { assert!(pool.used() <= before); } - /// A writer names the group it fills. - #[test] - fn info_names_the_group() { - let producer = Info { sequence: 7 }.produce(); - assert_eq!(producer.info(), Info { sequence: 7 }); - } - /// An open frame still reads its group: which one it is, how many frames it holds, /// and whether anyone reads it. Finishing the frame hands the same writer back. #[test] @@ -4625,7 +4613,7 @@ mod test { let mut frame = producer.create_frame(sized(1)).unwrap(); let group = frame.group(); - assert_eq!(group.info().sequence, 3); + assert_eq!(group.sequence, 3); assert_eq!(group.frame_count(), 2, "the open frame counts"); assert!(group.demand().is_used()); assert!(group.poll_closed(&kio::Waiter::noop()).is_pending()); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 78ac900c6d1..c8dba85a33a 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2975,7 +2975,8 @@ impl Producer { /// with it and their writers fail. A reader following the track's routes resumes from /// another one where this copy stopped. Use [`Self::finish`] or [`Self::finish_at`] /// when the end is known. - pub fn close(self) -> Result<()> { + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn close(self) -> Result<()> { let mut state = self.modify()?; state.sealed = true; let open = state.tear_down(Error::Stream(StreamError::Session(SessionError::Cancel))); @@ -3017,7 +3018,7 @@ impl Producer { /// boundary it falls just before (a peer may name only the boundary first). Any other /// second end is [`Error::ProtocolViolation`]. /// - /// On a [received](Self::received) copy, groups below the end may still be in flight: + /// On a copy a session received, groups below the end may still be in flight: /// readers wait for a missing one until a grace passes, the subscription's max delay /// or one second without one, before ending at the boundary. pub fn finish_at(&mut self, end: Position) -> Result<()> { @@ -3051,7 +3052,7 @@ impl Producer { /// still promised. Pass `None` to clear it, for demand at the live edge: /// its floor is unknown until the feed declares one. /// - /// Resolves a start left pending by [`Request::request_start`]. + /// Resolves a start a subscriber session left pending. pub fn start_at(&mut self, sequence: impl Into>) -> Result<()> { self.modify()?.set_start(sequence.into(), false); Ok(()) @@ -3063,7 +3064,8 @@ impl Producer { /// Its readers' relative and time anchors, other than [`Range::live`](crate::track::Range::live) /// and [`Range::oldest`](crate::track::Range::oldest), fail with [`Error::Unsupported`] /// unless its session's wire lets it resolve them, since it holds only part of the track. - pub fn received(&self) { + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn received(&self) { if let Ok(mut state) = self.modify() { state.received = true; } @@ -7424,7 +7426,7 @@ impl Ordered { /// request resolve only at [`Self::commit`], with everything written by then. Dropping it /// uncommitted rejects the request with [`Error::Dropped`]. #[must_use = "a staged track is rejected unless committed"] -pub struct Staged { +pub(crate) struct Staged { // Taken only by `commit`. producer: Option, // The request's fetch claim, held until the track is published like an unanswered @@ -7434,7 +7436,7 @@ pub struct Staged { impl Staged { /// Publish the track: every waiting subscriber resolves now, with what was written. - pub fn commit(mut self) -> Producer { + pub(crate) fn commit(mut self) -> Producer { let producer = self.producer.take().expect("taken only by commit"); if let Ok(mut state) = producer.state.write() { state.staged = false; @@ -7472,8 +7474,7 @@ impl Drop for Staged { /// Yielded by [`poll_requested_track`](crate::broadcast::Dynamic::poll_requested_track), /// or created up front with [`broadcast::Producer::reserve_track`](crate::broadcast::Producer::reserve_track). /// Subscribers block until the request is -/// resolved: call [`accept`](Self::accept) to serve it with a [`Producer`] (or -/// [`stage`](Self::stage) to write content before they see it), or +/// resolved: call [`accept`](Self::accept) to serve it with a [`Producer`], or /// [`reject`](Self::reject) to fail them. Dropping it without either rejects with /// [`Error::Dropped`]. /// @@ -7559,7 +7560,8 @@ impl Request { /// as they arrive, but a relay serving the track waits for the resolved start before /// declaring its own. Applied atomically with [`Self::accept`], before any reader can /// see the track. - pub fn request_start(mut self, floor: impl Into>) -> Self { + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn request_start(mut self, floor: impl Into>) -> Self { self.requested_start = Some(floor.into()); self } @@ -7641,8 +7643,6 @@ impl Request { /// (`None` for the defaults). If the track was already aborted, the returned /// [`Producer`] is inert: writes fail with the abort error, as if it had been /// aborted immediately after accepting. - /// - /// The same as [`Self::stage`] followed by [`Staged::commit`]. pub fn accept(self, info: impl Into>) -> Producer { self.stage(info).commit() } @@ -7653,7 +7653,7 @@ impl Request { /// For a relay that already holds groups of the track: write them, then /// [`Staged::commit`]. Dropping the [`Staged`] uncommitted rejects the request with /// [`Error::Dropped`]. - pub fn stage(self, info: impl Into>) -> Staged { + pub(crate) fn stage(self, info: impl Into>) -> Staged { let info = info.into().unwrap_or_default(); // A closed state means the track was aborted under us. Mirror `reject` and // tolerate it: the Producer we hand back simply can't write.