Fix bandwidth estimation: use Signal directly, fix version matching and lifecycle bugs

- Replace Bandwidth wrapper class with Signal<number | undefined> type alias
- Fix Rust version matching to default forward for newest drafts (per convention)
- Guard tokio::select! recv_bandwidth arm to prevent early exit on Lite01/Lite02
- Store BandwidthProducer in Session instead of BandwidthConsumer (enables lazy polling)
- Add closed flag to JS Subscriber to stop probe loop on close()
- Move error handling inside send bandwidth polling loop (per-iteration, not outer)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: kixelated

js/lite/src/bandwidth.ts

@@ -1,33 +1,14 @@
-import { type Getter, Signal } from "@moq/signals";
+import { Signal } from "@moq/signals";
 
 /**
- * A bandwidth estimate that can be read synchronously or observed reactively.
+ * A bandwidth estimate in bits per second, or undefined if unknown.
  *
- * Created internally by the connection. Consumers read from it via the signal.
+ * This is a Signal that can be read synchronously via `peek()`,
+ * observed reactively via `effect.get()`, or updated via `set()`.
  */
-export class Bandwidth {
-	#bitrate = new Signal<number | undefined>(undefined);
+export type Bandwidth = Signal<number | 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);
-		});
-	}
+/** Create a new bandwidth signal. */
+export function createBandwidth(): Bandwidth {
+	return new Signal<number | undefined>(undefined);
 }

js/lite/src/lite/connection.ts

@@ -1,5 +1,5 @@
 import type { Announced } from "../announced.ts";
