Revise bitrate estimation API based on feedback

- BandwidthProducer: add close(err)/closed() for cancellation, rename
  get() to peek()
- Move send bandwidth polling to Session (version-agnostic, depends on
  QUIC backend support not protocol version)
- Remove send bandwidth from lite/ietf start() return values
- Make recv bandwidth errors fatal (no retry loop)
- Use forward-compatible version matching (list older versions explicitly)
- JS: make send bandwidth version-agnostic (check getStats existence),
  use setInterval with closed promise instead of fire-and-forget
- JS: make probe async and fatal, called from connection Promise.all
- JS watch: handle target.name early before ABR, remove MIN_BITRATE
- JS publish: read sendBandwidth from connection directly instead of
  piping through props
- CLAUDE.md: add JS async patterns convention (Effect.interval/timer)

https://claude.ai/code/session_01C68pEkoUhQqtQALYxRajhx

Author: claude

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/lite/connection.ts

@@ -1,5 +1,5 @@
 import type { Announced } from "../announced.ts";
-import { type Bandwidth, createBandwidth } from "../bandwidth.ts";
+import type { Bandwidth } from "../bandwidth.ts";
 import type { Broadcast } from "../broadcast.ts";
 import type { Established } from "../connection/established.ts";
 import * as Path from "../path.ts";
@@ -42,9 +42,6 @@ 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;
 
@@ -66,10 +63,15 @@ 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 = createBandwidth();
-			this.recvBandwidth = createBandwidth();
+		// 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 = new Bandwidth();
+		}
+
+		// Recv bandwidth requires PROBE support (not available in older drafts).
+		if (version !== Version.DRAFT_01 && version !== Version.DRAFT_02) {
+			this.recvBandwidth = new Bandwidth();
 		}
 
 		this.#publisher = new Publisher(this.#quic, this.#version);
@@ -82,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();
 
@@ -97,21 +96,20 @@ 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()];
 
-		// Start polling send bandwidth if supported.
 		if (this.sendBandwidth) {
-			this.#runSendBandwidth(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();
 		}
@@ -233,31 +231,31 @@ export class Connection implements Established {
 
 	/**
 	 * Polls the QUIC congestion controller for estimated send rate.
+	 * Resolves when the connection is closed.
 	 */
-	#runSendBandwidth(bandwidth: Bandwidth) {
+	async #runSendBandwidth(bandwidth: Bandwidth): Promise<void> {
 		// getStats is not yet in the TypeScript WebTransport type definitions.
 		const quic = this.#quic as unknown as {
-			getStats?: () => Promise<{ estimatedSendRate: number | null }>;
+			getStats: () => Promise<{ estimatedSendRate: number | null }>;
 		};
 
