Add bitrate estimation API for send and receive rates

Expose BandwidthProducer/Consumer pairs on Session (Rust) and Connection (JS)
for estimated send rate (from congestion controller, polled every 100ms) and
receive rate (from PROBE mechanism, moq-lite-03+ only). The subscriber now
initiates PROBE streams lazily when consumers exist. @moq/watch uses recv
bandwidth for automatic ABR rendition selection, and @moq/publish caps
encoder bitrate to the estimated send bandwidth.

https://claude.ai/code/session_01C68pEkoUhQqtQALYxRajhx

Author: claude

js/lite/src/bandwidth.ts

@@ -0,0 +1,33 @@
+import { type Getter, Signal } from "@moq/signals";
+
+/**
+ * A bandwidth estimate that can be read synchronously or observed reactively.
+ *
+ * Created internally by the connection. Consumers read from it via the signal.
+ */
+export class Bandwidth {
+	#bitrate = new Signal<number | undefined>(undefined);
+
+	/** Reactive signal for the current bandwidth estimate in bits per second. */
+	readonly signal: Getter<number | undefined> = this.#bitrate;
+
+	/**
+	 * Update the bandwidth estimate. Called internally by the connection/subscriber.
+	 * @internal
+	 */
+	set(bitrate: number | undefined): void {
+		this.#bitrate.set(bitrate);
+	}
+
+	/** Get the current bandwidth estimate synchronously. */
+	get(): number | undefined {
+		return this.#bitrate.peek();
+	}
+
+	/** Wait for the bandwidth estimate to change. */
+	async changed(): Promise<number | undefined> {
+		return new Promise<number | undefined>((resolve) => {
+			this.#bitrate.changed(resolve);
+		});
+	}
+}

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 { Bandwidth } 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.
@@ -42,6 +45,12 @@ export class Connection implements Established {
 	// 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.
 	 * @param url - The URL of the connection
@@ -57,8 +66,14 @@ export class Connection implements Established {
 		this.version = versionName(version);
 		this.#version = version;
 
+		// Set up bandwidth estimation for Lite03+.
+		if (version === Version.DRAFT_03) {
+			this.sendBandwidth = new Bandwidth();
+			this.recvBandwidth = new Bandwidth();
+		}
+
 		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();
 	}
@@ -86,6 +101,11 @@ export class Connection implements Established {
 		const bidis = this.#runBidis();
 		const unis = this.#runUnis();
 
+		// Start polling send bandwidth if supported.
+		if (this.sendBandwidth) {
+			this.#runSendBandwidth(this.sendBandwidth);
+		}
+
 		try {
 			await Promise.all([session, bidis, unis]);
 		} catch (err) {
@@ -211,6 +231,37 @@ export class Connection implements Established {
 		}
 	}
 
+	/**
+	 * Polls the QUIC congestion controller for estimated send rate.
+	 */
+	#runSendBandwidth(bandwidth: Bandwidth) {
+		// getStats is not yet in the TypeScript WebTransport type definitions.
+		const quic = this.#quic as unknown as {
+			getStats?: () => Promise<{ estimatedSendRate: number | null }>;
+		};
+
+		const getStats = quic.getStats?.bind(quic);
+		if (!getStats) return;
+
+		const run = async () => {
+			try {
+				while (!this.#closed) {
+					const timeout = new Promise<void>((resolve) => setTimeout(resolve, SEND_BW_POLL_INTERVAL));
+					await timeout;
+
+					if (this.#closed) break;
+
+					const stats = await getStats();
+					bandwidth.set(stats.estimatedSendRate ?? undefined);
+				}
+			} catch {
+				// Connection closed.
+			}
+		};
+
+		void run();
+	}
+
 	/**
 	 * Returns a promise that resolves when the connection is closed.
 	 * @returns A promise that resolves when closed

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,25 @@ 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 (Lite03+ only)
 	 *
 	 * @internal
 	 */
-	constructor(quic: WebTransport, version: Version) {
+	constructor(quic: WebTransport, version: Version, recvBandwidth?: Bandwidth) {
 		this.#quic = quic;
 		this.version = version;
+		this.#recvBandwidth = recvBandwidth;
+
+		if (recvBandwidth && version === Version.DRAFT_03) {
+			this.#runProbe();
+		}
 	}
 
 	/**
@@ -195,6 +207,38 @@ export class Subscriber {
 		}
 	}
 
+	/**
+	 * Opens a PROBE bidi stream to receive bandwidth estimates from the publisher.
+	 * Retries on error.
+	 */
+	#runProbe() {
+		const run = async () => {
+			for (;;) {
+				try {
+					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);
+					}
+				} catch (err: unknown) {
+					const e = error(err);
+					console.debug(`probe recv error: ${e.message}`);
+				}
+
+				// Clear the bitrate on disconnect.
+				this.#recvBandwidth?.set(undefined);
+
+				// Wait before retrying.
+				await new Promise((resolve) => setTimeout(resolve, 1000));
+			}
+		};
+
+		void run();
+	}
+
 	close() {
 		for (const track of this.#subscribes.values()) {
 			track.close();

js/publish/src/broadcast.ts

@@ -35,6 +35,9 @@ export class Broadcast {
 	preview: Preview;
 	user: User.Info;
 
+	// Derived signal tracking the connection's send bandwidth estimate.
+	#sendBandwidth = new Signal<number | undefined>(undefined);
+
 	signals = new Effect();
 
 	constructor(props?: BroadcastProps) {
@@ -43,13 +46,25 @@ 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, sendBandwidth: this.#sendBandwidth });
 		this.location = new Location.Root(props?.location);
 		this.chat = new Chat.Root(props?.chat);
 		this.preview = new Preview(props?.preview);
 		this.user = new User.Info(props?.user);
 
 		this.signals.run(this.#run.bind(this));
+		this.signals.run(this.#runSendBandwidth.bind(this));
+	}
+
+	#runSendBandwidth(effect: Effect) {
+		const connection = effect.get(this.connection);
+		if (!connection?.sendBandwidth) {
+			effect.set(this.#sendBandwidth, undefined);
+			return;
+		}
+
+		const estimate = effect.get(connection.sendBandwidth.signal);
+		effect.set(this.#sendBandwidth, estimate);
 	}
 
 	#run(effect: Effect) {

js/publish/src/video/encoder.ts

@@ -10,6 +10,10 @@ export interface EncoderProps {
 	enabled?: boolean | Signal<boolean>;
 	config?: EncoderConfig | Signal<EncoderConfig | undefined>;
 	container?: Catalog.Container;
+
+	// Optional: estimated send bandwidth in bps. When set and no explicit maxBitrate
+	// is configured, the encoder will cap its bitrate to this value.
+	sendBandwidth?: Getter<number | undefined>;
 }
 
 // TODO support signals?
@@ -57,11 +61,15 @@ export class Encoder {
 	// True when the encoder is actively serving a track.
 	active = new Signal<boolean>(false);
 
+	// Optional: estimated send bandwidth for adaptive bitrate capping.
+	sendBandwidth: Getter<number | undefined>;
+
 	constructor(frame: Getter<VideoFrame | undefined>, source: Signal<Source | undefined>, props?: EncoderProps) {
 		this.frame = frame;
 		this.source = source;
 		this.enabled = Signal.from(props?.enabled ?? false);
 		this.config = Signal.from(props?.config);
+		this.sendBandwidth = props?.sendBandwidth ?? new Signal<number | undefined>(undefined);
 
 		this.#signals.run(this.#runCatalog.bind(this));
 		this.#signals.run(this.#runConfig.bind(this));
@@ -203,6 +211,18 @@ 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 sendBw = effect.get(this.sendBandwidth);
+				if (sendBw != null) {
+					// Reserve ~10% for audio and protocol overhead.
+					const cap = Math.round(sendBw * 0.9);
+					// Don't go below 100kbps to keep encoding usable.
+					const MIN_BITRATE = 100_000;
+					bitrate = Math.max(MIN_BITRATE, Math.min(bitrate, cap));
+				}
+			}
+
 			const config: VideoEncoderConfig = {
 				codec,
 				width: dimensions.width,

js/publish/src/video/index.ts

@@ -1,5 +1,5 @@
 import * as Catalog from "@moq/hang/catalog";
-import { Effect, Signal } from "@moq/signals";
+import { Effect, type Getter, Signal } from "@moq/signals";
 import { Encoder, type EncoderProps } from "./encoder";
 import { TrackProcessor } from "./polyfill";
 import type { Source } from "./types";
@@ -12,6 +12,7 @@ export type Props = {
 	hd?: EncoderProps;
 	sd?: EncoderProps;
 	flip?: boolean | Signal<boolean>;
+	sendBandwidth?: Getter<number | undefined>;
 };
 
 export class Root {
@@ -34,8 +35,10 @@ export class Root {
 	constructor(props?: Props) {
 		this.source = Signal.from(props?.source);
 
-		this.hd = new Encoder(this.frame, this.source, props?.hd);
-		this.sd = new Encoder(this.frame, this.source, props?.sd);
+		const hdProps = props?.sendBandwidth ? { ...props?.hd, sendBandwidth: props.sendBandwidth } : props?.hd;
+		const sdProps = props?.sendBandwidth ? { ...props?.sd, sendBandwidth: props.sendBandwidth } : props?.sd;
+		this.hd = new Encoder(this.frame, this.source, hdProps);
+		this.sd = new Encoder(this.frame, this.source, sdProps);
 
 		this.flip = Signal.from(props?.flip ?? false);
 

js/watch/src/video/source.ts

@@ -212,9 +212,25 @@ export class Source {
 
 		const target = effect.get(this.target);
 
+		// If no explicit bitrate target, use the recv bandwidth estimate from the connection.
+		let effectiveTarget = target;
+		if (!target?.bitrate && !target?.name) {
+			const broadcast = effect.get(this.broadcast);
+			const connection = broadcast ? effect.get(broadcast.connection) : undefined;
+			const recvBw = connection?.recvBandwidth;
+			if (recvBw) {
+				const estimate = effect.get(recvBw.signal);
+				if (estimate != null) {
+					// Apply a safety margin (80%) to avoid oscillation.
+					const safeBitrate = Math.round(estimate * 0.8);
+					effectiveTarget = { ...target, bitrate: safeBitrate };
+				}
+			}
+		}
+
 		// Manual selection by name
-		const manual = target?.name;
-		const selected = manual && manual in available ? manual : this.#select(available, target);
+		const manual = effectiveTarget?.name;
+		const selected = manual && manual in available ? manual : this.#select(available, effectiveTarget);
 		if (!selected) return;
 
 		const config = available[selected];