Add bandwidth estimation for adaptive bitrate control (#1208)

Co-authored-by: Claude <noreply@anthropic.com>

Author: kixelated

CLAUDE.md

@@ -111,6 +111,7 @@ match version {
 - **Formatting/Linting**: Biome for JS/TS formatting and linting
 - **UI**: Solid.js for Web Components in `@moq/watch/ui` and `@moq/publish/ui`
 - **Builds**: Nix flake for reproducible builds (optional)
+- **JS async patterns**: Use `Effect.interval()`, `Effect.timer()`, and `Effect.event()` helpers from `@moq/signals` instead of raw `setInterval`, `setTimeout`, `addEventListener`. These handle cleanup automatically when the Effect is closed.
 
 ## Testing Approach
 

js/lite/src/bandwidth.ts

@@ -0,0 +1,14 @@
+import { Signal } from "@moq/signals";
+
+/**
+ * A bandwidth estimate in bits per second, or undefined if unknown.
+ *
+ * This is a Signal that can be read synchronously via `peek()`,
+ * observed reactively via `effect.get()`, or updated via `set()`.
+ */
+export type Bandwidth = Signal<number | undefined>;
+
+/** Create a new bandwidth signal. */
+export function createBandwidth(): Bandwidth {
+	return new Signal<number | undefined>(undefined);
+}

js/lite/src/connection/established.ts

@@ -1,4 +1,5 @@
 import type { Announced } from "../announced.ts";
+import type { Bandwidth } from "../bandwidth.ts";
 import type { Broadcast } from "../broadcast.ts";
 import type * as Path from "../path.ts";
 
@@ -7,6 +8,12 @@ export interface Established {
 	readonly url: URL;
 	readonly version: string;
 
+	/** Estimated send bitrate from the congestion controller (if supported). */
+	readonly sendBandwidth?: Bandwidth;
+
+	/** Estimated receive bitrate from PROBE (moq-lite-03+ only). */
+	readonly recvBandwidth?: Bandwidth;
+
 	announced(prefix?: Path.Valid): Announced;
 	publish(path: Path.Valid, broadcast: Broadcast): void;
 	consume(broadcast: Path.Valid): Broadcast;

js/lite/src/index.ts

@@ -1,5 +1,6 @@
 export * as Signals from "@moq/signals";
 export * from "./announced.ts";
+export * from "./bandwidth.ts";
 export * from "./broadcast.ts";
 export * as Connection from "./connection/index.ts";
 export * from "./group.ts";

js/lite/src/lite/connection.ts

@@ -1,4 +1,5 @@
 import type { Announced } from "../announced.ts";
+import { type Bandwidth, createBandwidth } from "../bandwidth.ts";
 import type { Broadcast } from "../broadcast.ts";
 import type { Established } from "../connection/established.ts";
 import * as Path from "../path.ts";
@@ -10,7 +11,9 @@ import { SessionInfo } from "./session.ts";
 import { StreamId } from "./stream.ts";
 import { Subscribe } from "./subscribe.ts";
 import { Subscriber } from "./subscriber.ts";
-import { type Version, versionName } from "./version.ts";
+import { Version, versionName } from "./version.ts";
+
+const SEND_BW_POLL_INTERVAL = 100; // ms
 
 /**
  * Represents a connection to a MoQ server.
@@ -39,8 +42,11 @@ export class Connection implements Established {
 	// Module for distributing tracks.
 	#subscriber: Subscriber;
 
-	// Just to avoid logging when `close()` is called.
-	#closed = false;
+	/** Estimated send bitrate from the congestion controller. */
+	readonly sendBandwidth?: Bandwidth;
+
+	/** Estimated receive bitrate from PROBE (moq-lite-03+ only). */
+	readonly recvBandwidth?: Bandwidth;
 
 	/**
 	 * Creates a new Connection instance.
@@ -57,8 +63,19 @@ export class Connection implements Established {
 		this.version = versionName(version);
 		this.#version = version;
 
+		// Send bandwidth is version-agnostic: depends on browser/QUIC support.
+		const hasGetStats = typeof (quic as unknown as { getStats?: unknown }).getStats === "function";
+		if (hasGetStats) {
+			this.sendBandwidth = createBandwidth();
+		}
+
+		// Recv bandwidth requires PROBE support (not available in older drafts).
+		if (version !== Version.DRAFT_01 && version !== Version.DRAFT_02) {
+			this.recvBandwidth = createBandwidth();
+		}
+
 		this.#publisher = new Publisher(this.#quic, this.#version);
-		this.#subscriber = new Subscriber(this.#quic, this.#version);
+		this.#subscriber = new Subscriber(this.#quic, this.#version, this.recvBandwidth);
 
 		this.#run();
 	}
@@ -67,9 +84,6 @@ export class Connection implements Established {
 	 * Closes the connection.
 	 */
 	close() {
-		if (this.#closed) return;
-
-		this.#closed = true;
 		this.#publisher.close();
 		this.#subscriber.close();
 
@@ -82,62 +96,46 @@ export class Connection implements Established {
 	}
 
 	async #run(): Promise<void> {
-		const session = this.#runSession();
-		const bidis = this.#runBidis();
-		const unis = this.#runUnis();
+		const tasks: Promise<void>[] = [this.#runSession(), this.#runBidis(), this.#runUnis()];
+
+		if (this.sendBandwidth) {
+			tasks.push(this.#runSendBandwidth(this.sendBandwidth));
+		}
+
+		if (this.recvBandwidth) {
+			tasks.push(this.#subscriber.runProbe());
+		}
 
 		try {
-			await Promise.all([session, bidis, unis]);
+			await Promise.all(tasks);
 		} catch (err) {
-			if (!this.#closed) {
-				console.error("fatal error running connection", err);
-			}
+			console.error("fatal error running connection", err);
 		} finally {
 			this.close();
 		}
 	}
 
-	/**
-	 * Publishes a broadcast to the connection.
-	 * @param name - The broadcast path to publish
-	 * @param broadcast - The broadcast to publish
-	 */
 	publish(path: Path.Valid, broadcast: Broadcast) {
 		this.#publisher.publish(path, broadcast);
 	}
 
-	/**
-	 * Gets the next announced broadcast.
-	 */
 	announced(prefix = Path.empty()): Announced {
 		return this.#subscriber.announced(prefix);
 	}
 
-	/**
-	 * Consumes a broadcast from the connection.
-	 *
-	 * @remarks
-	 * If the broadcast is not found, a "not found" error will be thrown when requesting any tracks.
-	 *
-	 * @param broadcast - The path of the broadcast to consume
-	 * @returns A Broadcast instance
-	 */
 	consume(broadcast: Path.Valid): Broadcast {
 		return this.#subscriber.consume(broadcast);
 	}
 
 	async #runSession() {
 		if (!this.#session) {
-			// moq-lite draft-03 doesn't use a session stream.
 			return;
 		}
 
 		try {
-			// Receive messages until the connection is closed.
 			for (;;) {
 				const msg = await SessionInfo.decodeMaybe(this.#session.reader, this.#version);
 				if (!msg) break;
-				// TODO use the session info
 			}
 		} finally {
 			console.debug("session stream closed");
@@ -147,9 +145,7 @@ export class Connection implements Established {
 	async #runBidis() {
 		for (;;) {
 			const stream = await Stream.accept(this.#quic);
-			if (!stream) {
-				break;
-			}
+			if (!stream) break;
 
 			this.#runBidi(stream)
 				.catch((err: unknown) => {
@@ -169,14 +165,11 @@ export class Connection implements Established {
 		} else if (typ === StreamId.Announce) {
 			const msg = await AnnounceInterest.decode(stream.reader);
 			await this.#publisher.runAnnounce(msg, stream);
-			return;
 		} else if (typ === StreamId.Subscribe) {
 			const msg = await Subscribe.decode(stream.reader, this.#version);
 			await this.#publisher.runSubscribe(msg, stream);
-			return;
 		} else if (typ === StreamId.Probe) {
 			await this.#publisher.runProbe(stream);
-			return;
 		} else {
 			throw new Error(`unknown stream type: ${typ.toString()}`);
 		}
@@ -187,9 +180,7 @@ export class Connection implements Established {
 
 		for (;;) {
 			const stream = await readers.next();
-			if (!stream) {
-				break;
-			}
+			if (!stream) break;
 
 			this.#runUni(stream)
 				.then(() => {
@@ -211,10 +202,29 @@ export class Connection implements Established {
 		}
 	}
 
-	/**
-	 * Returns a promise that resolves when the connection is closed.
-	 * @returns A promise that resolves when closed
-	 */
+	async #runSendBandwidth(bandwidth: Bandwidth): Promise<void> {
+		const quic = this.#quic as unknown as {
+			getStats: () => Promise<{ estimatedSendRate: number | null }>;
+		};
+
+		return new Promise<void>((resolve) => {
+			const id = setInterval(async () => {
+				try {
+					const stats = await quic.getStats();
+					bandwidth.set(stats.estimatedSendRate ?? undefined);
+				} catch {
+					clearInterval(id);
+					resolve();
+				}
+			}, SEND_BW_POLL_INTERVAL);
+
+			void this.closed.then(() => {
+				clearInterval(id);
+				resolve();
+			});
+		});
+	}
+
 	get closed(): Promise<void> {
 		return this.#quic.closed.then(() => undefined);
 	}

js/lite/src/lite/publisher.ts

@@ -270,7 +270,8 @@ export class Publisher {
 			getStats?: () => Promise<{ estimatedSendRate: number | null }>;
 		};
 		if (!quic.getStats) {
-			stream.abort(new Error("stats not supported"));
+			// Close gracefully instead of aborting to avoid killing the session.
+			stream.close();
 			return;
 		}
 

js/lite/src/lite/subscriber.ts

@@ -1,4 +1,5 @@
 import { Announced } from "../announced.ts";
+import type { Bandwidth } from "../bandwidth.ts";
 import { Broadcast, type TrackRequest } from "../broadcast.ts";
 import { Group } from "../group.ts";
 import * as Path from "../path.ts";
@@ -7,6 +8,7 @@ import type { Track } from "../track.ts";
 import { error } from "../util/error.ts";
 import { Announce, AnnounceInit, AnnounceInterest } from "./announce.ts";
 import type { Group as GroupMessage } from "./group.ts";
+import { Probe } from "./probe.ts";
 import { StreamId } from "./stream.ts";
 import { decodeSubscribeResponse, Subscribe } from "./subscribe.ts";
 import { Version } from "./version.ts";
@@ -26,15 +28,21 @@ export class Subscriber {
 	#subscribes = new Map<bigint, Track>();
 	#subscribeNext = 0n;
 
+	// Recv bandwidth producer (Lite03+ only).
+	#recvBandwidth?: Bandwidth;
+
 	/**
 	 * Creates a new Subscriber instance.
 	 * @param quic - The WebTransport session to use
+	 * @param version - The protocol version
+	 * @param recvBandwidth - Optional bandwidth producer for PROBE
 	 *
 	 * @internal
 	 */
-	constructor(quic: WebTransport, version: Version) {
+	constructor(quic: WebTransport, version: Version, recvBandwidth?: Bandwidth) {
 		this.#quic = quic;
 		this.version = version;
+		this.#recvBandwidth = recvBandwidth;
 	}
 
 	/**
@@ -195,6 +203,27 @@ export class Subscriber {
 		}
 	}
 
+	/**
+	 * Opens a PROBE bidi stream to receive bandwidth estimates from the publisher.
+	 * Returns immediately if recv bandwidth is not supported.
+	 * Errors are fatal and propagate to the connection.
+	 *
+	 * @internal
+	 */
+	async runProbe(): Promise<void> {
+		if (!this.#recvBandwidth) return;
+		if (this.version === Version.DRAFT_01 || this.version === Version.DRAFT_02) return;
+
+		const stream = await Stream.open(this.#quic);
+		await stream.writer.u53(StreamId.Probe);
+
+		for (;;) {
+			const probe = await Probe.decodeMaybe(stream.reader, this.version);
+			if (!probe) break;
+			this.#recvBandwidth.set(probe.bitrate);
+		}
+	}
+
 	close() {
 		for (const track of this.#subscribes.values()) {
 			track.close();

js/publish/src/broadcast.ts

@@ -43,7 +43,7 @@ export class Broadcast {
 		this.name = Signal.from(props?.name ?? Moq.Path.empty());
 
 		this.audio = new Audio.Encoder(props?.audio);
-		this.video = new Video.Root(props?.video);
+		this.video = new Video.Root({ ...props?.video, connection: this.connection });
 		this.location = new Location.Root(props?.location);
 		this.chat = new Chat.Root(props?.chat);
 		this.preview = new Preview(props?.preview);

js/publish/src/video/encoder.ts

@@ -57,9 +57,18 @@ export class Encoder {
 	// True when the encoder is actively serving a track.
 	active = new Signal<boolean>(false);
 
-	constructor(frame: Getter<VideoFrame | undefined>, source: Signal<Source | undefined>, props?: EncoderProps) {
+	// Connection signal for reading send bandwidth.
+	connection: Getter<Moq.Connection.Established | undefined>;
+
+	constructor(
+		frame: Getter<VideoFrame | undefined>,
+		source: Signal<Source | undefined>,
+		connection: Getter<Moq.Connection.Established | undefined>,
+		props?: EncoderProps,
+	) {
 		this.frame = frame;
 		this.source = source;
+		this.connection = connection;
 		this.enabled = Signal.from(props?.enabled ?? false);
 		this.config = Signal.from(props?.config);
 
@@ -205,6 +214,20 @@ export class Encoder {
 
 			bitrate = Math.round(Math.min(bitrate, user.maxBitrate || bitrate));
 
+			// If no explicit maxBitrate, cap to the estimated send bandwidth (with 90% safety margin).
+			if (!user.maxBitrate) {
+				const conn = effect.get(this.connection);
+				const sendBw = conn?.sendBandwidth;
+				if (sendBw) {
+					const estimate = effect.get(sendBw);
+					if (estimate != null) {
+						// Reserve ~10% for audio and protocol overhead.
+						const cap = Math.round(estimate * 0.9);
+						bitrate = Math.min(bitrate, cap);
+					}
+				}
+			}
+
 			const config: VideoEncoderConfig = {
 				codec,
 				width: dimensions.width,