moq_native/

reconnect.rs

use std::task::{Poll, ready};
use std::time::Duration;

use moq_net::Version;
use moq_net::bandwidth::{Consumer as BandwidthConsumer, Producer as BandwidthProducer};
use moq_net::kio;
use rand::RngExt;
use url::Url;

use crate::{Client, Error, RedactedUrl};

/// Exponential backoff configuration for reconnection attempts.
///
/// This decides how long to wait between reconnect attempts and when to give up. The delays carry
/// jitter, so a fleet knocked offline together doesn't reconnect in lockstep.
///
/// [`timeout`](Self::timeout) is what ends a hopeless loop: every failure rides the same backoff,
/// and the short default budget is what surfaces a broken target instead of hiding it. The only
/// failures short-circuited are answers a server actually gave (an auth rejection, or a CONNECT
/// status that isn't an invitation to retry). A zero timeout removes the backstop, so it belongs
/// only where an unattended process must outlive an outage of any length.
#[derive(Clone, Debug, clap::Args, serde::Serialize, serde::Deserialize)]
#[serde(default, deny_unknown_fields)]
#[non_exhaustive]
pub struct Backoff {
	/// Initial delay before first reconnect attempt.
	#[arg(
		id = "backoff-initial",
		long,
		default_value = "1s",
		env = "MOQ_BACKOFF_INITIAL",
		value_parser = humantime::parse_duration,
	)]
	#[serde(with = "humantime_serde")]
	pub initial: Duration,

	/// Multiplier applied to delay after each failure.
	#[arg(id = "backoff-multiplier", long, default_value_t = 2, env = "MOQ_BACKOFF_MULTIPLIER")]
	pub multiplier: u32,

	/// Maximum delay between reconnect attempts.
	#[arg(
		id = "backoff-max",
		long,
		default_value = "5s",
		env = "MOQ_BACKOFF_MAX",
		value_parser = humantime::parse_duration,
	)]
	#[serde(with = "humantime_serde")]
	pub max: Duration,

	/// Maximum time to spend retrying before giving up.
	/// Resets after a stable connection (one that outlives the initial backoff), so a flapping
	/// session that reconnects then immediately drops still counts toward the timeout. Set to 0 for
	/// unlimited retries.
	#[arg(
		id = "backoff-timeout",
		long,
		default_value = "10s",
		env = "MOQ_BACKOFF_TIMEOUT",
		value_parser = humantime::parse_duration,
	)]
	#[serde(with = "humantime_serde")]
	pub timeout: Duration,
}

impl Default for Backoff {
	fn default() -> Self {
		Self {
			initial: Duration::from_secs(1),
			multiplier: 2,
			max: Duration::from_secs(5),
			timeout: Duration::from_secs(10),
		}
	}
}

impl Backoff {
	/// Reject a backoff that would retry without pacing.
	///
	/// The loop sleeps `delay`, then `delay = min(delay * multiplier, max)`. A zero in
	/// any of the three collapses that to zero forever, so a relay that is simply down
	/// becomes a hot loop of dials and DNS lookups, unbounded when
	/// [`timeout`](Self::timeout) is also zero. A zero `timeout` on its own is fine and
	/// documented: it means retry forever, which is only a problem unpaced.
	///
	/// A `multiplier` of 1 is allowed: the delay stays at `initial`, which is a
	/// constant-delay retry rather than an unpaced one.
	pub(crate) fn validate(&self) -> crate::Result<()> {
		match self.initial.is_zero() || self.multiplier == 0 || self.max.is_zero() {
			true => Err(crate::Error::BackoffUnpaced),
			false => Ok(()),
		}
	}

	/// Grow the retry delay without overflowing before applying the configured cap.
	fn next_delay(&self, delay: Duration) -> Duration {
		delay.saturating_mul(self.multiplier.max(1)).min(self.max)
	}

	/// How long broadcasts fed by a reconnecting session should outlive a session
	/// drop (see [`moq_net::origin::Info::linger`]): slightly past the give-up
	/// [`timeout`](Self::timeout), so when the loop does give up its error surfaces
	/// before the broadcasts tear down. A zero timeout retries forever, so the
	/// broadcasts linger forever too.
	pub fn linger(&self) -> Duration {
		match self.timeout.is_zero() {
			true => Duration::MAX,
			false => self.timeout.saturating_add(Duration::from_secs(1)),
		}
	}
}

