fix(loc): end each group with the duration marker

js/hang/src/container/consumer.test.ts

@@ -834,6 +834,36 @@ test("Consumer skips an empty LOC payload", async () => {
 	consumer.close();
 });
 
+test("Consumer skips the duration marker the LOC producer writes", async () => {
+	const track = new Track.Producer("test");
+	const producer = new LocProducer(track);
+	producer.encode(new Uint8Array([0xde, 0xad]), 0 as Time.Micro, true);
+	producer.encode(new Uint8Array([0xbe, 0xef]), 10_000 as Time.Micro, false);
+	producer.encode(new Uint8Array([0xca, 0xfe]), 20_000 as Time.Micro, true);
+	producer.close();
+
+	const consumer = new Consumer(replay(track), { format: new LocFormat("video"), maxDelay: 500 as Time.Milli });
+	await settle();
+	const first = await consumer.next();
+	expect(first?.frame?.payload).toEqual(new Uint8Array([0xde, 0xad]));
+	expect(first?.frame?.keyframe).toBe(true);
+	const second = await consumer.next();
+	expect(second?.frame?.payload).toEqual(new Uint8Array([0xbe, 0xef]));
+	const boundary = await consumer.next();
+	expect(boundary?.frame).toBeUndefined();
+	expect(boundary?.end).toBe(20_000 as Time.Micro);
+	const keyframe = await consumer.next();
+	expect(keyframe?.frame).toBeUndefined();
+	expect(keyframe?.end).toBeUndefined();
+	const next = await consumer.next();
+	expect(next?.frame?.payload).toEqual(new Uint8Array([0xca, 0xfe]));
+	expect(next?.frame?.keyframe).toBe(true);
+	const tail = await consumer.next();
+	expect(tail?.frame).toBeUndefined();
+	expect(tail?.end).toBe(30_000 as Time.Micro);
+	consumer.close();
+});
+
 test("Consumer preserves empty Legacy and LOC data frames", async () => {
 	for (const kind of ["legacy", "loc"]) {
 		const track = new Track.Producer("data");

js/loc/README.md

@@ -19,11 +19,15 @@ import * as Loc from "@moq/loc";
 const format = new Loc.Format();
 const frames = format.decode(rawFrameBytes);
 
-// Encode outgoing LOC frames
+// Encode outgoing LOC frames. A keyframe starts a group.
 const producer = new Loc.Producer(track);
 producer.encode(payload, timestampMicros, keyframe);
 ```
 
+The producer is for video tracks.
+An ordered group ends with an empty duration frame, at the next keyframe's timestamp or one interval after the last sample when the track closes.
+`Format.end()` identifies that frame for audio and video, and the hang consumer skips it.
+
 ## License
 
 MIT OR Apache-2.0

js/loc/src/index.test.ts

@@ -1,6 +1,6 @@
 import { expect, test } from "bun:test";
-import { type Time, Varint } from "@moq/net";
-import { Format } from "./index.ts";
+import { Time, Track, Varint } from "@moq/net";
+import { Format, Producer } from "./index.ts";
 
 const PROP_TIMESCALE = 0x08;
 const PROP_TIMESTAMP = 0x10;
@@ -149,3 +149,106 @@ test("Format rejects a timestamp past 2^53 - 1 instead of rounding", () => {
 	const props = concat(Varint.encode(PROP_TIMESTAMP), Varint.encode(2n ** 53n));
 	expect(() => new Format().decode(buildFrame(props, new Uint8Array()))).toThrow(/larger than 53-bits/);
 });
+
+/** Each group's (LOC timestamp, payload size, net timestamp) after the producer has closed the track. */
+async function readGroups(track: Track.Producer) {
+	const subscriber = track.subscribe({ maxDelay: Time.Milli(30_000) });
+	const format = new Format("video");
+	const groups: { timestamp: number; size: number; net: number | undefined }[][] = [];
+	for (;;) {
+		const group = await subscriber.recvGroup();
+		if (!group) break;
+		const frames: { timestamp: number; size: number; net: number | undefined }[] = [];
+		for (;;) {
+			const frame = await group.readFrame();
+			if (!frame) break;
+			const [decoded] = format.decode(frame.payload);
+			frames.push({
+				timestamp: decoded.timestamp,
+				size: decoded.payload.byteLength,
+				net: frame.timestamp?.asMicros(),
+			});
+		}
+		groups.push(frames);
+	}
+	return groups;
+}
+
+test("Producer ends a group with an empty frame at the next keyframe", async () => {
+	const track = new Track.Producer("video");
+	const producer = new Producer(track);
+	producer.encode(new Uint8Array([0xde, 0xad]), 0 as Time.Micro, true);
+	producer.encode(new Uint8Array([0xbe, 0xef]), 10_000 as Time.Micro, false);
+	producer.encode(new Uint8Array([0xca, 0xfe]), 20_000 as Time.Micro, true);
+	producer.close();
+
+	const groups = await readGroups(track);
+	expect(groups[0]).toEqual([
+		{ timestamp: 0, size: 2, net: 0 },
+		{ timestamp: 10_000, size: 2, net: 10_000 },
+		{ timestamp: 20_000, size: 0, net: 20_000 },
+	]);
+	// The tail has no successor, so the marker is one interval after the last sample.
+	expect(groups[1]).toEqual([
+		{ timestamp: 20_000, size: 2, net: 20_000 },
+		{ timestamp: 30_000, size: 0, net: 30_000 },
+	]);
+	expect(groups).toHaveLength(2);
+
+	const marker = groups[0][2];
+	expect(
+		new Format("video").end({
+			payload: new Uint8Array(marker.size),
+			timestamp: marker.timestamp as Time.Micro,
+			keyframe: false,
+		}),
+	).toBe(20_000 as Time.Micro);
+});
+
+test("Producer omits the marker when a group has no interval to close", async () => {
+	const track = new Track.Producer("video");
+	const producer = new Producer(track);
+	producer.encode(new Uint8Array([1]), 0 as Time.Micro, true);
+	producer.close();
+
+	expect(await readGroups(track)).toEqual([[{ timestamp: 0, size: 1, net: 0 }]]);
+});
+
+test("Producer estimates the marker when the next keyframe overlaps an ordered tail", async () => {
+	const track = new Track.Producer("video");
+	const producer = new Producer(track);
+	producer.encode(new Uint8Array([1]), 0 as Time.Micro, true);
+	producer.encode(new Uint8Array([1]), 10_000 as Time.Micro, false);
+	producer.encode(new Uint8Array([1]), 5_000 as Time.Micro, true);
+	producer.close();
+
+	expect(await readGroups(track)).toEqual([
+		[
+			{ timestamp: 0, size: 1, net: 0 },
+			{ timestamp: 10_000, size: 1, net: 10_000 },
+			{ timestamp: 20_000, size: 0, net: 20_000 },
+		],
+		[{ timestamp: 5_000, size: 1, net: 5_000 }],
+	]);
+});
+
+test("Producer omits a reordered group's marker and a keyframe that overlaps the tail", async () => {
+	const track = new Track.Producer("video");
+	const producer = new Producer(track);
+	for (const [index, timestamp] of [0, 120_000, 40_000, 80_000].entries()) {
+		producer.encode(new Uint8Array([1]), timestamp as Time.Micro, index === 0);
+	}
+	// Overlaps the previous frame, so it is not that group's end, and the cadence does not carry over.
+	producer.encode(new Uint8Array([1]), 50_000 as Time.Micro, true);
+	producer.close();
+
+	expect(await readGroups(track)).toEqual([
+		[
+			{ timestamp: 0, size: 1, net: 0 },
+			{ timestamp: 120_000, size: 1, net: 120_000 },
+			{ timestamp: 40_000, size: 1, net: 40_000 },
+			{ timestamp: 80_000, size: 1, net: 80_000 },
+		],
+		[{ timestamp: 50_000, size: 1, net: 50_000 }],
+	]);
+});

js/loc/src/index.ts

@@ -118,20 +118,37 @@ export interface Source {
  *
  * Each call to {@link encode} produces one moq-net frame containing a
  * property block with the 0x10 timestamp (in microseconds) and the codec
- * bitstream payload.
+ * bitstream payload. A keyframe starts a group. The group it closes, and the
+ * group still open at {@link close}, ends with an empty frame at the exclusive
+ * end of the last sample: the same duration marker `moq-mux` writes.
+ * Use it for video only: a data reader would see the marker as an empty object.
  */
 export class Producer {
 	#track: Moq.Track.Producer;
 	#group?: Moq.Group.Producer;
+	// Last timestamp in the open group. The next keyframe is this group's end unless it lands earlier.
+	#previous?: Time.Micro;
+	// Furthest timestamp in the open group. The tail marker estimates from this, not decode order.
+	#end?: Time.Micro;
+	// Gap between the last two increasing timestamps. Kept across groups so a clean close can bound the tail.
+	#interval?: Time.Micro;
+	// A frame's timestamp preceded the one before it. That group's presentation end does not bound its tail.
+	#reordered = false;
 
 	constructor(track: Moq.Track.Producer) {
 		this.#track = track;
 	}
 
-	/** Encode one frame and write it to the track. Keyframes start a new group. */
+	/**
+	 * Encode one frame and write it to the track.
+	 * A keyframe ends the open group with a duration marker, then starts the next.
+	 */
 	encode(data: Uint8Array | Source, timestamp: Time.Micro, keyframe: boolean) {
 		if (keyframe) {
-			this.#group?.close();
+			// A keyframe before the previous frame overlaps that frame, so it is not the group's end.
+			const overlaps = this.#previous !== undefined && timestamp < this.#previous;
+			this.#closeGroup(overlaps ? undefined : timestamp);
+			if (overlaps) this.#interval = undefined;
 			this.#group = this.#track.appendGroup();
 		} else if (!this.#group) {
 			throw new Error("must start with a keyframe");
@@ -141,6 +158,13 @@ export class Producer {
 			payload: this.#encode(data, timestamp),
 			timestamp: Time.Timestamp.fromMicros(timestamp),
 		});
+
+		this.#reordered ||= this.#previous !== undefined && timestamp < this.#previous;
+		if (this.#previous !== undefined && timestamp > this.#previous) {
+			this.#interval = (timestamp - this.#previous) as Time.Micro;
+		}
+		this.#previous = timestamp;
+		if (this.#end === undefined || timestamp > this.#end) this.#end = timestamp;
 	}
 
 	#encode(source: Uint8Array | Source, timestamp: Time.Micro): Uint8Array {
@@ -172,8 +196,36 @@ export class Producer {
 		return out;
 	}
 
-	/** Close the current group and the underlying track, optionally with an error. */
+	// End the open group. `end` is the next keyframe when one is already in hand; otherwise the
+	// last gap estimates where the tail stops. A reordered group omits the marker: its presentation
+	// end does not bound the last frame in decode order.
+	#closeGroup(end?: Time.Micro) {
+		const group = this.#group;
+		if (!group) return;
+
+		let marker = this.#reordered ? undefined : end;
+		if (marker === undefined && !this.#reordered && this.#end !== undefined && this.#interval !== undefined) {
+			marker = (this.#end + this.#interval) as Time.Micro;
+		}
+		if (marker !== undefined) {
+			group.writeFrame({
+				payload: this.#encode(new Uint8Array(), marker),
+				timestamp: Time.Timestamp.fromMicros(marker),
+			});
+		}
+		group.close();
+		this.#group = undefined;
+		this.#previous = undefined;
+		this.#end = undefined;
+		this.#reordered = false;
+	}
+
+	/**
+	 * Close the current group and the underlying track.
+	 * A clean close writes the duration marker; an error does not.
+	 */
 	close(err?: Error) {
+		if (!err) this.#closeGroup();
 		this.#group?.close();
 		this.#track.close(err);
 	}

quest/m2/README.md

@@ -71,7 +71,6 @@ consumer yet, waits in [m3](/quest/m3/README.md).
 - [Mobile completion](/quest/m2/mobile-completion.md) - verify the selected native/mobile path before closing #700
 - [Opus implementation](/quest/m2/audio-opus-backend.md) - compare Opus codec quality, CPU, build cost, and the loss recovery each backend offers
 - [Latency ledger](/quest/m2/latency-ledger.md) - a viewer reports its share of playback delay stage by stage: jitter buffer, decode, render, and device
-- [JS LOC duration marker](/quest/m2/js-loc-duration-marker.md) - `@moq/loc`'s producer ends each video group with the empty duration frame, as moq-mux does
 - [Media Foundation decode](/quest/m2/audio-decode-mediafoundation.md) - Windows decodes HE-AAC, multichannel AAC, and what else the MFTs offer
 - [Media Foundation encode](/quest/m2/audio-encode-mediafoundation.md) - Windows encodes AAC-LC
 - [MediaCodec decode](/quest/m2/audio-decode-mediacodec.md) - Android decodes HE-AAC, multichannel AAC, and what else the device offers