-import { Bandwidth } from "../bandwidth.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";
@@ -68,8 +68,8 @@ export class Connection implements Established {
 
 		// Set up bandwidth estimation for Lite03+.
 		if (version === Version.DRAFT_03) {
-			this.sendBandwidth = new Bandwidth();
-			this.recvBandwidth = new Bandwidth();
+			this.sendBandwidth = createBandwidth();
+			this.recvBandwidth = createBandwidth();
 		}
 
 		this.#publisher = new Publisher(this.#quic, this.#version);
@@ -244,18 +244,16 @@ export class Connection implements Established {
 		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;
+			while (!this.#closed) {
+				await new Promise<void>((resolve) => setTimeout(resolve, SEND_BW_POLL_INTERVAL));
+				if (this.#closed) break;
 
+				try {
 					const stats = await getStats();
 					bandwidth.set(stats.estimatedSendRate ?? undefined);
+				} catch {
+					if (this.#closed) break;
 				}
-			} catch {
-				// Connection closed.
 			}
 		};
 

js/lite/src/lite/subscriber.ts

@@ -207,13 +207,15 @@ export class Subscriber {
 		}
 	}
 
+	#closed = false;
+
 	/**
 	 * Opens a PROBE bidi stream to receive bandwidth estimates from the publisher.
 	 * Retries on error.
 	 */
 	#runProbe() {
 		const run = async () => {
-			for (;;) {
+			while (!this.#closed) {
 				try {
 					const stream = await Stream.open(this.#quic);
 					await stream.writer.u53(StreamId.Probe);
@@ -231,6 +233,8 @@ export class Subscriber {
 				// Clear the bitrate on disconnect.
 				this.#recvBandwidth?.set(undefined);
 
+				if (this.#closed) break;
+
 				// Wait before retrying.
 				await new Promise((resolve) => setTimeout(resolve, 1000));
 			}
@@ -240,6 +244,7 @@ export class Subscriber {
 	}
 
 	close() {
+		this.#closed = true;
 		for (const track of this.#subscribes.values()) {
 			track.close();
 		}

js/publish/src/broadcast.ts

@@ -63,7 +63,7 @@ export class Broadcast {
 			return;
 		}
 
-		const estimate = effect.get(connection.sendBandwidth.signal);
+		const estimate = effect.get(connection.sendBandwidth);
 		effect.set(this.#sendBandwidth, estimate);
 	}
 

js/watch/src/video/source.ts

@@ -219,7 +219,7 @@ export class Source {
 			const connection = broadcast ? effect.get(broadcast.connection) : undefined;
 			const recvBw = connection?.recvBandwidth;
 			if (recvBw) {
-				const estimate = effect.get(recvBw.signal);
+				const estimate = effect.get(recvBw);
 				if (estimate != null) {
 					// Apply a safety margin (80%) to avoid oscillation.
 					const safeBitrate = Math.round(estimate * 0.8);

rs/moq-lite/src/ietf/session.rs

@@ -1,5 +1,5 @@
 use crate::{
-	BandwidthConsumer, Error, OriginConsumer, OriginProducer,
+	BandwidthProducer, Error, OriginConsumer, OriginProducer,
 	coding::{Encode, Reader, Stream, Writer},
 	ietf::{self, FetchHeader, GroupFlags, RequestId},
 	setup,
@@ -8,7 +8,7 @@ use crate::{
 use super::{Control, Message, Publisher, Subscriber, Version, adapter::ControlStreamAdapter};
 
 /// Returned by `start()`: (send_bandwidth, recv_bandwidth)
-pub type Bandwidth = (Option<BandwidthConsumer>, Option<BandwidthConsumer>);
+pub type Bandwidth = (Option<BandwidthProducer>, Option<BandwidthProducer>);
 
 pub fn start<S: web_transport_trait::Session>(
 	session: S,

rs/moq-lite/src/lite/session.rs

@@ -2,14 +2,12 @@ use std::time::Duration;
 
 use web_transport_trait::Stats;
 
-use crate::{
-	BandwidthConsumer, BandwidthProducer, Error, OriginConsumer, OriginProducer, coding::Stream, lite::SessionInfo,
-};
+use crate::{BandwidthProducer, Error, OriginConsumer, OriginProducer, coding::Stream, lite::SessionInfo};
 
 use super::{Publisher, Subscriber, Version};
 
 /// Returned by `start()`: (send_bandwidth, recv_bandwidth)
-pub type Bandwidth = (Option<BandwidthConsumer>, Option<BandwidthConsumer>);
+pub type Bandwidth = (Option<BandwidthProducer>, Option<BandwidthProducer>);
 
 pub fn start<S: web_transport_trait::Session>(
 	session: S,
@@ -24,17 +22,12 @@ pub fn start<S: web_transport_trait::Session>(
 	version: Version,
 ) -> Result<Bandwidth, Error> {
 	let send_bw = BandwidthProducer::new();
-	let send_bw_consumer = send_bw.consume();
+	let send_bw_task = send_bw.clone();
 
 	let recv_bw = BandwidthProducer::new();
-	let recv_bw_consumer = match version {
-		Version::Lite03 => Some(recv_bw.consume()),
-		_ => None,
-	};
-
-	let recv_bw_for_sub = match version {
-		Version::Lite03 => Some(recv_bw),
-		_ => None,
+	let (recv_bw_ret, recv_bw_for_sub) = match version {
+		Version::Lite01 | Version::Lite02 => (None, None),
+		_ => (Some(recv_bw.clone()), Some(recv_bw)),
 	};
 
 	let publisher = Publisher::new(session.clone(), publish, version);
@@ -45,7 +38,7 @@ pub fn start<S: web_transport_trait::Session>(
 			Err(res) = run_session(setup) => Err(res),
 			res = publisher.run() => res,
 			res = subscriber.run() => res,
-			_ = run_send_bandwidth(&session, send_bw) => Ok(()),
+			_ = run_send_bandwidth(&session, send_bw_task) => Ok(()),
 		};
 
 		match res {
@@ -64,7 +57,7 @@ pub fn start<S: web_transport_trait::Session>(
 		}
 	});
 
-	Ok((Some(send_bw_consumer), recv_bw_consumer))
+	Ok((Some(send_bw), recv_bw_ret))
 }
 
 /// Polls the QUIC congestion controller for estimated send rate.

rs/moq-lite/src/lite/subscriber.rs

@@ -44,11 +44,12 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 	}
 
 	pub async fn run(self) -> Result<(), Error> {
+		let recv_bandwidth_enabled = self.recv_bandwidth.is_some();
 		let bw = self.clone();
 		tokio::select! {
 			Err(err) = self.clone().run_announce() => Err(err),
 			res = self.run_uni() => res,
-			_ = bw.run_recv_bandwidth() => Ok(()),
+			_ = bw.run_recv_bandwidth(), if recv_bandwidth_enabled => Ok(()),
 		}
 	}
 

rs/moq-lite/src/session.rs

@@ -1,6 +1,6 @@
 use std::{future::Future, pin::Pin, sync::Arc};
 
-use crate::{BandwidthConsumer, Error, Version};
+use crate::{BandwidthConsumer, BandwidthProducer, Error, Version};
 
 /// A MoQ transport session, wrapping a WebTransport connection.
 ///
@@ -11,17 +11,17 @@ use crate::{BandwidthConsumer, Error, Version};
 pub struct Session {
 	session: Arc<dyn SessionInner>,
 	version: Version,
-	send_bandwidth: Option<BandwidthConsumer>,
-	recv_bandwidth: Option<BandwidthConsumer>,
+	send_bandwidth: Option<BandwidthProducer>,
+	recv_bandwidth: Option<BandwidthProducer>,
 	closed: bool,
 }
 
 impl Session {
 	pub(super) fn new<S: web_transport_trait::Session>(
 		session: S,
 		version: Version,
-		send_bandwidth: Option<BandwidthConsumer>,
-		recv_bandwidth: Option<BandwidthConsumer>,
+		send_bandwidth: Option<BandwidthProducer>,
+		recv_bandwidth: Option<BandwidthProducer>,
 	) -> Self {
 		Self {
 			session: Arc::new(session),
@@ -41,14 +41,14 @@ impl Session {
 	///
 	/// Returns `None` if the QUIC backend or MoQ version doesn't support bandwidth estimation.
 	pub fn send_bandwidth(&self) -> Option<BandwidthConsumer> {
-		self.send_bandwidth.clone()
+		self.send_bandwidth.as_ref().map(|p| p.consume())
 	}
 
 	/// Returns a consumer for the estimated receive bitrate (from PROBE).
 	///
 	/// Returns `None` if the MoQ version doesn't support PROBE (requires moq-lite-03+).
 	pub fn recv_bandwidth(&self) -> Option<BandwidthConsumer> {
-		self.recv_bandwidth.clone()
+		self.recv_bandwidth.as_ref().map(|p| p.consume())
 	}
 
 	/// Close the underlying transport session.