/// When a reconnect sequence gives up, or `None` when [`Backoff::timeout`] is zero and it never
/// does. Measured from now, so it covers the connect attempts as well as the waits between them.
fn deadline_from(backoff: &Backoff) -> Option<tokio::time::Instant> {
	match backoff.timeout.is_zero() {
		true => None,
		false => Some(tokio::time::Instant::now() + backoff.timeout),
	}
}

/// A connection lifecycle transition reported by [`Reconnect::status`].
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum Status {
	/// A session connected (the first connect, or a reconnect after a drop).
	Connected,
	/// An established session dropped; a reconnect attempt follows.
	Disconnected,
}

/// Shared reconnect state, observed by consumers through a [`kio`] channel.
///
/// The channel closing (all producers dropped) is the terminal signal; `error`
/// distinguishes a permanent give-up from a graceful close.
#[derive(Default)]
struct State {
	/// Current connection status, or `None` before the first connect.
	status: Option<Status>,
	/// Cumulative connects and disconnects, bumped by the reconnect loop itself so a session that
	/// connects and drops before a consumer polls still counts.
	presence: moq_net::stats::Presence,
	/// The negotiated MoQ version of the live session, or `None` when disconnected.
	version: Option<Version>,
	/// Set when the reconnect loop permanently gives up: the backoff timeout expiring, or a server
	/// answer that redialing cannot change.
	error: Option<Error>,
	/// The currently-connected session, or `None` while reconnecting. Read by
	/// [`ConnectionStatsReader`] to snapshot live connection stats.
	session: Option<moq_net::Session>,
}

/// Statistics and protocol sampled from the same live connection.
#[non_exhaustive]
pub struct ConnectionSnapshot {
	/// Transport statistics at the time of the snapshot.
	pub stats: moq_net::ConnectionStats,
	/// Protocol negotiated by the connection that supplied these statistics.
	pub version: Version,
}

/// A cloneable read handle for the live connection stats of a [`Reconnect`] loop.
///
/// Obtained via [`Reconnect::stats`]. [`stats`](Self::stats) returns `None` while the loop is
/// between connections (reconnecting), and `Some` snapshot while a session is established.
#[derive(Clone)]
pub struct ConnectionStatsReader {
	state: kio::Consumer<State>,
	last_presence: moq_net::stats::Presence,
}

impl ConnectionStatsReader {
	/// Cumulative connects and disconnects of this reconnect loop, the same shape as a relay's
	/// sessions track: `sessions - sessions_closed` is 1 while connected, and a rate is a delta over
	/// any window.
	pub fn presence(&self) -> moq_net::stats::Presence {
		self.state.read().presence
	}

	/// Poll until either presence counter moves past what this handle last reported.
	pub fn poll_presence(&mut self, waiter: &kio::Waiter) -> Poll<crate::Result<moq_net::stats::Presence>> {
		let last = self.last_presence;
		let presence = match ready!(self.state.poll(waiter, |state| match state.presence {
			presence if presence != last => Poll::Ready(presence),
			_ => Poll::Pending,
		})) {
			Ok(presence) => presence,
			Err(state) => return Poll::Ready(Err(terminal(&state))),
		};

		self.last_presence = presence;
		Poll::Ready(Ok(presence))
	}

	/// Wait until either presence counter moves past what this handle last reported.
	///
	/// Unlike [`Reconnect::status`], a connect and disconnect that both land before the caller polls
	/// are not coalesced away: the counters still moved.
	pub async fn presence_changed(&mut self) -> crate::Result<moq_net::stats::Presence> {
		kio::wait(|waiter| self.poll_presence(waiter)).await
	}

	/// Snapshot the current connection's stats, or `None` if not currently connected.
	pub fn stats(&self) -> Option<moq_net::ConnectionStats> {
		self.state.read().session.as_ref().map(moq_net::Session::stats)
	}

	/// Snapshot statistics and protocol together, or `None` while disconnected.
	pub fn snapshot(&self) -> Option<ConnectionSnapshot> {
		let state = self.state.read();
		let session = state.session.as_ref()?;
		Some(ConnectionSnapshot {
			stats: session.stats(),
			version: session.version(),
		})
	}
}

