feat(mux,audio,json)!: finishing a media producer returns a Finished handle

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;

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();
 	}
 });

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();
+});

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);
 	}
 }

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<T> {
 		constructor(config: Snapshot.Producer.Config<T>);
 		cut(): void;
-		finish(): void;
+		finish(): Snapshot.Finished;
 		mutate(fn: (value: T) => void, at?: Time.Timestamp): void;
 		update({ ... }: Timed<T>): void;
 	}
@@ -97,6 +98,7 @@ export namespace Stream {
 		constructor(config?: Stream.Config<T>);
 		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<T> {
 		constructor(config: Stream.Producer.Config<T>);
 		append({ ... }: Timed<T>): 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<T> {
+		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<T>;
 		get offset(): number;
 		pop(count: number, at?: Time.Timestamp): void;
 		push({ ... }: Timed<T>): void;

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";

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<T> {
 		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 {

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<Value>({ 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();
+});

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";

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<T> {
 		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;
 	}
 }
 

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<Rec>({ 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]);
+});