diff --git a/js/hang/api/container.api.md b/js/hang/api/container.api.md index 0e16dbd15d8..d6d944fe695 100644 --- a/js/hang/api/container.api.md +++ b/js/hang/api/container.api.md @@ -45,6 +45,9 @@ export interface Frame { export namespace Legacy { export { BufferedRange }; export { BufferedRanges }; + export interface Finished { + abort(err: Error): void; + } export class Format implements Format_1 { constructor(config: AudioConfig | VideoConfig | "audio" | "video" | "data"); decode(frame: Uint8Array): Frame[]; @@ -55,9 +58,10 @@ export namespace Legacy { export { Frame }; export class Producer { constructor(track: Net.Track.Producer, format: Legacy.Format); - close(err?: Error): void; + abort(err: Error): void; discontinuity(endpoint?: Net.Clock.Duration): void; encode(data: Uint8Array | Legacy.Source, pts: Net.Clock.Duration, keyframe: boolean): void; + finish(): Legacy.Finished; } export interface Source { byteLength: number; diff --git a/js/hang/src/container/consumer.rewind.test.ts b/js/hang/src/container/consumer.rewind.test.ts index dbd8b143254..403c911ad13 100644 --- a/js/hang/src/container/consumer.rewind.test.ts +++ b/js/hang/src/container/consumer.rewind.test.ts @@ -76,7 +76,7 @@ for (const start of [16_000, 0]) { expect((await frame(consumer))?.group).toBe(1n); } finally { consumer.close(); - producer.close(); + producer.finish(); } }); } @@ -132,6 +132,6 @@ test("Producer refuses a rewind before closing the current group", () => { producer.encode(new Uint8Array([1]), Duration.micros(150_000), false); expect(() => producer.encode(new Uint8Array([1]), Duration.micros(99_000), false)).toThrow("below"); } finally { - producer.close(); + producer.finish(); } }); diff --git a/js/hang/src/container/consumer.test.ts b/js/hang/src/container/consumer.test.ts index e13fe20d41d..53731c956c3 100644 --- a/js/hang/src/container/consumer.test.ts +++ b/js/hang/src/container/consumer.test.ts @@ -162,7 +162,7 @@ test("Legacy Producer refuses a group below the previous group start", () => { expect(() => producer.encode(new Uint8Array([1]), Duration.micros(16_000), true)).toThrow( "below the previous group start", ); - producer.close(); + producer.finish(); }); test("Legacy Producer accepts B-frames within a group", () => { @@ -171,7 +171,7 @@ test("Legacy Producer accepts B-frames within a group", () => { producer.encode(new Uint8Array([1]), Duration.micros(0), true); producer.encode(new Uint8Array([1]), Duration.micros(66_000), false); producer.encode(new Uint8Array([1]), Duration.micros(33_000), false); - producer.close(); + producer.finish(); }); test("Legacy Producer accepts open-GOP leading pictures above the previous group", () => { @@ -181,7 +181,7 @@ test("Legacy Producer accepts open-GOP leading pictures above the previous group producer.encode(new Uint8Array([1]), Duration.micros(33_000), false); producer.encode(new Uint8Array([1]), Duration.micros(66_000), true); producer.encode(new Uint8Array([1]), Duration.micros(50_000), false); - producer.close(); + producer.finish(); }); test("Legacy Producer writes a duration marker at the next keyframe", async () => { @@ -191,7 +191,7 @@ test("Legacy Producer writes a duration marker at the next keyframe", async () = producer.encode(new Uint8Array([0xde, 0xad]), Duration.micros(0), true); producer.encode(new Uint8Array([0xbe, 0xef]), Duration.micros(10_000), false); producer.encode(new Uint8Array([0xca, 0xfe]), Duration.micros(20_000), true); - producer.close(); + producer.finish(); const group = await subscriber.recvGroup(); expect(group).toBeDefined(); @@ -218,7 +218,7 @@ test("Legacy Producer omits a reordered group's presentation endpoint marker", a } producer.encode(new Uint8Array([1]), Duration.micros(160_000), true); producer.discontinuity(Duration.micros(200_000)); - producer.close(); + producer.finish(); const group = await subscriber.recvGroup(); expect(group).toBeDefined(); let count = 0; @@ -242,7 +242,7 @@ test("Legacy Producer estimates the tail from the current cadence", async () => for (const [index, timestamp] of [0, 16_000, 32_000, 65_000, 98_000].entries()) { producer.encode(new Uint8Array([1]), Duration.micros(timestamp), index === 0); } - producer.close(); + producer.finish(); const group = await subscriber.recvGroup(); expect(group).toBeDefined(); let end = 0; @@ -263,7 +263,7 @@ test("Legacy Producer rejects a backwards discontinuity without closing the grou expect(() => producer.discontinuity(Duration.micros(10_000))).toThrow(); producer.encode(new Uint8Array([2]), Duration.micros(30_000), false); producer.discontinuity(Duration.micros(35_000)); - producer.close(); + producer.finish(); const group = await subscriber.recvGroup(); const timestamps = []; for (;;) { @@ -281,7 +281,7 @@ test("Legacy Producer keeps the cadence after rejecting a backwards discontinuit producer.encode(new Uint8Array([1]), Duration.micros(20_000), true); producer.encode(new Uint8Array([2]), Duration.micros(30_000), false); expect(() => producer.discontinuity(Duration.micros(10_000))).toThrow(); - producer.close(); + producer.finish(); const group = await subscriber.recvGroup(); const timestamps = []; for (;;) { @@ -300,7 +300,7 @@ test("Legacy Producer refuses a keyframe that rewinds the timeline", () => { expect(() => producer.encode(new Uint8Array([3]), Duration.micros(0), true)).toThrow( "below the previous group start", ); - producer.close(); + producer.finish(); }); /** Read every group a subscriber is served until `last`, as (sequence, [timestamp, payload size][]). */ @@ -330,7 +330,7 @@ test("Legacy Producer discontinuity marks the break with one empty frame at the producer.discontinuity(); producer.discontinuity(); // nothing new to mark producer.encode(new Uint8Array([1]), Duration.micros(5_000_000), true); - producer.close(); + producer.finish(); expect(await readGroups(subscriber, 3n)).toEqual([ [0n, [[0, 1]]], @@ -346,7 +346,7 @@ test("Legacy Producer discontinuity marks the break at the caller's end", async const producer = new LegacyProducer(track, new LegacyFormat("video")); producer.encode(new Uint8Array([1]), Duration.micros(0), true); producer.discontinuity(Duration.micros(33_000)); - producer.close(); + producer.finish(); expect(await readGroups(subscriber, 1n)).toEqual([ [ @@ -370,7 +370,7 @@ test("Legacy Producer discontinuity writes no end estimated from the cadence", a producer.discontinuity(); // The capture swap resumes sooner than one frame later: no end past it, so no rewind. producer.encode(new Uint8Array([1]), Duration.micros(80_000), true); - producer.close(); + producer.finish(); expect(await readGroups(subscriber, 2n)).toEqual([ [ @@ -402,7 +402,7 @@ test("Legacy Producer discontinuity on a data track reads as an end at the edge" producer.encode(new Uint8Array(), Duration.micros(20_000), true); producer.discontinuity(); producer.encode(new Uint8Array([1]), Duration.micros(5_000_000), true); - producer.close(); + producer.finish(); const consumer = new Consumer(subscriber, { format: new LegacyFormat("data"), @@ -449,7 +449,7 @@ test("Legacy Producer discontinuity keeps pre-pause media from reading as live", producer.encode(new Uint8Array([1]), Duration.micros(33_000), false); producer.discontinuity(); producer.encode(new Uint8Array([1]), Duration.micros(5_000_000), true); - producer.close(); + producer.finish(); const subscriber = track.subscribe({ delay: { max: Duration.millis(1_000) } }); expect((await readGroups(subscriber, 2n)).map(([sequence]) => sequence)).toEqual([1n, 2n]); @@ -1168,7 +1168,7 @@ test("Consumer preserves empty Legacy and LOC data frames", async () => { if (kind === "legacy") { const producer = new LegacyProducer(track, new LegacyFormat("data")); producer.encode(new Uint8Array(), Duration.micros(0), true); - producer.close(); + producer.finish(); } else { writeLoc(track, [[[0, []]]]); } @@ -2113,7 +2113,7 @@ test("audio discontinuity writes no duration marker", async () => { const group = await subscriber.recvGroup(); expect(await group?.readFrame()).toBeDefined(); expect(await group?.readFrame()).toBeUndefined(); - producer.close(); + producer.finish(); }); for (const code of [StreamCode.Cancel, StreamCode.Internal, StreamCode.Old, StreamCode.Evicted, StreamCode(1234)]) { @@ -2254,3 +2254,13 @@ test("LegacyFormat rejects a timestamp past 2^53 - 1 instead of rounding", () => const frame = Varint.encode(2n ** 53n + 1n); expect(() => new LegacyFormat("video").decode(frame)).toThrow(/larger than 53-bits/); }); + +// Mirrors the Rust `container::Finished` doctest: the track `finish` returns takes no frame, at +// compile time and at run time. +test("finishing a Legacy Producer returns a track that takes no frame", () => { + const producer = new LegacyProducer(new Track.Producer("test"), new LegacyFormat("video")); + producer.encode(new Uint8Array([1]), Duration.micros(0), true); + const finished = producer.finish(); + // @ts-expect-error a finished track takes no frame + expect(() => finished.encode(new Uint8Array([1]), Duration.micros(33_000), true)).toThrow(); +}); diff --git a/js/hang/src/container/legacy.ts b/js/hang/src/container/legacy.ts index 0853c53bc65..0a08ad42bd5 100644 --- a/js/hang/src/container/legacy.ts +++ b/js/hang/src/container/legacy.ts @@ -68,6 +68,16 @@ function micros(timestamp: Duration): number { return Math.floor(timestamp.asMicros()); } +/** + * A track its {@link Producer} finished: every frame is written and it takes no more. + * {@link Producer.finish} returns it, mirroring the Rust `container::Finished`; the producer itself + * still refuses a frame after the finish. + */ +export interface Finished { + /** Abort the track with an error; see {@link Producer.abort}. */ + abort(err: Error): void; +} + /** Writes legacy-container frames into a MoQ track, starting a new group on each keyframe. */ export class Producer { #track: Moq.Track.Producer; @@ -190,15 +200,18 @@ export class Producer { this.#reordered = false; } - /** Close the track and current group, optionally with an error. */ - close(err?: Error) { - if (!err) this.#close(); - if (err) this.#group?.abort(err); - else this.#group?.finish(); - if (err) this.#track.abort(err); - else { - this.#track.finish(); - this.#track.close(); - } + /** Close the current group and finish the track, returning it as {@link Finished}. */ + finish(): Finished { + this.#close(); + this.#track.finish(); + this.#track.close(); + return this; + } + + /** Abort the current group and the track with `err`, which readers see in place of a clean end. */ + abort(err: Error) { + this.#group?.abort(err); + this.#group = undefined; + this.#track.abort(err); } } diff --git a/js/json/api/index.api.md b/js/json/api/index.api.md index 31d28825535..47fa3074d55 100644 --- a/js/json/api/index.api.md +++ b/js/json/api/index.api.md @@ -57,6 +57,7 @@ export namespace Snapshot { update(value: T): Snapshot.Pending | undefined; get value(): T | undefined; } + export interface Finished {} export { MissingSnapshot }; export interface Pending extends Snapshot.Encoded { commit(): void; @@ -69,7 +70,7 @@ export namespace Snapshot { export class Producer { constructor(config: Snapshot.Producer.Config); cut(): void; - finish(): void; + finish(): Snapshot.Finished; mutate(fn: (value: T) => void, at?: Time.Timestamp): void; update({ ... }: Timed): void; } @@ -97,6 +98,7 @@ export namespace Stream { constructor(config?: Stream.Config); encode(value: T): Stream.Pending; } + export interface Finished {} export interface Pending { commit(): void; payload: Uint8Array; @@ -109,7 +111,7 @@ export namespace Stream { export class Producer { constructor(config: Stream.Producer.Config); append({ ... }: Timed): void; - finish(): void; + finish(): Stream.Finished; } export class Rolled extends Error { constructor(); @@ -160,6 +162,11 @@ export namespace Window { } | { skip: Window.Span; }; + export interface Finished { + abort(err: Error): void; + readonly offset: number; + readonly window: T[]; + } export interface Group { decode(payload: Uint8Array): void; } @@ -170,7 +177,7 @@ export namespace Window { constructor(config: Window.ProducerConfig); abort(err: Error): void; cut(): void; - finish(): void; + finish(): Window.Finished; get offset(): number; pop(count: number, at?: Time.Timestamp): void; push({ ... }: Timed): void; diff --git a/js/json/src/snapshot/index.ts b/js/json/src/snapshot/index.ts index 311ecbfdd2d..7b81e4b889c 100644 --- a/js/json/src/snapshot/index.ts +++ b/js/json/src/snapshot/index.ts @@ -40,4 +40,4 @@ export { MissingSnapshot } from "../error.ts"; export { Consumer } from "./consumer.ts"; export { Decoder } from "./decoder.ts"; export { type Config, type Encoded, Encoder, type Pending } from "./encoder.ts"; -export { Producer } from "./producer.ts"; +export { type Finished, Producer } from "./producer.ts"; diff --git a/js/json/src/snapshot/producer.ts b/js/json/src/snapshot/producer.ts index 671bbd09b49..ed5d45be3db 100644 --- a/js/json/src/snapshot/producer.ts +++ b/js/json/src/snapshot/producer.ts @@ -4,6 +4,14 @@ import { MAX_FRAME_BYTES } from "../limits.ts"; import { type Config as CodecConfig, type Encoded, Encoder } from "./encoder.ts"; +/** + * A value its producer finished: it takes no more updates. {@link Producer.finish} returns it, + * mirroring the Rust `snapshot::Finished`; the producer itself still refuses an update after the + * finish. + */ +// biome-ignore lint/suspicious/noEmptyInterface: nothing is left to call once the track is released. +export interface Finished {} + /** * Publishes a JSON value over a track, choosing snapshots and deltas automatically. * @@ -182,9 +190,10 @@ export class Producer { this.update({ value, at }); } - /** Finish the track, closing any open group. */ - finish(): void { + /** Finish the track, closing any open group, and return it as {@link Finished}. */ + finish(): Finished { this.#close(); + return this; } #close(abort?: Error): void { diff --git a/js/json/src/snapshot/snapshot.test.ts b/js/json/src/snapshot/snapshot.test.ts index 76142215667..39702a4c44b 100644 --- a/js/json/src/snapshot/snapshot.test.ts +++ b/js/json/src/snapshot/snapshot.test.ts @@ -739,3 +739,13 @@ test("each state reads back with its frame's timestamp", async () => { expect(head?.value.n).toBe(3); expect(head?.at?.as(Time.Timescale.MILLI)).toBe(3_000); }); + +// Mirrors the Rust `snapshot::Finished` doctest: the value `finish` returns takes no update, at +// compile time and at run time. +test("finishing returns a value that publishes nothing", () => { + const producer = new Producer({ track: new Track.Producer("test") }); + producer.update({ value: { a: 1 } }); + const finished = producer.finish(); + // @ts-expect-error a finished value takes no update + expect(() => finished.update({ value: { a: 2 } })).toThrow(); +}); diff --git a/js/json/src/stream/index.ts b/js/json/src/stream/index.ts index 4915c0bbb68..146f6f19a0f 100644 --- a/js/json/src/stream/index.ts +++ b/js/json/src/stream/index.ts @@ -23,4 +23,4 @@ export { Consumer, Rolled } from "./consumer.ts"; export { Decoder } from "./decoder.ts"; export { type Config, Encoder, type Pending } from "./encoder.ts"; -export { Producer } from "./producer.ts"; +export { type Finished, Producer } from "./producer.ts"; diff --git a/js/json/src/stream/producer.ts b/js/json/src/stream/producer.ts index 5987826425a..c74a4413b70 100644 --- a/js/json/src/stream/producer.ts +++ b/js/json/src/stream/producer.ts @@ -3,6 +3,14 @@ import { Error as NetError, type Timed } from "@moq/net"; import { type Config as CodecConfig, Encoder, type Pending } from "./encoder.ts"; +/** + * A log its producer finished: it takes no more records. {@link Producer.finish} returns it, + * mirroring the Rust `stream::Finished`; the producer itself still refuses an append after the + * finish. + */ +// biome-ignore lint/suspicious/noEmptyInterface: nothing is left to call once the track is released. +export interface Finished {} + /** * Publishes an ordered log of JSON records to a track, one record per frame in a single group. * @@ -76,12 +84,13 @@ export class Producer { this.#track.abort(abort); } - /** Finish the track, closing the group. */ - finish(): void { + /** Finish the track, closing the group, and return it as {@link Finished}. */ + finish(): Finished { this.#group?.finish(); this.#group = undefined; this.#track.finish(); this.#track.close(); + return this; } } diff --git a/js/json/src/stream/stream.test.ts b/js/json/src/stream/stream.test.ts index 70f05ef37b8..a80cb0e2cec 100644 --- a/js/json/src/stream/stream.test.ts +++ b/js/json/src/stream/stream.test.ts @@ -250,3 +250,15 @@ for (const compress of [false, true]) { expect(await fast.next()).toBeUndefined(); }); } + +// Mirrors the Rust `stream::Finished` doctest: the log `finish` returns takes no record, at +// compile time and at run time. +test("finishing returns a log that appends nothing", async () => { + const track = new Track.Producer("test"); + const producer = new Producer({ track }); + producer.append({ value: { n: 0 } }); + const finished = producer.finish(); + // @ts-expect-error a finished log appends no record + expect(() => finished.append({ value: { n: 1 } })).toThrow(); + expect(await drain(track.subscribe())).toEqual([0]); +}); diff --git a/js/json/src/window/index.ts b/js/json/src/window/index.ts index 3dce0f6f663..05c267a7982 100644 --- a/js/json/src/window/index.ts +++ b/js/json/src/window/index.ts @@ -39,4 +39,4 @@ export { Consumer, type ConsumerConfig } from "./consumer.ts"; export { Decoder, type Event, type Group, type Span } from "./decoder.ts"; export { type Config, type Encoded, Encoder, type Pending } from "./encoder.ts"; -export { Producer, type ProducerConfig } from "./producer.ts"; +export { type Finished, Producer, type ProducerConfig } from "./producer.ts"; diff --git a/js/json/src/window/producer.ts b/js/json/src/window/producer.ts index f0630fd701b..1b7be001787 100644 --- a/js/json/src/window/producer.ts +++ b/js/json/src/window/producer.ts @@ -6,6 +6,22 @@ import { type Config, type Encoded, Encoder } from "./encoder.ts"; /** Window producer options, including the destination track. */ export type ProducerConfig = Config & { track: Moq.Track.Producer }; +/** + * A window its producer finished: its records can still be read, but not edited. + * {@link Producer.finish} returns it, mirroring the Rust `window::Finished`; the producer itself + * still refuses an edit after the finish. + * + * @public + */ +export interface Finished { + /** The retained checkpoint suffix, oldest first; see {@link Producer.window}. */ + readonly window: T[]; + /** Absolute index of the oldest retained record; see {@link Producer.offset}. */ + readonly offset: number; + /** Abort the track with an error; see {@link Producer.abort}. */ + abort(err: Error): void; +} + /** * Publishes a sliding window of JSON records to a track. * @@ -156,9 +172,10 @@ export class Producer { group.finish(); } - /** Finish the track, closing any open group. */ - finish(): void { + /** Finish the track, closing any open group, and return it as {@link Finished}. */ + finish(): Finished { this.#close(); + return this; } /** Abort the track with `err`, so readers see the cause rather than a complete window. Idempotent. */ diff --git a/js/json/src/window/window.test.ts b/js/json/src/window/window.test.ts index 10cc0bfdca8..c91204ca030 100644 --- a/js/json/src/window/window.test.ts +++ b/js/json/src/window/window.test.ts @@ -614,3 +614,15 @@ test("events on an untimed track have no timestamp", async () => { expect(events.length).toBe(2); expect(events.every((event) => event.at === undefined)).toBeTrue(); }); + +// Mirrors the Rust `window::Finished` doctest: the window `finish` returns still reads, but takes +// no edit, at compile time and at run time. +test("finishing returns a window that reads but takes no edit", () => { + const producer = new Producer({ track: new Track.Producer("test") }); + producer.push({ value: { n: 1 } }); + const finished = producer.finish(); + expect(finished.window).toEqual([{ n: 1 }]); + expect(finished.offset).toBe(0); + // @ts-expect-error a finished window takes no record + expect(() => finished.push({ value: { n: 2 } })).toThrow("track is closed"); +}); diff --git a/quest/m2/README.md b/quest/m2/README.md index 017490c24d6..9a2c9848ee8 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -203,7 +203,6 @@ quests that require them. - [Finalize moq-lite-07](/quest/m2/lite07-finalize.md) - when the maintainer cuts it, lite-07 negotiates as `moq-lite-07` and the next release ships it - [Unbounded groups](/quest/m2/unbounded-groups/README.md) - groups and frames lose their size caps; a group keeps a 32 MB tail, and readers behind it replay through paced flights or refresh - [Small API fixes before the cut](/quest/m2/api-shape/README.md) - one path and one spelling per item, names by role, and no dead or wrong-role public surface across net, transport, media, native, apps, JS, and bindings -- [Finishing a media producer returns a Finished handle](/quest/m2/finished-handle.md) - `finish(self)` consumes the media crates' producers, as it does moq-net's, so appending after finish doesn't compile - [Probe](/quest/m2/probe/README.md) - a session probes up to the bitrate it wants to send or receive, so a stopped simulcast rendition resumes and a viewer up-switches; browser padding is parked in m3 (held in m2 by open claims; moves to m3) - [Finish the milestone re-cut](/quest/m2/recut-held.md) - claim-held epics move to the milestone their kind and priority give them once their claims merge - [A watcher reads connecting, not offline, while its connection is down](/quest/m2/watch-connecting-status.md) - `@moq/watch` status says OFFLINE only when connected and unannounced diff --git a/quest/m2/finished-handle.md b/quest/m2/finished-handle.md deleted file mode 100644 index 71cde4738b0..00000000000 --- a/quest/m2/finished-handle.md +++ /dev/null @@ -1,74 +0,0 @@ -# [M] Finishing a media producer returns a Finished handle - -## Goal - -The media producers and importers take `finish(self)` and return a -`Finished` handle, as moq-net's `track::Producer` and `group::Producer` -already do, so appending through the handle that finished is a compile error -instead of a runtime one. The handle keeps what finishing still allows: -aborting, which reports a later failure and releases the cache, and, where -the type supports it today, inserting below the end. It has no append. The -types are #8210's: moq-mux's container producer and the MKV, fMP4, FLV, and -TS importers, moq-audio's producer, and moq-json's `snapshot`, `stream`, and -`window` producers, with their `js/json` and `js/hang` mirrors. moq-rtmp and -moq-srt already consume on finish. - -The compile-time guarantee covers only the consumed handle, and only in -Rust. A `Clone` producer that shares its state (such as moq-json's -producers and moq-mux's `catalog::Producer`) leaves other clones that can -still append, so those clones keep a shared runtime refusal after any clone -finishes. The JS mirrors, moq-ffi, and moq-c can't consume a handle, so they -keep a runtime guard that refuses an append after finish. - -## Plan - -Decided 2026-11-07 (maintainer, /quest-complete), a follow-up of -[moq-dev/moq#8210](https://github.com/moq-dev/moq/pull/8210), which made a -failed finish final with explicit state and recommended against -`finish(self)`. The maintainer chose the typestate: - -- ✅ **Scope: moq-net and media.** Not chosen: moq-net only; media only. -- ✅ **Name: `Finished`**, as `track::Finished`, `group::Finished`, and so - on. -- ✅ **Settle it in m2 before the cut** (maintainer, /quest-plan from the - API sweep, 2026-11-07; rejected: shipping `finish(&mut self)` plus a - refusal, then breaking after the cut). - -[moq-dev/moq#8276](https://github.com/moq-dev/moq/pull/8276) landed the -moq-net layer and its `js/net` mirror. Follow its shape: - -- A `Finished` wraps the producer (or its shared handle) privately and - exposes only what finishing still allows. `track::Finished` keeps - `create_group` below the end, `abort`, `poll_closed`, `consume`, and - `subscribe`; `group::Finished` keeps `abort`, `frame_count`, and - `consume`. -- Where dropping the last handle releases something readers still need, - as dropping the last `track::Finished` unlists the track, mark the - `Finished` `#[must_use]` and say so in its docs. -- A shared holder that must keep its producer (a session, a cloned output) - finishes a clone, or a crate-private `finish_shared(&self)`, and keeps - its own clone, whose appends the end refuses. -- A `compile_fail,E0599` doctest pins each missing append. -- JS: `finish()` returns `this` typed as a narrowing `Finished` interface; - the runtime guard stays on the producer. - -Decide per type what `Finished` exposes beyond abort, such as read access to -the cache or insert below the end, without adding anything the producer -can't do today. A `finish` that can fail returns the producer in its error -when the caller can retry, and the PR says which do. None in moq-net could: -a failed finish there is final. - -This breaks every caller that finishes and then keeps using the producer. -List them in the PR, with every changed signature and the migration for each -package in its Impact section, including the runtime guards kept in -moq-ffi, moq-c, and the JS mirrors. - -Public API: breaking in moq-mux, moq-audio, moq-json, `@moq/json`, and -`@moq/hang`; moq-ffi and moq-c keep their signatures and runtime checks. -Wire: none. - -## Related - -- [moq-audio codecs flush](https://github.com/moq-dev/moq/pull/8270) - rewrites `moq-audio`'s `encode/producer.rs`, whose `finish` this changes -- [Mux lifecycles hold one phase each](https://github.com/moq-dev/moq/pull/8211) - rewrites moq-mux's `catalog::Producer` lifecycle -- [A failed finish is final](https://github.com/moq-dev/moq/pull/8210) - the explicit state this builds on diff --git a/rs/moq-audio/benches/batches.rs b/rs/moq-audio/benches/batches.rs index 10f36902de2..275959d6fcc 100644 --- a/rs/moq-audio/benches/batches.rs +++ b/rs/moq-audio/benches/batches.rs @@ -23,7 +23,7 @@ use std::time::{Duration, Instant}; use bytes::Bytes; use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group}; -use moq_audio::encode::{Codec, Config, Input, Options, Producer}; +use moq_audio::encode::{Codec, Config, Finished, Input, Options, Producer}; use moq_audio::{Frame, Layout}; use moq_net::Timestamp; @@ -205,7 +205,7 @@ impl Case { } /// Time individual writes only in the separate latency pass, outside instruction and throughput measurements. -fn run(producers: &mut [Producer], input: &[Frame]) -> Duration { +fn run(mut producers: Vec, input: &[Frame]) -> (Duration, Vec) { let mut slowest = Duration::ZERO; for frame in input { for producer in producers.iter_mut() { @@ -216,10 +216,11 @@ fn run(producers: &mut [Producer], input: &[Frame]) -> Durati } } } - for producer in producers.iter_mut() { - producer.finish().unwrap(); - } - slowest + let finished = producers + .into_iter() + .map(|producer| producer.finish().unwrap()) + .collect(); + (slowest, finished) } fn cases() -> Vec { @@ -269,20 +270,20 @@ fn counts() { ); for case in cases() { let input = case.input(); - let (broadcast, mut producers) = case.producers(); + let (broadcast, producers) = case.producers(); ALLOCATIONS.store(0, Ordering::Relaxed); ALLOCATED.store(0, Ordering::Relaxed); COUNTING.store(true, Ordering::Relaxed); if let Some(counter) = &mut counter { counter.start(); } - run::(&mut producers, &input); + let (_, finished) = run::(producers, &input); let instructions = counter.as_mut().map(instructions::Counter::stop); COUNTING.store(false, Ordering::Relaxed); - drop((producers, broadcast)); - let (broadcast, mut producers) = case.producers(); - let slowest = run::(&mut producers, &input); - drop((producers, broadcast)); + drop((finished, broadcast)); + let (broadcast, producers) = case.producers(); + let (slowest, finished) = run::(producers, &input); + drop((finished, broadcast)); let per_sample = match instructions { Some(instructions) => format!("{:.1}", instructions as f64 / case.samples() as f64), @@ -309,9 +310,9 @@ fn bench_batches(c: &mut Criterion) { group.bench_function(BenchmarkId::from_parameter(case.name()), |b| { b.iter_batched( || case.producers(), - |(broadcast, mut producers)| { - run::(&mut producers, &input); - (broadcast, producers) + |(broadcast, producers)| { + let (_, finished) = run::(producers, &input); + (broadcast, finished) }, BatchSize::PerIteration, ); diff --git a/rs/moq-audio/src/decode/conformance.rs b/rs/moq-audio/src/decode/conformance.rs index 77a2f0ea01d..f367a02fb17 100644 --- a/rs/moq-audio/src/decode/conformance.rs +++ b/rs/moq-audio/src/decode/conformance.rs @@ -149,14 +149,12 @@ fn import( Container::Ts(bytes) => { let mut importer = moq_mux::container::ts::Import::new(broadcast, catalog.reserve()); importer.decode(bytes).unwrap(); - importer.finish().unwrap(); - Box::new(importer) + Box::new(importer.finish().unwrap()) } Container::Cmaf(bytes) => { let mut importer = moq_mux::container::fmp4::Import::new(broadcast, catalog.reserve()); importer.decode(bytes).unwrap(); - importer.finish().unwrap(); - Box::new(importer) + Box::new(importer.finish().unwrap()) } }; let snapshot = catalog.snapshot(); diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index 95ccb49848d..ba19caebcc2 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -913,7 +913,7 @@ mod tests { producer.cut(None).unwrap(); } } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut consumer = Consumer::new( &broadcast.consume(), &catalog, @@ -1130,7 +1130,7 @@ mod tests { for packet in 1..=20 { write(packet); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut stamps = Vec::new(); while let Some(read) = consumer.read().await.unwrap() { @@ -1190,7 +1190,7 @@ mod tests { } producer.cut(None).unwrap(); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, 48_000, 1); catalog.warmup = Some(Duration::from_millis(80)); @@ -1243,7 +1243,7 @@ mod tests { keyframe: true, }) .unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut consumer = Consumer::new( &broadcast.consume(), &catalog, @@ -1299,7 +1299,7 @@ mod tests { }) .unwrap(); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut consumer = Consumer::new( &broadcast.consume(), @@ -1365,7 +1365,7 @@ mod tests { }) .unwrap(); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut consumer = Consumer::new( &broadcast.consume(), &catalog, @@ -1433,7 +1433,7 @@ mod tests { }) .unwrap(); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut stamps = Vec::new(); while let Some(frame) = next_frame(&mut consumer).await.unwrap() { stamps.push(frame.timestamp.as_micros()); @@ -1572,7 +1572,7 @@ mod tests { }) .unwrap(); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut reads = Vec::new(); while let Some(event) = consumer.read().await.unwrap() { reads.push(match event { @@ -1629,7 +1629,7 @@ mod tests { producer.discontinuity().unwrap(); } } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut before = Vec::new(); let mut ended = None; let mut after = Vec::new(); @@ -1766,7 +1766,7 @@ mod tests { keyframe: true, }) .unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let first = next_frame(&mut consumer).await.unwrap().expect("decoded frame"); let first_frames = first.data.len() / size_of::(); @@ -1850,7 +1850,7 @@ mod tests { .unwrap(); producer.cut(None).unwrap(); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let expected = first_dtx.expect("silence should enter Opus DTX"); let mut actual = None; @@ -1928,7 +1928,7 @@ mod tests { }) .unwrap(); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut read = Vec::new(); while let Some(frame) = next_frame(&mut consumer).await.unwrap() { @@ -2079,7 +2079,7 @@ mod tests { // Same group: a new group at 1s would sit below the marker's live edge. write(&mut producer, 3 * 48_000, Bytes::new(), false); write(&mut producer, 48_000, dtx.payload, false); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let frame = next_frame(&mut consumer).await.unwrap().expect("decoded frame"); // The output begins with the first packet's samples, so it is stamped and @@ -2567,7 +2567,7 @@ mod tests { 2, crate::single(encoder.encode_all(&pcm).unwrap()).payload, ); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let first = next_frame(&mut consumer).await.unwrap().expect("first decoded frame"); let _second = next_frame(&mut consumer).await.unwrap().expect("second decoded frame"); diff --git a/rs/moq-audio/src/encode/capture.rs b/rs/moq-audio/src/encode/capture.rs index 3a62c0c46d5..acfaff02e84 100644 --- a/rs/moq-audio/src/encode/capture.rs +++ b/rs/moq-audio/src/encode/capture.rs @@ -285,33 +285,44 @@ impl Driver { let end = self.drive().await; let mut producer = self.producer.take().unwrap(); let now = self.clock.now(); + // Held until the state closes, so the track outlives its terminal status. + let mut finished = None; let result = match end { Ok(End::Failed(failure)) => { // Retention already reports the failure. It stays the cause even when the track // can no longer end cleanly, such as one aborted while it still retained audio. - let ended = producer - .set_active(false, now) - .and_then(|()| producer.fail(moq_net::Error::Transport(failure.to_string()))); - if ended.is_err() { - producer.abort(moq_net::Error::Transport(failure.to_string())); + let cause = moq_net::Error::Transport(failure.to_string()); + match producer.set_active(false, now) { + Ok(()) => finished = producer.fail(cause).ok(), + Err(_) => producer.abort(cause), } Err(failure) } // The track closed, or encoding or publishing failed. - end => match end.and_then(|_| producer.set_active(false, now).and_then(|()| producer.finish())) { - Ok(()) => { - self.update(Status::Ended); - Ok(()) - } - Err(err) => { - self.update(Status::Failed(err.clone())); - producer.abort(moq_net::Error::Transport(err.to_string())); - Err(err) + end => { + let ended = match end.and_then(|_| producer.set_active(false, now)) { + // A finish that fails ends the track with an error itself. + Ok(()) => producer.finish().map(|track| finished = Some(track)), + Err(err) => { + producer.abort(moq_net::Error::Transport(err.to_string())); + Err(err) + } + }; + match ended { + Ok(()) => { + self.update(Status::Ended); + Ok(()) + } + Err(err) => { + self.update(Status::Failed(err.clone())); + Err(err) + } } - }, + } }; self.publish_level(Level::default()); let _ = self.state.close(); + drop(finished); result } diff --git a/rs/moq-audio/src/encode/mod.rs b/rs/moq-audio/src/encode/mod.rs index b4342eb408d..f6c62c8a511 100644 --- a/rs/moq-audio/src/encode/mod.rs +++ b/rs/moq-audio/src/encode/mod.rs @@ -28,7 +28,7 @@ mod capture; pub use encoded::Encoded; pub use encoder::{Codec, Config, Encoder, Finish, Input, Kind}; -pub use producer::{Options, Producer}; +pub use producer::{Finished, Options, Producer}; #[cfg(feature = "capture")] pub use capture::{Capture, Driver, Publication, State, Status, publish_capture}; diff --git a/rs/moq-audio/src/encode/producer.rs b/rs/moq-audio/src/encode/producer.rs index a82d2f061b5..8770f5f94bd 100644 --- a/rs/moq-audio/src/encode/producer.rs +++ b/rs/moq-audio/src/encode/producer.rs @@ -78,9 +78,10 @@ impl Default for Options { /// a subscriber that opens the catalog before any frames arrive still sees the /// track. pub struct Producer { - /// The codec while input is accepted. Ending keeps the track and catalog rendition, not the - /// codec. - codec: Codec, + encoder: Encoder, + /// A publish that failed after the codec advanced past its packet, which ended the track with an + /// error: every later call reports the cause rather than stamping what follows. + failed: Option, bitrate: moq_net::bandwidth::Rate, input: Input, /// Converts the input layout to the codec's, when they differ. @@ -113,33 +114,6 @@ pub struct Producer { activity: Activity, } -/// The codec while the producer accepts input, or how it ended. -enum Codec { - Live(Encoder), - /// [`finish`](Producer::finish) published every packet and ended the track. - Finished, - /// Ending failed, aborting the track: every later call reports the cause. - Failed(Error), -} - -impl Codec { - fn get(&self) -> Result<&Encoder, Error> { - match self { - Self::Live(encoder) => Ok(encoder), - Self::Finished => Err(moq_net::Error::Closed.into()), - Self::Failed(err) => Err(err.clone()), - } - } - - fn get_mut(&mut self) -> Result<&mut Encoder, Error> { - match self { - Self::Live(encoder) => Ok(encoder), - Self::Finished => Err(moq_net::Error::Closed.into()), - Self::Failed(err) => Err(err.clone()), - } - } -} - struct Terminal { packets: Vec, end: Timestamp, @@ -249,7 +223,8 @@ impl Reserved { active: true, reset_pending: false, bitrate, - codec: Codec::Live(registered.encoder), + encoder: registered.encoder, + failed: None, input: registered.input, remix: registered.remix, resampler: registered.resampler, @@ -299,6 +274,23 @@ impl Producer { } } + /// The cause of an earlier publish failure, which ended the track. + fn check(&self) -> Result<(), Error> { + self.failed.clone().map_or(Ok(()), Err) + } + + /// End the track with an error past everything published, so subscribers stop waiting on it + /// while this producer is still held. Readers keep every packet published before it. + fn fail_track(&mut self) { + let _ = self.track.cut(None); + let track = self.track.track(); + let next = track.largest().map_or(0, |largest| largest.group.saturating_add(1)); + let _ = track.clone().finish_at( + moq_net::track::Position::group(next), + Some(moq_net::StreamError::Internal.into()), + ); + } + /// The name of the published track, which is [`Options::track`] resolved. pub fn track_name(&self) -> &str { self.track.name() @@ -335,7 +327,8 @@ impl Producer { /// Retune the live encoder to `bitrate`. pub fn set_bitrate(&mut self, bitrate: moq_net::bandwidth::Rate) -> Result<(), Error> { - let encoder = self.codec.get_mut()?; + self.check()?; + let encoder = &mut self.encoder; let previous = encoder.bitrate(); encoder.set_bitrate(bitrate)?; // Advertise the configured target in place, so encoded-byte estimates cannot replace it. @@ -350,10 +343,6 @@ impl Producer { Ok(()) } - fn encoder(&self) -> Result<&Encoder, Error> { - self.codec.get() - } - /// Re-anchor the timeline to the next frame's timestamp, dropping any /// partial input frame. Call this when resuming after an idle gap (e.g. a /// released-then-reopened microphone) so the gap appears in the PTS and @@ -362,7 +351,7 @@ impl Producer { /// `reanchor`. If the codec had started, a marker group is published before /// the next packet so subscribers jump the playhead. pub fn reanchor(&mut self) { - if self.codec.get().is_ok_and(Encoder::started) || !self.backlog.is_empty() { + if self.encoder.started() || !self.backlog.is_empty() { self.pending_discontinuity = true; } self.reset_state(); @@ -374,10 +363,8 @@ impl Producer { self.frames_produced = 0; self.pending.clear(); self.reset_pending = true; - if self.backlog.is_empty() - && let Ok(encoder) = self.codec.get_mut() - { - encoder.reset(); + if self.backlog.is_empty() { + self.encoder.reset(); self.in_flight.clear(); } // The resampler holds samples of its own, plus filter state primed by them. @@ -403,11 +390,8 @@ impl Producer { /// /// [`Frame::activity`] is ignored: the encoder classifies what it actually /// produced, which [`activity`](Self::activity) reports. - /// - /// Writes after a successful [`finish`](Self::finish) fail with - /// [`moq_net::Error::Closed`], and after a failed one with its cause. pub fn write(&mut self, frame: &Frame) -> Result<(), Error> { - self.encoder()?; + self.check()?; let timestamp_us = u64::try_from(frame.timestamp.as_micros()) .map_err(|_| Error::Unsupported(format!("frame timestamp {:?} out of range", frame.timestamp)))?; let anchor_us = *self.anchor_us.get_or_insert(timestamp_us); @@ -435,7 +419,7 @@ impl Producer { /// Frames are read in place and the remainder moves to the front once, so a /// large batch costs linear rather than quadratic copying. fn publish_full_frames(&mut self, anchor_us: u64) -> Result<(), Error> { - let frame_samples = self.encoder()?.frame_size() * self.encoder()?.codec_channels() as usize; + let frame_samples = self.encoder.frame_size() * self.encoder.codec_channels() as usize; // Taken so each frame can be borrowed while `self` encodes it; restored below on every path. let mut pending = std::mem::take(&mut self.pending); let mut consumed = 0; @@ -456,11 +440,11 @@ impl Producer { let timestamp = Self::timestamp( anchor_us, self.frames_produced, - self.encoder()?.folded_delay(), - self.encoder()?.codec_rate(), + self.encoder.folded_delay(), + self.encoder.codec_rate(), )?; - self.frames_produced += self.encoder()?.frame_size() as u64; - let end = Self::timestamp(anchor_us, self.frames_produced, 0, self.encoder()?.codec_rate())?; + self.frames_produced += self.encoder.frame_size() as u64; + let end = Self::timestamp(anchor_us, self.frames_produced, 0, self.encoder.codec_rate())?; let reset = std::mem::take(&mut self.reset_pending); let discontinuity = std::mem::take(&mut self.pending_discontinuity); if self.active && self.backlog.is_empty() { @@ -484,13 +468,14 @@ impl Producer { reset: bool, discontinuity: bool, ) -> Result<(), Error> { + self.check()?; // A break before anything was published, such as history expiring before the first // subscriber, has nothing to separate. if discontinuity && !self.decoder_boundary { self.track.discontinuity()?; self.decoder_boundary = true; } - let encoder = self.codec.get_mut()?; + let encoder = &mut self.encoder; if reset { encoder.reset(); self.in_flight.clear(); @@ -503,18 +488,20 @@ impl Producer { /// the oldest held stamp, then `submitted`, the frame just encoded, which is queued only /// when the codec holds its packet, so a codec that holds nothing never fills the queue. fn publish_ready(&mut self, mut submitted: Option) -> Result<(), Error> { - while let Some(packet) = self.codec.get_mut()?.receive()? { + while let Some(packet) = self.encoder.receive()? { let Some(timestamp) = self.in_flight.pop_front().or_else(|| submitted.take()) else { return Err(Error::Codec(format!( "{} returned more packets than it was given frames", - self.encoder()?.name() + self.encoder.name() ))); }; self.activity = packet.activity; // The codec and its timestamp queue already advanced past this packet, so no later // write could stamp what follows: end the producer with the cause. if let Err(err) = self.publish(packet, timestamp) { - return self.close(Err(err)); + self.fail_track(); + self.failed = Some(err.clone()); + return Err(err); } self.decoder_boundary = false; } @@ -556,9 +543,9 @@ impl Producer { if !self.backlog.is_empty() { return Ok(()); } - let channels = self.encoder()?.codec_channels() as usize; + let channels = self.encoder.codec_channels() as usize; let frames = self.frames_produced + (self.pending.len() / channels) as u64; - let end = Self::timestamp(anchor_us, frames, 0, self.encoder()?.codec_rate())?; + let end = Self::timestamp(anchor_us, frames, 0, self.encoder.codec_rate())?; if self.backlog.expired(end, now) { self.reanchor(); } @@ -609,7 +596,7 @@ impl Producer { duration: None, }; self.track.write(mux_frame)?; - self.grouped += self.encoder()?.frame_size() as u64; + self.grouped += self.encoder.frame_size() as u64; // Cut as soon as the group is long enough rather than when the next packet arrives, so // the relay sees it finish without waiting. No boundary to give: the next packet bounds // this one, and codec frames have a deterministic duration anyway. @@ -664,65 +651,47 @@ impl Producer { /// Flush pending samples, resampler output, and codec lookahead, then finalize /// the track. /// - /// Borrows rather than consumes, so a later [`abort`](Self::abort) can still - /// run after a successful finish. The codec and input scratch are released, while the - /// track and its catalog rendition stay published. Writes and bitrate changes then - /// fail with [`moq_net::Error::Closed`]. A finish that fails ends the track with an error - /// instead, and every later call returns its cause. - pub fn finish(&mut self) -> Result<(), Error> { + /// Consumes the producer, releasing the codec and input scratch. The returned [`Finished`] + /// keeps the track and its catalog rendition published, and can still + /// [`abort`](Finished::abort) them. A finish that fails ends the track with an error instead. + pub fn finish(self) -> Result { self.end(None) } /// [`finish`](Self::finish) the track with `err`, for a source that failed: subscribers keep /// every packet published, then get `err` instead of the end of stream. #[cfg(feature = "capture")] - pub(crate) fn fail(&mut self, err: moq_net::Error) -> Result<(), Error> { + pub(crate) fn fail(self, err: moq_net::Error) -> Result { self.end(Some(err)) } - fn end(&mut self, err: Option) -> Result<(), Error> { - match &self.codec { - Codec::Live(_) => {} - Codec::Finished => return Ok(()), - Codec::Failed(cause) => return Err(cause.clone()), - } - let result = self.drain(err); - self.close(result) - } - - /// Settle the codec as finished or failed with `result`, releasing encoding resources. - fn close(&mut self, result: Result<(), Error>) -> Result<(), Error> { - if !matches!(self.codec, Codec::Live(_)) { - return result; + /// Publish everything still held, then end the track with `err` or cleanly. + fn end(mut self, err: Option) -> Result { + if let Err(cause) = self.check().and_then(|()| self.drain()) { + // Every published packet stays readable, then subscribers get an error rather than an + // end that would claim the source was published whole: `err` when the source failed. + let _ = self + .track + .fail(err.unwrap_or_else(|| moq_net::StreamError::Internal.into())); + return Err(cause); } - self.codec = match &result { - Ok(()) => Codec::Finished, - Err(cause) => { - // Every published packet stays readable, then subscribers get an error rather than an - // end that would claim the source was published whole. - let _ = self.track.fail(moq_net::StreamError::Internal.into()); - Codec::Failed(cause.clone()) - } + let track = match err { + None => self.track.finish()?, + Some(err) => self.track.fail(err)?, }; - self.pending = Vec::new(); - self.in_flight = VecDeque::new(); - self.remix = None; - self.resampler = None; - result + Ok(Finished { track }) } - /// Publish everything still held, then end the track with `err` or cleanly. - fn drain(&mut self, err: Option) -> Result<(), Error> { + /// Publish everything the codec, resampler, and backlog still hold. + fn drain(&mut self) -> Result<(), Error> { // Preserve the final retained history after capture ends, even without live demand. self.active = true; while let Some(raw) = self.backlog.pop() { self.track.history(raw.timestamp); self.encode_frame(&raw.samples, raw.timestamp, raw.reset, raw.discontinuity)?; } - if self.reset_pending - && let Ok(encoder) = self.codec.get_mut() - { - encoder.reset(); + if self.reset_pending { + self.encoder.reset(); self.in_flight.clear(); } // Whatever the resampler still holds belongs to this track: its last partial @@ -738,12 +707,12 @@ impl Producer { self.publish_full_frames(anchor_us)?; // What the codec still holds came from full frames, so it goes out at their stamps, leaving // the terminal drain's packets to start where the source's partial tail does. - self.codec.get_mut()?.flush()?; + self.encoder.flush()?; self.publish_ready(None)?; if !self.in_flight.is_empty() { return Err(Error::Codec(format!( "{} kept {} packet(s) past its flush", - self.encoder()?.name(), + self.encoder.name(), self.in_flight.len() ))); } @@ -754,15 +723,15 @@ impl Producer { self.decoder_boundary = true; } - let frame_size = self.encoder()?.frame_size(); - let codec_rate = self.encoder()?.codec_rate(); - let channels = self.encoder()?.codec_channels() as usize; + let frame_size = self.encoder.frame_size(); + let codec_rate = self.encoder.codec_rate(); + let channels = self.encoder.codec_channels() as usize; let source_frames = self.pending.len() / channels; - let delay = self.encoder()?.folded_delay(); + let delay = self.encoder.folded_delay(); let start = Self::timestamp(anchor_us, self.frames_produced, delay, codec_rate)?; // The source ends where it ends: priming only moves the packets carrying it. let end = Self::timestamp(anchor_us, self.frames_produced + source_frames as u64, 0, codec_rate)?; - let finish = self.codec.get_mut()?.drain(&self.pending)?; + let finish = self.encoder.drain(&self.pending)?; let discard_padding = finish.discard_padding(); let packets = finish.into_packets(); @@ -785,18 +754,44 @@ impl Producer { self.frames_produced += frame_size as u64; } } - - match err { - None => self.track.finish()?, - Some(err) => self.track.fail(err)?, - } Ok(()) } /// Abort the track with `err` instead of finishing it, so subscribers see the /// real cause rather than [`moq_net::Error::Dropped`]. Pending samples are dropped. - /// - /// Consumes the producer. Still callable after [`finish`](Self::finish). + pub fn abort(self, err: moq_net::Error) { + self.track.abort(err); + } +} + +/// An audio track whose [`Producer`] [finished](Producer::finish) it: every packet is published, +/// and it can still be watched or aborted, but not written to. +/// +/// Dropping it releases the track like dropping a [`Producer`]. +/// +/// ```compile_fail,E0599 +/// # fn cannot_write(track: moq_audio::encode::Finished, frame: &moq_audio::Frame) { +/// track.write(frame).unwrap(); +/// # } +/// ``` +#[must_use = "dropping it releases the track and its catalog rendition"] +pub struct Finished { + track: moq_mux::container::Finished, +} + +impl Finished { + /// The name of the published track, which is [`Options::track`] resolved. + pub fn track_name(&self) -> &str { + self.track.name() + } + + /// A watch-only handle to the track's subscriber demand; see [`Producer::demand`]. + pub fn demand(&self) -> moq_net::track::Demand { + self.track.demand() + } + + /// Abort the track with `err`, so subscribers that have not drained it see the real cause; see + /// [`Producer::abort`]. pub fn abort(self, err: moq_net::Error) { self.track.abort(err); } @@ -901,9 +896,10 @@ mod tests { idle.as_secs_f64() * 1e3, catchup.as_secs_f64() * 1e3 ); - for producer in &mut producers { - producer.finish().unwrap(); - } + let _finished: Vec<_> = producers + .into_iter() + .map(|producer| producer.finish().unwrap()) + .collect(); } } } @@ -939,7 +935,7 @@ mod tests { )) .unwrap(); } - assert!(!producer.encoder().unwrap().started(), "no codec work without demand"); + assert!(!producer.encoder.started(), "no codec work without demand"); let track = consumer .track("audio") .unwrap() @@ -979,7 +975,7 @@ mod tests { .write(&pcm_frame(&vec![0.5; 1920], at(151).as_micros() as u64)) .unwrap(); assert_eq!(packet(&mut reader).await.unwrap().timestamp, at(151)); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); } /// Replayed groups reach the track as fresh arrivals, so only the encoder can tell they are old: @@ -1022,7 +1018,7 @@ mod tests { panic!("expected a replayed frame"); }; assert_eq!(frame.timestamp, at(143)); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); } /// Replay catches up behind the live edge another track already reached; advertising that as @@ -1061,13 +1057,13 @@ mod tests { .write(&pcm_frame(&vec![0.1; 1920], at(index).as_micros() as u64)) .unwrap(); } - producer.finish().unwrap(); + let producer = producer.finish().unwrap(); let delay = producer.track.estimate().delay; assert!( delay.is_none_or(|delay| delay < Duration::from_millis(100)), "{delay:?}" ); - other.finish().unwrap(); + let _finished = other.finish().unwrap(); } /// History expiring right before capture ends still marks the gap before the final partial frame. @@ -1105,7 +1101,7 @@ mod tests { producer .set_active(false, Timestamp::from_micros(31_045_000).unwrap()) .unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let first = reader.read().await.unwrap().unwrap(); assert!(matches!(first, moq_mux::container::Read::Frame { .. }), "{first:?}"); @@ -1150,7 +1146,7 @@ mod tests { producer .set_active(false, Timestamp::from_micros(31_100_000).unwrap()) .unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let first = reader.read().await.unwrap().unwrap(); assert!( @@ -1181,8 +1177,8 @@ mod tests { .set_active(true, Timestamp::from_micros(30_020_000).unwrap()) .unwrap(); assert!(producer.backlog.is_empty()); - assert!(!producer.encoder().unwrap().started()); - producer.finish().unwrap(); + assert!(!producer.encoder.started()); + let _finished = producer.finish().unwrap(); } #[tokio::test] @@ -1194,7 +1190,7 @@ mod tests { track: Some("audio".into()), ..Options::default() }; - let mut producer = Producer::new(&mut broadcast, catalog, Input::default(), &options).unwrap(); + let producer = Producer::new(&mut broadcast, catalog, Input::default(), &options).unwrap(); let mut demand = producer.demand(); assert_eq!(demand.name(), "audio"); @@ -1222,8 +1218,7 @@ mod tests { .unwrap(); assert_eq!(demand.wanted(), None); - producer.finish().unwrap(); - drop(producer); + drop(producer.finish().unwrap()); let closed = kio::clock::timeout(Duration::from_secs(1), demand.closed()) .await .expect("producer closed"); @@ -1255,20 +1250,8 @@ mod tests { producer.write(&pcm_frame(&vec![0.1; 4096], 0)).unwrap(); assert!(producer.pending.capacity() > 0); assert!(producer.remix.is_some()); - let bitrate = producer.bitrate(); - producer.finish().unwrap(); + retired.push(producer.finish().unwrap()); assert_eq!(backend.resources(), 0, "retired {index} codecs"); - assert_eq!(producer.pending.capacity(), 0); - assert_eq!(producer.in_flight.capacity(), 0); - assert!(producer.remix.is_none()); - assert!(producer.resampler.is_none()); - assert_eq!(producer.bitrate(), bitrate); - producer.finish().unwrap(); - assert!(matches!( - producer.set_bitrate(bitrate), - Err(Error::Net(moq_net::Error::Closed)) - )); - retired.push(producer); // Retention still owns both the published track and its catalog rendition. let mut reader = consumer @@ -1337,7 +1320,7 @@ mod tests { pcm[impulse] = 1.0; let data: Vec = pcm.iter().flat_map(|sample| sample.to_le_bytes()).collect(); producer.write(&Frame::new(Bytes::from(data), Timestamp::ZERO)).unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut decoded = Vec::new(); while let Some(frame) = decoded_frame(&mut audio).await.unwrap() { @@ -1393,7 +1376,7 @@ mod tests { let at = (index * batch) as u64 * 1_000_000 / input_rate as u64; producer.write(&pcm_frame(chunk, 1_000_000 + at)).unwrap(); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut published = Vec::new(); while let Some(read) = reader.read().await.unwrap() { @@ -1468,7 +1451,7 @@ mod tests { producer .write(&Frame::new(data.into(), moq_net::Timestamp::ZERO)) .unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut packets = 0; while packet(&mut track).await.is_some() { @@ -1477,40 +1460,6 @@ mod tests { assert_eq!(packets, 11); } - /// A failed finish is final: writes are refused with its cause rather than reaching the codec - /// at the input rate, now that the resampler is gone, and a retry reports the cause rather than - /// publishing the terminal packets again. - #[tokio::test] - async fn a_failed_finish_refuses_later_calls() { - let input = Input { - format: Format::F32, - sample_rate: 44_100, - layout: Layout::Mono, - }; - let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); - let options = Options { - track: Some("audio".to_string()), - config: Config::new(48_000, Layout::Mono), - ..Options::default() - }; - let mut producer = Producer::new(&mut broadcast, catalog, input, &options).unwrap(); - let frame = |samples: usize| { - let data: Vec = vec![0.25f32; samples].iter().flat_map(|s| s.to_le_bytes()).collect(); - Frame::new(data.into(), moq_net::Timestamp::ZERO) - }; - producer.write(&frame(4_410)).unwrap(); - - // Another handle ends the track, so finishing cannot publish the tail. - producer.track.track().clone().abort(moq_net::Error::Cancel).unwrap(); - let cause = producer.finish().unwrap_err().to_string(); - - // Short of a codec frame, so nothing but the guard could refuse it. - let err = producer.write(&frame(100)).unwrap_err(); - assert_eq!(err.to_string(), cause); - assert_eq!(producer.finish().unwrap_err().to_string(), cause); - } - /// `reanchor` promises to drop buffered samples, and the resampler buffers /// samples of its own. Leaving those behind let `finish` flush pre-reset audio /// onto the track, stamped at an anchor that no longer exists. @@ -1548,7 +1497,7 @@ mod tests { .unwrap(); producer.reanchor(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); // The reset dropped everything, so the track ends without a packet. assert!(track.read().await.unwrap().is_none()); @@ -1586,7 +1535,7 @@ mod tests { producer.write(&full_frame(1_000_000)).unwrap(); producer.reanchor(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); assert!(packet(&mut track).await.is_some()); assert!(packet(&mut track).await.is_none()); @@ -1637,7 +1586,7 @@ mod tests { producer.reanchor(); producer.write(&full_frame(1_000_000)).unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut resumed_frames = 0; while let Some(frame) = decoded_frame(&mut audio).await.unwrap() { @@ -1856,7 +1805,7 @@ mod tests { } assert_eq!(count, packets, "{frame_us} us packets"); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); } } @@ -2003,7 +1952,7 @@ mod tests { // The stub holds the third frame, so only `finish` releases it. producer.write(&pcm_frame(&[0.1; 3 * 1024], 1_000_000)).unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut published = Vec::new(); while let Some(frame) = packet(&mut reader).await { @@ -2085,33 +2034,6 @@ mod tests { assert_eq!(pts, vec![0, 5_000_000]); } - /// A sub-frame write never reaches the closed track, so the producer must - /// refuse it itself rather than buffering samples that cannot be published. - #[tokio::test] - async fn write_after_finish_is_closed() { - let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); - let options = Options { - track: Some("audio".to_string()), - ..Options::default() - }; - let mut producer = Producer::new( - &mut broadcast, - catalog, - Input { - layout: Layout::Mono, - ..Input::default() - }, - &options, - ) - .unwrap(); - - producer.finish().unwrap(); - let err = producer.write(&pcm_frame(&[0.1; 100], 0)).unwrap_err(); - assert!(matches!(err, Error::Net(moq_net::Error::Closed))); - producer.abort(moq_net::Error::Cancel); - } - /// `Options::track = None` derives a codec-suffixed name rather than making /// the caller invent one, mirroring the video side. Pins the exact name the /// docs promise, and that a second producer doesn't collide with the first. diff --git a/rs/moq-audio/tests/integration/import.rs b/rs/moq-audio/tests/integration/import.rs index 413705a061c..36948a62109 100644 --- a/rs/moq-audio/tests/integration/import.rs +++ b/rs/moq-audio/tests/integration/import.rs @@ -17,7 +17,7 @@ async fn ts_surround_opus_decodes() { let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); let mut import = moq_mux::container::ts::Import::new(broadcast, catalog.reserve()); import.decode(&data[..]).unwrap(); - import.finish().unwrap(); + let _finished = import.finish().unwrap(); let snapshot = catalog.snapshot(); let (name, config) = snapshot.audio.renditions.iter().next().expect("an Opus track"); diff --git a/rs/moq-audio/tests/integration/roundtrip.rs b/rs/moq-audio/tests/integration/roundtrip.rs index 706f6d246ba..3baf277fd47 100644 --- a/rs/moq-audio/tests/integration/roundtrip.rs +++ b/rs/moq-audio/tests/integration/roundtrip.rs @@ -94,7 +94,7 @@ async fn opus_roundtrip(case: Case) { let mut consumer = decode::Consumer::new(&source, rendition, "audio", options) .await .unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); // Decoded from the track's first packet, so only the codec's pre-skip is dropped. let mut pcm = Vec::new(); while let Some(frame) = next_frame(&mut consumer).await.unwrap() { @@ -168,7 +168,7 @@ async fn pcm_round_trip_is_lossless() { let mut consumer = decode::Consumer::new(&broadcast_consumer, catalog, "pcm", decode::Options::default()) .await .unwrap(); - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let frame = next_frame(&mut consumer).await.unwrap().unwrap(); assert_eq!(frame.timestamp, Timestamp::from_micros(123_000).unwrap()); @@ -232,7 +232,7 @@ async fn pcm_five_one_reads_back_at_any_layout() { .unwrap(), ); } - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); for (mut consumer, (_, layout, frame)) in consumers.into_iter().zip(cases) { assert_eq!(consumer.layout(), layout); diff --git a/rs/moq-c/src/audio.rs b/rs/moq-c/src/audio.rs index 3fb66bbee92..30db9020c99 100644 --- a/rs/moq-c/src/audio.rs +++ b/rs/moq-c/src/audio.rs @@ -540,18 +540,20 @@ pub extern "C" fn moq_encode_audio_finish(producer: u32) -> i32 { let shared = State::lock().audio.producer(id)?; let finished: Result<(), Error> = (|| { let mut guard = shared.lock(); - let AudioEncoder { - mut producer, children, .. - } = guard.take().ok_or(Error::MediaNotFound)?; - if let Err(err) = producer.finish() { - // Readers see the failure rather than `Dropped`, as a failed video drain does. - children.forget(crate::publish::Child::Audio(id)); - producer.abort(moq_net::Error::Transport(err.to_string())); - return Err(err.into()); + let AudioEncoder { producer, children, .. } = guard.take().ok_or(Error::MediaNotFound)?; + match producer.finish() { + // The broadcast keeps the finished track, so its finish still holds it. + Ok(finished) => { + children.keep(crate::publish::Child::Audio(id), finished); + Ok(()) + } + // A failed finish already ended the track with an error, so readers see the + // failure rather than `Dropped`, as a failed video drain does. + Err(err) => { + children.forget(crate::publish::Child::Audio(id)); + Err(err.into()) + } } - // The broadcast keeps the finished track, so its finish still holds it. - children.keep(crate::publish::Child::Audio(id), producer); - Ok(()) })(); // A racing close may have released the id already. let _ = State::lock().audio.remove(id); diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index ce95185e4ba..1d74bba0feb 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -718,17 +718,24 @@ impl Publish { pub fn container_finish(&mut self, container: Id) -> Result<(), Error> { let id = container; - let mut container = self.containers.remove(id).ok_or(Error::MediaNotFound)?; - if let Err(err) = container.inner.finish() { - container.children.forget(Child::Container(id)); - let driver = container.driver.clone(); - container.abort(cause(&err)); - driver.check()?; - return Err(err.into()); + let Container { + inner, + driver, + children, + } = self.containers.remove(id).ok_or(Error::MediaNotFound)?; + match inner.finish() { + Ok(finished) => { + driver.check()?; + children.keep(Child::Container(id), finished); + Ok(()) + } + // A failed finish already ended every track, with an error where it could not finish. + Err(err) => { + children.forget(Child::Container(id)); + driver.check()?; + Err(err.into()) + } } - container.driver.check()?; - container.children.keep(Child::Container(id), container.inner); - Ok(()) } /// Insert or replace a caller-authored video rendition in the broadcast's catalog. @@ -1487,15 +1494,17 @@ impl Publish { pub fn json_window_finish(&mut self, window: Id) -> Result<(), Error> { let id = window; let window = self.json_window.remove(id).ok_or(Error::TrackNotFound)?; - let Data { - mut producer, children, .. - } = window.data; - if let Err(err) = producer.finish() { - children.forget(Child::JsonWindow(id)); - return Err(err.into()); + let Data { producer, children, .. } = window.data; + match producer.finish() { + Ok(finished) => { + children.keep(Child::JsonWindow(id), finished); + Ok(()) + } + Err(err) => { + children.forget(Child::JsonWindow(id)); + Err(err.into()) + } } - children.keep(Child::JsonWindow(id), producer); - Ok(()) } /// Write a frame into a raw group, `None` when untimed. diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs index 3ba6d3e9c23..2373e071f74 100644 --- a/rs/moq-cli/src/hls.rs +++ b/rs/moq-cli/src/hls.rs @@ -217,10 +217,10 @@ pub async fn import(target: ImportTarget, args: ImportArgs, instance: moq_net::I // The playlist ended: serve the finished broadcast for the linger, so a subscriber that // arrives after the end still gets the tail. An interrupt drops it at once. - importer.finish()?; + let finished = importer.finish()?; let linger = ingest.linger; tracing::info!(?linger, "playlist ended; serving the finished broadcast"); - moq_mux::Linger::new([route.broadcast().clone()], importer, linger) + moq_mux::Linger::new([route.broadcast().clone()], finished, linger) .wait() .await; Ok(()) diff --git a/rs/moq-cli/src/hls/record.rs b/rs/moq-cli/src/hls/record.rs index f091ede1519..f0b936ad49c 100644 --- a/rs/moq-cli/src/hls/record.rs +++ b/rs/moq-cli/src/hls/record.rs @@ -606,6 +606,7 @@ mod tests { broadcast: moq_net::origin::Route, catalog: moq_mux::catalog::Producer, video: Vec