/// Handle to a background reconnect loop.
///
/// Spawns a tokio task that connects, waits for session close, then reconnects with exponential
/// backoff until [`Backoff::timeout`] runs out. This loop is the only retry owner for the connection:
/// a caller that rebuilds it on failure restarts the backoff from its initial delay, which turns the
/// escalation back into a tight loop. Watch [`closed`](Self::closed) instead.
///
/// The read surface mirrors [`moq_net::Session`] so a caller can treat it like a session
/// that transparently reconnects: [`version`](Self::version), [`send_bandwidth`](Self::send_bandwidth),
/// and [`recv_bandwidth`](Self::recv_bandwidth) track the live session and reset while disconnected.
/// The extra toggle a plain session doesn't have is the connection lifecycle: [`connected`](Self::connected)
/// reads it synchronously and [`status`](Self::status) waits for the next change. [`closed`](Self::closed)
/// waits for the loop to stop. Dropping the handle aborts the background task.
pub struct Reconnect {
	abort: tokio::task::AbortHandle,
	state: kio::Consumer<State>,
	/// Persistent send-bitrate estimate, fed by the loop from each live session.
	send_bandwidth: BandwidthConsumer,
	/// Persistent recv-bitrate estimate, fed by the loop from each live session.
	recv_bandwidth: BandwidthConsumer,
	/// The last status returned by [`status`](Self::status), for change detection.
	last_reported: Option<Status>,
}

impl Reconnect {
	pub(crate) fn new(client: Client, url: Url, backoff: Backoff) -> Self {
		let producer = kio::Producer::<State>::default();
		let state = producer.consume();

		// The loop feeds these across every reconnect, so a consumer's handle survives session churn
		// (unlike a session's own bandwidth consumer, which dies with the session).
		let send_bw = BandwidthProducer::new();
		let recv_bw = BandwidthProducer::new();
		let send_bandwidth = send_bw.consume();
		let recv_bandwidth = recv_bw.consume();

		let task = tokio::spawn(async move {
			if let Err(err) = Self::run(&producer, &send_bw, &recv_bw, client, url, backoff).await {
				tracing::error!(%err, "reconnect loop exited");
				if let Ok(mut state) = producer.write() {
					state.error = Some(err);
				}
			}
			// Dropping the producers here closes the channels, signaling consumers.
		});
		Self {
			abort: task.abort_handle(),
			state,
			send_bandwidth,
			recv_bandwidth,
			last_reported: None,
		}
	}

	async fn run(
		state: &kio::Producer<State>,
		send_bw: &BandwidthProducer,
		recv_bw: &BandwidthProducer,
		client: Client,
		url: Url,
		backoff: Backoff,
	) -> crate::Result<()> {
		// The escalating wait between attempts, and the instant the give-up budget expires. Both
		// restart after a session that stayed healthy, so a one-off drop reconnects promptly. A zero
		// timeout means no deadline at all: retry for as long as the process lives.
		let mut delay = backoff.initial;
		let mut deadline = deadline_from(&backoff);
		let mut last_error: Option<Error> = None;

		// The dial target usually carries an auth token in its query, so every line
		// below logs the redacted form.
		let url_log = RedactedUrl::new(&url);

		loop {
			tracing::info!(url = %url_log, "connecting");

			match client.connect(url.clone()).await {
				Ok(session) => {
					tracing::info!(url = %url_log, "connected");
					if let Ok(mut state) = state.write() {
						state.presence.sessions += 1;
						state.status = Some(Status::Connected);
						state.version = Some(session.version());
						state.session = Some(session.clone());
					}

					let connected = tokio::time::Instant::now();
					// Wait for the session to close, forwarding its bandwidth estimates into the
					// persistent producers meanwhile so consumers track the live stats across the connection.
					let closed = run_session(send_bw, recv_bw, &session).await;
					if let Ok(mut state) = state.write() {
						state.presence.sessions_closed += 1;
						state.status = Some(Status::Disconnected);
						state.version = None;
						state.session = None;
					}
					// The estimates belonged to the now-closed session; reset until the next connect.
					let _ = send_bw.set(None);
					let _ = recv_bw.set(None);

					if connected.elapsed() >= backoff.initial {
						// Stayed up past the initial backoff: a healthy session. Reset the backoff
						// window so a one-off drop reconnects promptly.
						tracing::warn!(url = %url_log, "session closed, reconnecting");
						delay = backoff.initial;
						deadline = deadline_from(&backoff);
						last_error = None;
					} else {
						// Connected then dropped almost immediately (e.g. the server accepts then
						// resets). Treat it as a failed connection: keep the close reason so the
						// give-up timeout reports a real cause, and fall through to the shared backoff
						// sleep below so repeated flaps escalate instead of spinning the CPU.
						if let Err(err) = closed {
							let err = Error::from(err);
							tracing::warn!(url = %url_log, %err, "session severed immediately, retrying");
							last_error = Some(err);
						} else {
							tracing::warn!(url = %url_log, "session severed immediately, retrying");
						}
					}
				}
				Err(err) => {
					// The two answers a server can give that redialing cannot change: it rejected our
					// credentials, or it answered the CONNECT with a status that isn't an invitation
					// to come back. Everything else falls through to the backoff, whose budget is
					// what stops the loop.
					if err.is_auth() {
						return Err(err);
					}
					if let Some(status) = err.status()
						&& !crate::error::status_retryable(status)
					{
						return Err(err);
					}
					last_error = Some(err);
				}
			}

			let now = tokio::time::Instant::now();
			if deadline.is_some_and(|deadline| now >= deadline) {
				let timeout = backoff.timeout;
				let msg = match last_error {
					Some(err) => format!("reconnect timed out after {timeout:?}: {err}"),
					None => format!("reconnect timed out after {timeout:?}"),
				};
				return Err(Error::Reconnect(msg));
			}

			// Jittered so a fleet knocked offline together doesn't reconnect on the same tick, and
			// never past the deadline the budget promised.
			let mut wait = delay.mul_f64(0.5 + rand::rng().random::<f64>() / 2.0);
			if let Some(deadline) = deadline {
				wait = wait.min(deadline - now);
			}
			delay = backoff.next_delay(delay);

			tracing::warn!(url = %url_log, ?wait, "reconnecting after backoff");
			tokio::time::sleep(wait).await;
		}
	}

