moq-lite-03: Support dynamic latency

js/clock/src/main.ts

@@ -80,13 +80,13 @@ async function publish(config: Config) {
 
 	// Wait until we get a subscription for the track
 	for (;;) {
-		const request = await broadcast.requested();
-		if (!request) break;
+		const track = await broadcast.requested();
+		if (!track) break;
 
-		if (request.track.name === config.track) {
-			publishTrack(request.track);
+		if (track.name === config.track) {
+			publishTrack(track);
 		} else {
-			request.track.close(new Error("not found"));
+			track.close(new Error("not found"));
 		}
 	}
 }
@@ -103,7 +103,8 @@ async function publishTrack(track: Moq.Track) {
 
 		// Send the base timestamp (everything but seconds) - matching Rust format
 		const base = `${now.toISOString().slice(0, 16).replace("T", " ")}:`;
-		group.writeString(base);
+		const frame = Moq.Frame.fromString(base);
+		group.writeFrame(frame);
 
 		// Send individual seconds for this minute
 		const currentMinute = now.getMinutes();
@@ -112,7 +113,8 @@ async function publishTrack(track: Moq.Track) {
 			const secondsNow = new Date();
 			const seconds = secondsNow.getSeconds().toString().padStart(2, "0");
 
-			group.writeString(seconds);
+			const frame = Moq.Frame.fromString(seconds);
+			group.writeFrame(frame);
 
 			// Wait until next second
 			const nextSecond = new Date(secondsNow);
@@ -138,7 +140,7 @@ async function subscribe(config: Config) {
 	console.log("✅ Connected to relay:", config.url);
 
 	const broadcast = connection.consume(Moq.Path.from(config.broadcast));
-	const track = broadcast.subscribe(config.track, 0);
+	const track = broadcast.subscribe({ name: config.track });
 
 	console.log("✅ Subscribed to track:", config.track);
 
@@ -157,16 +159,16 @@ async function subscribe(config: Config) {
 			continue;
 		}
 
-		const base = new TextDecoder().decode(baseFrame);
+		const base = baseFrame.toString();
 
 		// Read individual second frames
 		for (;;) {
-			const frame = await group.readString();
+			const frame = await group.readFrame();
 			if (!frame) {
 				break; // End of group
 			}
 
-			const seconds = parseInt(frame, 10);
+			const seconds = parseInt(frame.toString(), 10);
 
 			// Clock emoji positions
 			const clockEmojis = ["🕛", "🕐", "🕑", "🕒", "🕓", "🕔", "🕕", "🕖", "🕗", "🕘", "🕙", "🕚"];

js/hang/src/catalog/audio.ts

@@ -1,12 +1,7 @@
 import { z } from "zod";
 import { ContainerSchema, DEFAULT_CONTAINER } from "./container";
 import { u53Schema } from "./integers";
-
-// Backwards compatibility: old track schema
-const TrackSchema = z.object({
-	name: z.string(),
-	priority: z.number().int().min(0).max(255),
-});
+import { TrackSchema } from "./track";
 
 // Mirrors AudioDecoderConfig
 // https://w3c.github.io/webcodecs/#audio-decoder-config
@@ -40,8 +35,9 @@ export const AudioSchema = z
 		// This is not an array so it will work with JSON Merge Patch.
 		renditions: z.record(z.string(), AudioConfigSchema),
 
-		// The priority of the audio track, relative to other tracks in the broadcast.
-		priority: z.number().int().min(0).max(255),
+		// DEPRECATED: The priority of the audio track, relative to other tracks in the broadcast.
+		// The subscriber is expected to choose its own priority, instead of being told.
+		priority: z.number().int().min(0).max(255).default(0),
 	})
 	.or(
 		// Backwards compatibility: transform old {track, config} format to new object format

js/hang/src/catalog/root.ts

@@ -41,5 +41,5 @@ export function decode(raw: Uint8Array): Root {
 export async function fetch(track: Moq.Track): Promise<Root | undefined> {
 	const frame = await track.readFrame();
 	if (!frame) return undefined;
-	return decode(frame);
+	return decode(frame.payload);
 }

js/hang/src/catalog/track.ts

@@ -2,6 +2,8 @@ import { z } from "zod";
 
 export const TrackSchema = z.object({
 	name: z.string(),
-	priority: z.number().int().min(0).max(255),
+	// DEPRECATED: The priority of the track, relative to other tracks in the broadcast.
+	// The subscriber is supposed to choose its own priority, instead of being told.
+	priority: z.number().int().min(0).max(255).default(0),
 });
 export type Track = z.infer<typeof TrackSchema>;

js/hang/src/catalog/video.ts

@@ -1,12 +1,7 @@
 import { z } from "zod";
 import { ContainerSchema, DEFAULT_CONTAINER } from "./container";
 import { u53Schema } from "./integers";
-
-// Backwards compatibility: old track schema
-const TrackSchema = z.object({
-	name: z.string(),
-	priority: z.number().int().min(0).max(255),
-});
+import { TrackSchema } from "./track";
 
 // Based on VideoDecoderConfig
 export const VideoConfigSchema = z.object({
@@ -54,7 +49,8 @@ export const VideoSchema = z
 		renditions: z.record(z.string(), VideoConfigSchema),
 
 		// The priority of the video track, relative to other tracks in the broadcast.
-		priority: z.number().int().min(0).max(255),
+		// TODO: Remove this; it's for backwards compatibility only
+		priority: z.number().int().min(0).max(255).default(0),
 
 		// Render the video at this size in pixels.
 		// This is separate from the display aspect ratio because it does not require reinitialization.

js/hang/src/frame.ts

@@ -1,4 +1,4 @@
-import type * as Moq from "@moq/lite";
+import * as Moq from "@moq/lite";
 import { Time } from "@moq/lite";
 import { Effect, Signal } from "@moq/signals";
 import type * as Catalog from "./catalog";
@@ -16,8 +16,9 @@ export interface Frame {
 	group: number;
 }
 
-export function encode(source: Uint8Array | Source, timestamp: Time.Micro, container?: Catalog.Container): Uint8Array {
+export function encode(source: Uint8Array | Source, timestamp: Time.Micro, container?: Catalog.Container): Moq.Frame {
 	// Encode timestamp using the specified container format
+	// TODO This should be delta encoded, not the full timestamp.
 	const timestampBytes = Container.encodeTimestamp(timestamp, container);
 
 	// Allocate buffer for timestamp + payload
@@ -34,14 +35,18 @@ export function encode(source: Uint8Array | Source, timestamp: Time.Micro, conta
 		source.copyTo(data.subarray(timestampBytes.byteLength));
 	}
 
-	return data;
+	// NOTE: We encode the timestamp into the MoQ layer as well, but in milliseconds.
+	// TODO: Once this is widespread enough, we should use it at least as the base.
+	return new Moq.Frame({ payload: data, instant: Time.Milli.fromMicro(timestamp) });
 }
 
 // NOTE: A keyframe is always the first frame in a group, so it's not encoded on the wire.
-export function decode(buffer: Uint8Array, container?: Catalog.Container): { data: Uint8Array; timestamp: Time.Micro } {
+export function decode(frame: Moq.Frame, container?: Catalog.Container): { data: Uint8Array; timestamp: Time.Micro } {
 	// Decode timestamp using the specified container format
-	const [timestamp, data] = Container.decodeTimestamp(buffer, container);
-	return { timestamp: timestamp as Time.Micro, data };
+	// TODO This should be delta encoded, not the full timestamp.
+	// TODO: Use frame.instant to avoid double encoding the timestamp.
+	const [timestamp, data] = Container.decodeTimestamp(frame.payload, container);
+	return { timestamp, data };
 }
 
 export class Producer {

js/hang/src/publish/audio/encoder.ts

@@ -5,8 +5,8 @@ import type * as Catalog from "../../catalog";
 import { DEFAULT_CONTAINER } from "../../catalog";
 import { u53 } from "../../catalog/integers";
 import * as Frame from "../../frame";
+import { PRIORITY } from "../../priority";
 import * as libav from "../../util/libav";
-import { PRIORITY } from "../priority";
 import type * as Capture from "./capture";
 import type { Source } from "./types";
 

js/hang/src/publish/broadcast.ts

@@ -72,45 +72,43 @@ export class Broadcast {
 
 	async #runBroadcast(broadcast: Moq.Broadcast, effect: Effect) {
 		for (;;) {
-			const request = await broadcast.requested();
-			if (!request) break;
+			const track = await broadcast.requested();
+			if (!track) break;
 
-			effect.cleanup(() => request.track.close());
+			effect.cleanup(() => track.close());
 
 			effect.effect((effect) => {
-				if (effect.get(request.track.state.closed)) return;
-
-				switch (request.track.name) {
+				switch (track.name) {
 					case Broadcast.CATALOG_TRACK:
-						this.#serveCatalog(request.track, effect);
+						this.#serveCatalog(track, effect);
 						break;
 					case Location.Window.TRACK:
-						this.location.window.serve(request.track, effect);
+						this.location.window.serve(track, effect);
 						break;
 					case Location.Peers.TRACK:
-						this.location.peers.serve(request.track, effect);
+						this.location.peers.serve(track, effect);
 						break;
 					case Preview.TRACK:
-						this.preview.serve(request.track, effect);
+						this.preview.serve(track, effect);
 						break;
 					case Chat.Typing.TRACK:
-						this.chat.typing.serve(request.track, effect);
+						this.chat.typing.serve(track, effect);
 						break;
 					case Chat.Message.TRACK:
-						this.chat.message.serve(request.track, effect);
+						this.chat.message.serve(track, effect);
 						break;
 					case Audio.Encoder.TRACK:
-						this.audio.serve(request.track, effect);
+						this.audio.serve(track, effect);
 						break;
 					case Video.Root.TRACK_HD:
-						this.video.hd.serve(request.track, effect);
+						this.video.hd.serve(track, effect);
 						break;
 					case Video.Root.TRACK_SD:
-						this.video.sd.serve(request.track, effect);
+						this.video.sd.serve(track, effect);
 						break;
 					default:
-						console.error("received subscription for unknown track", request.track.name);
-						request.track.close(new Error(`Unknown track: ${request.track.name}`));
+						console.error("received subscription for unknown track", track.name);
+						track.close(new Error(`Unknown track: ${track.name}`));
 						break;
 				}
 			});
@@ -120,7 +118,8 @@ export class Broadcast {
 	#serveCatalog(track: Moq.Track, effect: Effect): void {
 		if (!effect.get(this.enabled)) {
 			// Clear the catalog.
-			track.writeFrame(Catalog.encode({}));
+			const frame = new Moq.Frame({ payload: Catalog.encode({}) });
+			track.writeFrame(frame);
 			return;
 		}
 
@@ -134,8 +133,8 @@ export class Broadcast {
 			preview: effect.get(this.preview.catalog),
 		};
 
-		const encoded = Catalog.encode(catalog);
-		track.writeFrame(encoded);
+		const frame = new Moq.Frame({ payload: Catalog.encode(catalog) });
+		track.writeFrame(frame);
 	}
 
 	close() {

js/hang/src/publish/chat/message.ts

@@ -1,7 +1,7 @@
-import type * as Moq from "@moq/lite";
+import * as Moq from "@moq/lite";
 import { Effect, Signal } from "@moq/signals";
 import type * as Catalog from "../../catalog";
-import { PRIORITY } from "../priority";
+import { PRIORITY } from "../../priority";
 
 export type MessageProps = {
 	enabled?: boolean | Signal<boolean>;
@@ -37,7 +37,8 @@ export class Message {
 		if (!enabled) return;
 
 		const latest = effect.get(this.latest);
-		track.writeString(latest ?? "");
+		const frame = Moq.Frame.fromString(latest ?? "");
+		track.writeFrame(frame);
 	}
 
 	close() {

js/hang/src/publish/chat/typing.ts

@@ -1,7 +1,7 @@
-import type * as Moq from "@moq/lite";
+import * as Moq from "@moq/lite";
 import { Effect, Signal } from "@moq/signals";
 import type * as Catalog from "../../catalog";
-import { PRIORITY } from "../priority";
+import { PRIORITY } from "../../priority";
 
 export type TypingProps = {
 	enabled?: boolean | Signal<boolean>;
@@ -37,7 +37,8 @@ export class Typing {
 		if (!enabled) return;
 
 		const active = effect.get(this.active);
-		track.writeBool(active);
+		const frame = Moq.Frame.fromBool(active);
+		track.writeFrame(frame);
 	}
 
 	close() {

js/hang/src/publish/location/peers.ts

@@ -2,7 +2,7 @@ import type * as Moq from "@moq/lite";
 import * as Zod from "@moq/lite/zod";
 import { Effect, Signal } from "@moq/signals";
 import * as Catalog from "../../catalog";
-import { PRIORITY } from "../priority";
+import { PRIORITY } from "../../priority";
 
 export interface PeersProps {
 	enabled?: boolean | Signal<boolean>;

js/hang/src/publish/location/window.ts

@@ -2,7 +2,7 @@ import type * as Moq from "@moq/lite";
 import * as Zod from "@moq/lite/zod";
 import { Effect, Signal } from "@moq/signals";
 import * as Catalog from "../../catalog";
-import { PRIORITY } from "../priority";
+import { PRIORITY } from "../../priority";
 
 export type WindowProps = {
 	// If true, then we'll publish our position to the broadcast.

js/hang/src/publish/preview.ts

@@ -1,7 +1,7 @@
-import type * as Moq from "@moq/lite";
+import * as Moq from "@moq/lite";
 import { Effect, Signal } from "@moq/signals";
 import type * as Catalog from "../catalog";
-import { PRIORITY } from "./priority";
+import { PRIORITY } from "../priority";
 
 export type PreviewProps = {
 	enabled?: boolean | Signal<boolean>;
@@ -35,7 +35,8 @@ export class Preview {
 		const info = effect.get(this.info);
 		if (!info) return;
 
-		track.writeJson(info);
+		const frame = Moq.Frame.fromJson(info);
+		track.writeFrame(frame);
 	}
 
 	close() {

js/hang/src/publish/video/index.ts

@@ -1,6 +1,6 @@
 import { Effect, Signal } from "@moq/signals";
 import * as Catalog from "../../catalog";
-import { PRIORITY } from "../priority";
+import { PRIORITY } from "../../priority";
 import { Encoder, type EncoderProps } from "./encoder";
 import { TrackProcessor } from "./polyfill";
 import type { Source } from "./types";

js/hang/src/watch/audio/source.ts

@@ -3,6 +3,7 @@ import type { Time } from "@moq/lite";
 import { Effect, type Getter, Signal } from "@moq/signals";
 import type * as Catalog from "../../catalog";
 import * as Frame from "../../frame";
+import { PRIORITY } from "../../priority";
 import * as Hex from "../../util/hex";
 import * as libav from "../../util/libav";
 import type * as Render from "./render";
@@ -163,7 +164,7 @@ export class Source {
 		const active = effect.get(this.active);
 		if (!active) return;
 
-		const sub = broadcast.subscribe(active, catalog.priority);
+		const sub = broadcast.subscribe({ name: active, priority: PRIORITY.audio, maxLatency: this.latency });
 		effect.cleanup(() => sub.close());
 
 		// Create consumer with slightly less latency than the render worklet to avoid underflowing.

js/hang/src/watch/broadcast.ts

@@ -1,7 +1,7 @@
 import type * as Moq from "@moq/lite";
 import { Effect, type Getter, Signal } from "@moq/signals";
 import * as Catalog from "../catalog";
-import { PRIORITY } from "../publish/priority";
+import { PRIORITY } from "../priority";
 import * as Audio from "./audio";
 import { Chat, type ChatProps } from "./chat";
 import * as Location from "./location";
@@ -101,7 +101,6 @@ export class Broadcast {
 
 				// Require full equality
 				if (update.path !== path) {
-					console.warn("ignoring announce", update.path);
 					continue;
 				}
 
@@ -131,7 +130,7 @@ export class Broadcast {
 
 		this.status.set("loading");
 
-		const catalog = broadcast.subscribe("catalog.json", PRIORITY.catalog);
+		const catalog = broadcast.subscribe({ name: "catalog.json", priority: PRIORITY.catalog });
 		effect.cleanup(() => catalog.close());
 
 		effect.spawn(this.#fetchCatalog.bind(this, catalog));