moq-net 0.3.21

//! An idle viewer session costs its server nothing per tick: no driver wakes and
//! no transport stats reads while nothing moves. A server holds thousands of these,
//! so a fixed-interval timer per session adds up to whole cores.

mod support;

use std::{
	sync::{
		Arc,
		atomic::{AtomicUsize, Ordering},
	},
	time::Duration,
};

use moq_net::{Hop, Version, bandwidth::Rate, origin};
use support::harness::{MockConnectOptions, MockPair, connect_mock};
use support::mock::MockStats;

const VERSIONS: [&str; 2] = ["moq-lite-06", "moq-lite-07-wip"];

const IDLE: Duration = Duration::from_secs(60);

/// PROBE refreshes an unchanged report this often.
const PROBE_MAX_AGE: Duration = Duration::from_secs(10);

const STATS: MockStats = MockStats {
	estimated_send_rate: Some(1_000_000),
	rtt: Some(Duration::from_millis(40)),
};

/// A viewer session whose server publishes `origin`, with the viewer's PROBE
/// stream open (as a browser viewer keeps one) and its first report received.
async fn viewer(
	version: &str,
	origin: &origin::Producer,
	polls: Option<Arc<AtomicUsize>>,
) -> (MockPair, moq_net::bandwidth::Consumer) {
	let mut options = MockConnectOptions::new(version.parse::<Version>().unwrap());
	options.server_publish = Some(origin.consume());
	options.client_subscribe = Some(origin::Producer::new(origin::Config::new(Hop::new(2).unwrap())).0);
	options.stats = STATS;
	options.server_polls = polls;
	let pair = connect_mock(options).await;

	let mut probe = pair.client.recv_bandwidth().expect("lite carries PROBE");
	assert_eq!(probe.changed().await.unwrap(), Some(Rate::from_bps(1_000_000)));
	tokio::time::sleep(Duration::from_secs(1)).await;
	(pair, probe)
}

fn origin() -> origin::Producer {
	let (origin, driver) = origin::Producer::new(origin::Config::new(Hop::new(1).unwrap()));
	tokio::spawn(support::harness::run(driver));
	origin
}

#[tokio::test(start_paused = true)]
async fn idle_viewer_does_not_wake_the_server() {
	for version in VERSIONS {
		let origin = origin();
		let polls = Arc::new(AtomicUsize::new(0));
		let (pair, _probe) = viewer(version, &origin, Some(polls.clone())).await;

		let reads = &pair.server_transport;
		let (start_polls, start_reads) = (polls.load(Ordering::Relaxed), reads.stats_reads());
		assert!(start_polls > 0, "the requested poll counter must observe the handshake");
		tokio::time::sleep(IDLE).await;
		let polls = polls.load(Ordering::Relaxed) - start_polls;
		let reads = reads.stats_reads() - start_reads;

		// Nothing moves, so the only work left is the max-age refresh: one sample,
		// and the deadline's poll plus the re-poll its own wake schedules.
		let refreshes = (IDLE.as_secs() / PROBE_MAX_AGE.as_secs()) as usize;
		assert!(reads <= refreshes, "{version}: {reads} stats reads over {IDLE:?}");
		assert!(polls <= 2 * refreshes, "{version}: {polls} driver polls over {IDLE:?}");
	}
}

/// The flip side: a session the server is doing work for samples as it goes, so a
/// moved estimate reaches the viewer with that work, not at the next max age.
#[tokio::test(start_paused = true)]
async fn busy_viewer_sees_the_estimate_move() {
	for version in VERSIONS {
		let origin = origin();
		let (pair, mut probe) = viewer(version, &origin, None).await;

		pair.server_transport.set_stats(MockStats {
			estimated_send_rate: Some(2_000_000),
			..STATS
		});
		// Server work for this session: announcing a broadcast to the viewer.
		let broadcast = origin.create_broadcast("live").unwrap();
		broadcast.announce(Default::default()).unwrap();

		let moved = tokio::time::timeout(Duration::from_millis(1), probe.changed()).await;
		assert_eq!(
			moved.expect("reported with the work").unwrap(),
			Some(Rate::from_bps(2_000_000)),
			"{version}"
		);
	}
}