	/// Poll for the next connection status change since this handle last reported one.
	///
	/// `Ready(Ok(status))` on a change, `Ready(Err)` once the loop has stopped (the give-up error,
	/// or a generic one when the handle is dropped), `Pending` otherwise.
	pub fn poll_status(&mut self, waiter: &kio::Waiter) -> Poll<crate::Result<Status>> {
		let last = self.last_reported;
		let status = match ready!(self.state.poll(waiter, |state| match state.status {
			Some(status) if Some(status) != last => Poll::Ready(status),
			_ => Poll::Pending,
		})) {
			Ok(status) => status,
			Err(state) => return Poll::Ready(Err(terminal(&state))),
		};

		self.last_reported = Some(status);
		Poll::Ready(Ok(status))
	}

	/// Wait until the connection status changes from what this handle last reported.
	///
	/// Returns the current [`Status`]. The loop alternates `Connected`/`Disconnected`, so successive
	/// calls alternate too; but a status that flips and flips back before the caller polls is
	/// reported once. This tracks the *current* state, not every edge.
	pub async fn status(&mut self) -> crate::Result<Status> {
		kio::wait(|waiter| self.poll_status(waiter)).await
	}

	/// Whether a session is currently connected.
	///
	/// The synchronous read behind [`status`](Self::status), for callers that just want the current
	/// state rather than the next change.
	pub fn connected(&self) -> bool {
		self.state.read().status == Some(Status::Connected)
	}

	/// The negotiated MoQ version of the live session, or `None` while disconnected.
	///
	/// The [`moq_net::Session::version`] counterpart; `Option` because a reconnecting handle can be
	/// between sessions.
	pub fn version(&self) -> Option<Version> {
		self.state.read().version
	}

	/// A consumer for the live session's estimated send bitrate, mirroring
	/// [`moq_net::Session::send_bandwidth`].
	///
	/// Unlike the session's, this handle is persistent: the reconnect loop forwards each session's
	/// estimate into it, so it survives reconnects. Its value is `None` while disconnected or when the
	/// backend has no estimate.
	pub fn send_bandwidth(&self) -> BandwidthConsumer {
		self.send_bandwidth.clone()
	}

	/// A consumer for the live session's estimated receive bitrate, mirroring
	/// [`moq_net::Session::recv_bandwidth`]. Persistent across reconnects like
	/// [`send_bandwidth`](Self::send_bandwidth); `None` while disconnected or unavailable.
	pub fn recv_bandwidth(&self) -> BandwidthConsumer {
		self.recv_bandwidth.clone()
	}

	/// Poll whether the reconnect loop has stopped.
	///
	/// `Ready(Err)` if it permanently gave up (a failure no retry can clear, or the backoff timeout
	/// expiring), `Ready(Ok(()))` if stopped by dropping the handle, `Pending` while it's still
	/// running.
	pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll<crate::Result<()>> {
		ready!(self.state.poll_closed(waiter));
		Poll::Ready(match &self.state.read().error {
			Some(err) => Err(err.clone()),
			None => Ok(()),
		})
	}