-		const getStats = quic.getStats?.bind(quic);
-		if (!getStats) return;
-
-		const run = async () => {
-			while (!this.#closed) {
-				await new Promise<void>((resolve) => setTimeout(resolve, SEND_BW_POLL_INTERVAL));
-				if (this.#closed) break;
-
+		return new Promise<void>((resolve) => {
+			const id = setInterval(async () => {
 				try {
-					const stats = await getStats();
+					const stats = await quic.getStats();
 					bandwidth.set(stats.estimatedSendRate ?? undefined);
 				} catch {
-					if (this.#closed) break;
+					// Connection likely closed.
+					clearInterval(id);
+					resolve();
 				}
-			}
-		};
+			}, SEND_BW_POLL_INTERVAL);
 
-		void run();
+			void this.closed.then(() => {
+				clearInterval(id);
+				resolve();
+			});
+		});
 	}
 
 	/**

js/lite/src/lite/subscriber.ts

@@ -35,18 +35,14 @@ export class Subscriber {
 	 * 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)
+	 * @param recvBandwidth - Optional bandwidth producer for PROBE
 	 *
 	 * @internal
 	 */
 	constructor(quic: WebTransport, version: Version, recvBandwidth?: Bandwidth) {
 		this.#quic = quic;
 		this.version = version;
 		this.#recvBandwidth = recvBandwidth;
-
-		if (recvBandwidth && version === Version.DRAFT_03) {
-			this.#runProbe();
-		}
 	}
 
 	/**
@@ -211,36 +207,23 @@ export class Subscriber {
 
 	/**
 	 * Opens a PROBE bidi stream to receive bandwidth estimates from the publisher.
-	 * Retries on error.
+	 * Returns immediately if recv bandwidth is not supported.
+	 * Errors are fatal and propagate to the connection.
+	 *
+	 * @internal
 	 */
-	#runProbe() {
-		const run = async () => {
-			while (!this.#closed) {
-				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}`);
-				}
+	async runProbe(): Promise<void> {
+		if (!this.#recvBandwidth) return;
+		if (this.version === Version.DRAFT_01 || this.version === Version.DRAFT_02) return;
 
-				// Clear the bitrate on disconnect.
-				this.#recvBandwidth?.set(undefined);
-
-				if (this.#closed) break;
-
-				// Wait before retrying.
-				await new Promise((resolve) => setTimeout(resolve, 1000));
-			}
-		};
+		const stream = await Stream.open(this.#quic);
+		await stream.writer.u53(StreamId.Probe);
 
-		void run();
+		for (;;) {
+			const probe = await Probe.decodeMaybe(stream.reader, this.version);
+			if (!probe) break;
+			this.#recvBandwidth.set(probe.bitrate);
+		}
 	}
 
 	close() {

js/publish/src/broadcast.ts

@@ -35,9 +35,6 @@ 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) {
@@ -46,25 +43,13 @@ 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, sendBandwidth: this.#sendBandwidth });
+		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);
 		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);
-		effect.set(this.#sendBandwidth, estimate);
 	}
 
 	#run(effect: Effect) {

js/publish/src/video/encoder.ts

@@ -10,10 +10,6 @@ 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?
@@ -61,15 +57,20 @@ 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>;
+	// Connection signal for reading send bandwidth.
+	connection: Getter<Moq.Connection.Established | undefined>;
 
-	constructor(frame: Getter<VideoFrame | undefined>, source: Signal<Source | undefined>, props?: EncoderProps) {
+	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);
-		this.sendBandwidth = props?.sendBandwidth ?? new Signal<number | undefined>(undefined);
 
 		this.#signals.run(this.#runCatalog.bind(this));
 		this.#signals.run(this.#runConfig.bind(this));
@@ -213,13 +214,15 @@ export class Encoder {
 
 			// 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 conn = effect.get(this.connection);
+				const sendBw = conn?.sendBandwidth;
+				if (sendBw) {
+					const estimate = effect.get(sendBw.signal);
+					if (estimate != null) {
+						// Reserve ~10% for audio and protocol overhead.
+						const cap = Math.round(estimate * 0.9);
+						bitrate = Math.min(bitrate, cap);
+					}
 				}
 			}
 

js/publish/src/video/index.ts

@@ -1,4 +1,5 @@
 import * as Catalog from "@moq/hang/catalog";
+import type * as Moq from "@moq/lite";
 import { Effect, type Getter, Signal } from "@moq/signals";
 import { Encoder, type EncoderProps } from "./encoder";
 import { TrackProcessor } from "./polyfill";
@@ -12,7 +13,7 @@ export type Props = {
 	hd?: EncoderProps;
 	sd?: EncoderProps;
 	flip?: boolean | Signal<boolean>;
-	sendBandwidth?: Getter<number | undefined>;
+	connection?: Getter<Moq.Connection.Established | undefined>;
 };
 
 export class Root {
@@ -35,10 +36,9 @@ export class Root {
 	constructor(props?: Props) {
 		this.source = Signal.from(props?.source);
 
-		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);
+		const connection = props?.connection ?? new Signal<Moq.Connection.Established | undefined>(undefined);
+		this.hd = new Encoder(this.frame, this.source, connection, props?.hd);
+		this.sd = new Encoder(this.frame, this.source, connection, props?.sd);
 
 		this.flip = Signal.from(props?.flip ?? false);
 

js/watch/src/video/source.ts

@@ -212,9 +212,18 @@ export class Source {
 
 		const target = effect.get(this.target);
 
-		// If no explicit bitrate target, use the recv bandwidth estimate from the connection.
+		// Manual selection by name — skip all ABR logic.
+		if (target?.name && target.name in available) {
+			const config = available[target.name];
+			effect.set(this.#track, target.name);
+			effect.set(this.#config, config);
+			effect.set(this.sync.video, config.jitter as Moq.Time.Milli | undefined);
+			return;
+		}
+
+		// Auto-select: use recv bandwidth if no explicit bitrate target.
 		let effectiveTarget = target;
-		if (!target?.bitrate && !target?.name) {
+		if (!target?.bitrate) {
 			const broadcast = effect.get(this.broadcast);
 			const connection = broadcast ? effect.get(broadcast.connection) : undefined;
 			const recvBw = connection?.recvBandwidth;
@@ -228,9 +237,7 @@ export class Source {
 			}
 		}
 
-		// Manual selection by name
-		const manual = effectiveTarget?.name;
-		const selected = manual && manual in available ? manual : this.#select(available, effectiveTarget);
+		const selected = this.#select(available, effectiveTarget);
 		if (!selected) return;
 
 		const config = available[selected];