	/// Wait until the reconnect loop stops.
	pub async fn closed(&self) -> crate::Result<()> {
		kio::wait(|waiter| self.poll_closed(waiter)).await
	}

	/// A cloneable handle for reading the current connection's stats.
	///
	/// The handle keeps working across reconnects, reporting `None` between connections.
	pub fn stats(&self) -> ConnectionStatsReader {
		ConnectionStatsReader {
			state: self.state.clone(),
			last_presence: moq_net::stats::Presence::default(),
		}
	}
}

/// Wait for `session` to close, forwarding its send/recv bandwidth estimates into the persistent
/// producers meanwhile so [`Reconnect`] consumers track the live estimates across the connection.
/// Returns the session's close result (the loop uses it to distinguish a healthy drop from an
/// immediate sever).
///
/// One `poll_*` step drives it all: [`poll_forward`] mirrors each kio bandwidth estimate, and the
/// transport's close future (the one non-kio source) is polled through the waiter's own waker.
async fn run_session(
	send_bw: &BandwidthProducer,
	recv_bw: &BandwidthProducer,
	session: &moq_net::Session,
) -> Result<(), moq_net::Error> {
	let mut send = session.send_bandwidth();
	let mut recv = session.recv_bandwidth();
	let closed = session.closed();
	tokio::pin!(closed);

	let err = kio::wait(|waiter| {
		poll_forward(&mut send, send_bw, waiter);
		poll_forward(&mut recv, recv_bw, waiter);
		waiter.poll_future(closed.as_mut())
	})
	.await;

	Err(err)
}

/// Mirror `bw`'s live estimate into `out` for as long as it changes, dropping the source handle once
/// the session's producer is gone so we don't keep polling a dead arm. A `poll_*` step: on return,
/// `waiter` is registered for the next change (unless the source is gone). Seeding is implicit
/// (the first call forwards the current value if there is one).
///
/// A `None` estimate is forwarded but keeps the arm alive: the backend reporting nothing right now
/// isn't the same as the session ending, and the caller resets `out` to `None` on disconnect anyway.
fn poll_forward(bw: &mut Option<BandwidthConsumer>, out: &BandwidthProducer, waiter: &kio::Waiter) {
	loop {
		let Some(consumer) = bw.as_mut() else { return };
		let Poll::Ready(res) = consumer.poll_changed(waiter) else {
			return;
		};
		match res {
			Ok(rate) => {
				let _ = out.set(rate);
			}
			Err(_) => {
				*bw = None;
				return;
			}
		}
	}
}

impl Drop for Reconnect {
	fn drop(&mut self) {
		self.abort.abort();
	}
}

/// The terminal error read from a closed channel's final state.
fn terminal(state: &State) -> Error {
	match &state.error {
		Some(err) => err.clone(),
		None => Error::Reconnect("reconnect stopped".to_string()),
	}
}

#[cfg(test)]
mod tests {
	#[tokio::test]
	async fn snapshot_uses_one_live_session() {
		let mut config = crate::ServerConfig {
			bind: Some("[::]:0".into()),
			..Default::default()
		};
		config.tls.generate = vec!["localhost".into()];
		let mut server = config.init().unwrap();
		let url = format!("moqt://localhost:{}", server.local_addr().unwrap().port())
			.parse()
			.unwrap();
		let mut config = crate::ClientConfig::default();
		config.tls.disable_verify = Some(true);
		let client = config.init().unwrap();
		let (accepted, connected) = tokio::time::timeout(Duration::from_secs(10), async {
			tokio::join!(
				async { server.accept().await.unwrap().ok().await.unwrap() },
				client.connect(url)
			)
		})
		.await
		.unwrap();
		let session = connected.unwrap();
		let version = session.version();
		let producer = kio::Producer::<State>::default();
		let reader = ConnectionStatsReader {
			state: producer.consume(),
			last_presence: Default::default(),
		};
		assert_eq!(reader.presence().active(), 0);
		assert!(reader.snapshot().is_none());
		{
			let mut state = producer.write().ok().unwrap();
			state.presence.sessions = 2;
			state.presence.sessions_closed = 1;
			state.session = Some(session);
		}
		assert_eq!(reader.presence().sessions, 2);
		assert_eq!(reader.presence().active(), 1);
		// The snapshot must read the protocol from that same session, without a second state query.
		assert_eq!(reader.snapshot().unwrap().version, version);
		producer.write().ok().unwrap().session = None;
		assert!(reader.snapshot().is_none());
		drop(accepted);
	}

	/// A connect and disconnect that both land before the consumer polls leave the status where it
	/// was, but the counters still report the flap.
	#[tokio::test]
	async fn presence_change_survives_coalesced_status() {
		let producer = kio::Producer::<State>::default();
		let mut reader = ConnectionStatsReader {
			state: producer.consume(),
			last_presence: Default::default(),
		};
		{
			let mut state = producer.write().ok().unwrap();
			state.presence.sessions = 1;
			state.status = Some(Status::Connected);
		}
		{
			let mut state = producer.write().ok().unwrap();
			state.presence.sessions_closed = 1;
			state.status = Some(Status::Disconnected);
		}
		let presence = reader.presence_changed().await.unwrap();
		assert_eq!((presence.sessions, presence.sessions_closed), (1, 1));
		assert_eq!(presence.active(), 0);
	}

	/// The retry loop is `delay = min(delay * multiplier, max)`, so a zero anywhere
	/// pins the delay at zero and turns an unreachable relay into a hot dial loop,
	/// unbounded when the give-up timeout is also zero.
	#[test]
	fn backoff_rejects_an_unpaced_retry() {
		assert!(Backoff::default().validate().is_ok());

		for bad in [
			Backoff {
				initial: Duration::ZERO,
				..Default::default()
			},
			Backoff {
				multiplier: 0,
				..Default::default()
			},
			Backoff {
				max: Duration::ZERO,
				..Default::default()
			},
		] {
			assert!(
				matches!(bad.validate(), Err(crate::Error::BackoffUnpaced)),
				"{bad:?} should be rejected"
			);
		}

		// A zero timeout is documented as retry-forever, which is only a hazard
		// unpaced, and a multiplier of 1 is a constant delay rather than no delay.
		let forever = Backoff {
			timeout: Duration::ZERO,
			multiplier: 1,
			..Default::default()
		};
		assert!(forever.validate().is_ok());
	}

	#[test]
	fn backoff_growth_saturates_before_applying_the_cap() {
		let backoff = Backoff {
			multiplier: u32::MAX,
			..Default::default()
		};
		assert_eq!(backoff.next_delay(Duration::MAX), backoff.max);
	}

	use super::*;

	#[test]
	fn test_backoff_default() {
		let backoff = Backoff::default();
		assert_eq!(backoff.initial, Duration::from_secs(1));
		assert_eq!(backoff.multiplier, 2);
		assert_eq!(backoff.max, Duration::from_secs(5));
		assert_eq!(backoff.timeout, Duration::from_secs(10));
	}

	/// The linger outlives the give-up timeout (so the reconnect error surfaces
	/// first), and an unlimited-retry timeout lingers forever.
	#[test]
	fn test_backoff_linger() {
		let backoff = Backoff::default();
		assert_eq!(backoff.linger(), backoff.timeout + Duration::from_secs(1));

		let unlimited = Backoff {
			timeout: Duration::ZERO,
			..Backoff::default()
		};
		assert_eq!(unlimited.linger(), Duration::MAX);
	}

	#[test]
	fn poll_forward_mirrors_until_the_source_closes() {
		let src = BandwidthProducer::new();
		let out = BandwidthProducer::new();
		let out_rx = out.consume();
		let waiter = kio::Waiter::noop();

		// No estimate yet: nothing forwarded, source retained.
		let mut bw = Some(src.consume());
		poll_forward(&mut bw, &out, &waiter);
		assert_eq!(out_rx.peek(), None);
		assert!(bw.is_some());

		// A value is mirrored through.
		src.set(Some(3_000)).unwrap();
		poll_forward(&mut bw, &out, &waiter);
		assert_eq!(out_rx.peek(), Some(3_000));

		// The estimate becoming unavailable is mirrored, but the arm stays: the
		// backend reporting nothing right now is not the session ending.
		src.set(None).unwrap();
		poll_forward(&mut bw, &out, &waiter);
		assert_eq!(out_rx.peek(), None);
		assert!(bw.is_some());

		// So a later value on the same live session still gets through. Dropping the
		// arm on the `None` above would have stranded the estimate at `None` for the
		// rest of the session.
		src.set(Some(9_000)).unwrap();
		poll_forward(&mut bw, &out, &waiter);
		assert_eq!(out_rx.peek(), Some(9_000));

		// Closing the source is what retires the arm, so we stop polling a dead one.
		src.abort(moq_net::Error::Cancel).unwrap();
		poll_forward(&mut bw, &out, &waiter);
		assert!(bw.is_none());
	}
}