diff --git a/.github/workflows/interop.yml b/.github/workflows/interop.yml index f32bc874569..2ec1584cfad 100644 --- a/.github/workflows/interop.yml +++ b/.github/workflows/interop.yml @@ -21,6 +21,9 @@ on: - "test/max-age/**" - "test/portrait/**" - "rs/moq-tokio/tests/integration/max_age.rs" + - "test/receive-credit/**" + - "rs/moq-tokio/tests/integration/receive_credit.rs" + - "rs/moq-uring/tests/integration/receive_credit.rs" # The interop job also depends on the recipe that invokes it and the # workspace membership that links the JS clients to source. - "test/justfile" @@ -89,6 +92,14 @@ jobs: run: nix develop --command just test max-age shell: bash -leo pipefail {0} + - name: Receive credit recovery + if: ${{ !cancelled() && steps.setup.outcome == 'success' }} + run: nix develop --command just test receive-credit + shell: bash -leo pipefail {0} + env: + # Fail rather than skip the io_uring arm if the runner's kernel lacks it. + MOQ_STRICT: 1 + - name: Harness regressions if: ${{ !cancelled() && steps.setup.outcome == 'success' }} run: nix develop --command just test harness diff --git a/bun.lock b/bun.lock index 0129813d97f..14f8878274e 100644 --- a/bun.lock +++ b/bun.lock @@ -435,6 +435,13 @@ "vite": "^8.3.2", }, }, + "test/receive-credit": { + "name": "@moq/receive-credit-test", + "dependencies": { + "@moq/net": "workspace:*", + "@moq/web-transport": "^0.1.5", + }, + }, "test/video": { "name": "@moq/video-test", "dependencies": { @@ -755,6 +762,8 @@ "@moq/qmux": ["@moq/qmux@0.3.3", "", { "dependencies": { "@moq/web-socket-stream": "^0.1.1" } }, "sha512-Ez1GCVwpyNGB0eWvhKIXOjXrfRlMdmO0lpJ7gqInZFzsOsYTyML0WRgh24Itczp9Kev0CXsUA/K+XrGs1dRKUA=="], + "@moq/receive-credit-test": ["@moq/receive-credit-test@workspace:test/receive-credit"], + "@moq/room": ["@moq/room@workspace:js/room"], "@moq/signals": ["@moq/signals@workspace:js/signals"], diff --git a/package.json b/package.json index 078b408bb87..4b17431a00f 100644 --- a/package.json +++ b/package.json @@ -48,6 +48,7 @@ "test/interop/clients/js", "test/interop/clients/js-native", "test/drain", + "test/receive-credit", "test/video", "test/p2p", "test/wasm" diff --git a/quest/m1/quic/README.md b/quest/m1/quic/README.md index db943a6a4ac..c54965390cb 100644 --- a/quest/m1/quic/README.md +++ b/quest/m1/quic/README.md @@ -63,7 +63,6 @@ cache limits remain separate; bitrate pacing composes with this memory cap. ## Required - [Regression net](/quest/m1/quic/coverage/README.md) - benchmarks, fuzzing, characterization tests, and a seeded simulation, so later QUIC changes see every behavior or cost change -- [Prove receive memory](/quest/m1/quic/receive-memory-coverage.md) - a JavaScript sender exhausts both native receivers and they recover; nightly measures the cost - [A reset group keeps its header](/quest/m1/quic/group-reset.md) - a lite publisher resets a group with `RESET_STREAM_AT`, so its subscriber drops the tail grace diff --git a/quest/m1/quic/receive-memory-coverage.md b/quest/m1/quic/receive-memory-coverage.md deleted file mode 100644 index 81e8835d7f1..00000000000 --- a/quest/m1/quic/receive-memory-coverage.md +++ /dev/null @@ -1,39 +0,0 @@ -# [M] Prove receive memory with JavaScript senders and benchmarks - -## Goal - -A JavaScript sender drives both native receivers into receive-credit -exhaustion and they recover, and nightly measures what bounded receive memory -costs. This is coverage for the message-assembly work in #5735, with no new -transport behavior. - -## Plan - -Split out of #5735, which lands native credit holding, pressure recovery, and -the minimum receive window with Rust-only end-to-end tests over tokio and -io_uring. - -A JavaScript sender over the stock receivers cannot stall them. The minimum -window fits the largest unit, and a strict-priority sender finishes one -stream before the next, so every frame completes. The test therefore needs a -receiver whose connection window is far below the minimum. Neither stock -config allows that, so the test assembles a WebTransport (HTTP/3) server -from each raw driver: tokio's `quic::driver` with `quic::web::Server`, and -an io_uring endpoint in web mode. Keep it test-only; do not add a public way -to configure a window below `MIN_RECEIVE_WINDOW`. - -The sender is a Bun script using `@moq/web-transport`, pinned to the test -server's certificate hash with WebSocket disabled. It mirrors the Rust tests -in `rs/moq-tokio/tests/receive_credit.rs`. A low-priority track sends a frame -larger than the window. A high-priority frame then has to complete, and a -later low-priority group has to arrive after the low stream is stopped. Run -it from Rust the way `rs/moq-tokio/tests/max_age.rs` drives its Bun client, -and wire it into the interop workflow, since normal CI has no Bun. - -The benchmark sweeps concurrent streams and frame sizes, and compares a -finite stream window with one that follows the connection window. Report -retained bytes (held plus unread credit), throughput, and frame completion -latency. Sweep streams and frame sizes on separate axes, so a cost that grows -with the stream count shows up as a slope. Wire it into `sh/gh/nightly.sh` -next to `benchmark_held_reads`. Unit tests stay on mocked time; the benchmark -is the only wall-clock part. diff --git a/rs/moq-tokio/tests/integration/receive_credit.rs b/rs/moq-tokio/tests/integration/receive_credit.rs index 4bd54161e7d..8864ce7c336 100644 --- a/rs/moq-tokio/tests/integration/receive_credit.rs +++ b/rs/moq-tokio/tests/integration/receive_credit.rs @@ -1,38 +1,49 @@ //! Over native QUIC, incomplete frames hold the receiver's connection credit, and when //! they fill its window the receiver stops the least urgent stream and keeps going. //! -//! The server publishes with the stock config. The client dials over the raw driver so -//! its connection window can be far smaller than the stack allows, which keeps the -//! payloads small. Without recovery the high-priority frame never arrives: the window -//! is spent on a low-priority frame that can never complete. +//! The receiver runs on the raw driver so its connection window can be far smaller than +//! the stack allows, which keeps the payloads small. Without recovery the high-priority +//! frame never arrives: the window is spent on a low-priority frame that can never +//! complete. The Rust arm dials a stock server. The JavaScript arm serves WebTransport +//! to a Bun publisher; it needs Bun and runs in the interop workflow. #![cfg(feature = "quic")] -use std::{sync::Arc, time::Duration}; +use std::{process::Stdio, sync::Arc, time::Duration}; use bytes::Bytes; use moq_tokio::moq_net::{self, Timestamp, frame, track::Subscription}; -use moq_tokio::quic::driver; +use moq_tokio::quic::{driver, web}; +use tokio::io::AsyncWriteExt; const TIMEOUT: Duration = Duration::from_secs(10); -/// The client's connection receive window. +/// The receiver's connection receive window. const WINDOW: usize = 256 * 1024; fn timed() -> moq_net::track::Info { moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI) } +/// Raw-driver transport settings with a connection receive window of `window`. +fn transport(window: u64) -> driver::TransportConfig { + let mut transport = driver::TransportConfig::default(); + transport.receive_window(driver::VarInt::from_u64(window).unwrap()); + transport +} + /// A server publishing broadcast `room`, and a client subscribed to its origin over a -/// connection whose receive window is [`WINDOW`]. +/// connection with the given transport settings. struct Setup { broadcast: moq_net::broadcast::Producer, remote: moq_net::broadcast::Consumer, + /// The client's connection, for its credit stats. + conn: driver::Connection, _client: moq_net::Session, _endpoint: driver::Endpoint, } -async fn setup(version: &str) -> Setup { +async fn setup(version: &str, transport: driver::TransportConfig) -> Setup { let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); let version: moq_net::Version = version.parse().expect("version"); @@ -63,8 +74,6 @@ async fn setup(version: &str) -> Setup { .with_no_client_auth(); tls.alpn_protocols = vec![version.alpn().as_bytes().to_vec()]; let quic = driver::crypto::rustls::QuicClientConfig::try_from(tls).expect("quic config"); - let mut transport = driver::TransportConfig::default(); - transport.receive_window((WINDOW as u32).into()); let mut client = driver::ClientConfig::new(Arc::new(quic)); client.transport_config(Arc::new(transport)); @@ -76,10 +85,8 @@ async fn setup(version: &str) -> Setup { .expect("handshake"); let subscriber = moq_tokio::origin::spawn(); - let session = moq_tokio::transport::Session::new( - moq_tokio::quic::web::Session::raw(conn), - moq_net::transport::Binding::InBand, - ); + let session = + moq_tokio::transport::Session::new(web::Session::raw(conn.clone()), moq_net::transport::Binding::InBand); let (client, driver) = moq_net::Client::new() .with_versions(version.into()) .with_subscriber(subscriber.clone()) @@ -88,39 +95,47 @@ async fn setup(version: &str) -> Setup { .expect("client handshake"); tokio::spawn(moq_net::time::run(driver)); - let consumer = subscriber.consume(); + Setup { + broadcast, + remote: room(&subscriber).await, + conn, + _client: client, + _endpoint: endpoint, + } +} + +/// Broadcast `room`, once the peer announces it into `origin`. +async fn room(origin: &moq_net::origin::Producer) -> moq_net::broadcast::Consumer { + let consumer = origin.consume(); tokio::time::timeout(TIMEOUT, consumer.routed("room")) .await .expect("announce timeout") .expect("routed"); - let remote = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("room", None)) + tokio::time::timeout(TIMEOUT, consumer.request_broadcast("room", None)) .await .expect("resolve timeout") - .expect("broadcast resolves"); + .expect("broadcast resolves") +} - Setup { - broadcast, - remote, - _client: client, - _endpoint: endpoint, - } +/// Subscribe to track `name` of `remote` at `priority`. +async fn subscribe(remote: &moq_net::broadcast::Consumer, name: &str, priority: u8) -> moq_net::track::Subscriber { + tokio::time::timeout( + TIMEOUT, + remote + .track(name) + .unwrap() + .subscribe(Subscription::default().with_priority(priority)), + ) + .await + .expect("subscribe timeout") + .expect("subscribe") } impl Setup { /// Subscribe to a new track at `priority`. async fn subscribe(&self, name: &str, priority: u8) -> (moq_net::track::Producer, moq_net::track::Subscriber) { let track = self.broadcast.create_track(name, timed()).unwrap(); - let sub = tokio::time::timeout( - TIMEOUT, - self.remote - .track(name) - .unwrap() - .subscribe(Subscription::default().with_priority(priority)), - ) - .await - .expect("subscribe timeout") - .expect("subscribe"); - (track, sub) + (track, subscribe(&self.remote, name, priority).await) } } @@ -144,10 +159,33 @@ async fn next_frame(sub: &mut moq_net::track::Subscriber) -> Bytes { .payload } +/// The next group's first frame, still arriving. +async fn partial_frame(sub: &mut moq_net::track::Subscriber) -> moq_net::frame::Consumer { + let mut group = next_group(sub).await; + tokio::time::timeout(TIMEOUT, group.next_frame()) + .await + .expect("frame header timed out") + .expect("frame") + .expect("group ended") +} + +/// Read `frame` until `bytes` of it arrived. +async fn read_partial(frame: &mut moq_net::frame::Consumer, bytes: usize) { + let mut seen = 0; + while seen < bytes { + let chunk = tokio::time::timeout(TIMEOUT, frame.read_chunk()) + .await + .expect("partial payload timed out") + .expect("chunk") + .expect("frame ended"); + seen += chunk.len(); + } +} + /// Two partial frames that together exceed the window: the lower-priority one is /// stopped, and the higher-priority one completes on the credit it returns. async fn the_less_urgent_partial_frame_gives_way(version: &str) { - let setup = setup(version).await; + let setup = setup(version, transport(WINDOW as u64)).await; let (low, mut low_sub) = setup.subscribe("low", 1).await; let (high, mut high_sub) = setup.subscribe("high", 200).await; @@ -162,21 +200,7 @@ async fn the_less_urgent_partial_frame_gives_way(version: &str) { low_frame.write(vec![1u8; WINDOW * 3 / 4]).unwrap(); // Wait until the receiver holds all of it, so nothing else got there first. - let mut partial = next_group(&mut low_sub).await; - let mut partial = tokio::time::timeout(TIMEOUT, partial.next_frame()) - .await - .expect("frame header timed out") - .expect("frame") - .expect("group ended"); - let mut seen = 0; - while seen < WINDOW * 3 / 4 { - let chunk = tokio::time::timeout(TIMEOUT, partial.read_chunk()) - .await - .expect("partial payload timed out") - .expect("chunk") - .expect("frame ended"); - seen += chunk.len(); - } + read_partial(&mut partial_frame(&mut low_sub).await, WINDOW * 3 / 4).await; // Half the window, which only fits once the low frame lets go. let payload = Bytes::from(vec![2u8; WINDOW / 2]); @@ -204,6 +228,243 @@ async fn ietf_partial_frame_gives_way() { the_less_urgent_partial_frame_gives_way("moq-transport-16").await; } +/// A Bun publisher (`test/receive-credit/sender.ts`) dials a WebTransport server whose +/// connection window is [`WINDOW`] and sends a low-priority frame twice that size. The +/// receiver stops it, a high-priority frame completes, and a later low group arrives. +async fn javascript_partial_frame_gives_way(version: &str) { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let version: moq_net::Version = version.parse().expect("version"); + + // The sender pins this certificate, so it needs no name or CA. + let rcgen::CertifiedKey { cert, signing_key } = + rcgen::generate_simple_self_signed(["localhost".to_string()]).expect("certificate"); + let dir = tempfile::tempdir().expect("tempdir"); + let pem = dir.path().join("cert.pem"); + std::fs::write(&pem, cert.pem()).expect("write certificate"); + let key = rustls::pki_types::PrivatePkcs8KeyDer::from(signing_key.serialize_der()); + let mut tls = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![cert.der().clone()], key.into()) + .expect("tls"); + tls.alpn_protocols = vec![web::ALPN.as_bytes().to_vec()]; + let quic = driver::crypto::rustls::QuicServerConfig::try_from(tls).expect("quic config"); + let mut config = driver::ServerConfig::with_crypto(Arc::new(quic)); + config.transport_config(Arc::new(transport(WINDOW as u64))); + let endpoint = driver::Endpoint::server(config, "127.0.0.1:0".parse().unwrap()).expect("endpoint"); + let port = endpoint.local_addr().expect("addr").port(); + let mut server = web::Server::new(endpoint); + + let script = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../test/receive-credit/sender.ts"); + let mut child = tokio::process::Command::new("bun") + .arg(script) + .arg(format!("https://127.0.0.1:{port}/")) + .arg(version.alpn()) + .arg(&pem) + .arg(WINDOW.to_string()) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .kill_on_drop(true) + .spawn() + .expect("spawn bun"); + let mut commands = child.stdin.take().unwrap(); + + let request = tokio::time::timeout(TIMEOUT, server.accept()) + .await + .expect("connect timeout") + .expect("request"); + let session = request + .respond(web::proto::ConnectResponse::OK.with_protocol(version.alpn())) + .await + .expect("respond"); + let origin = moq_tokio::origin::spawn(); + let (session, driver) = moq_net::Server::new() + .with_versions(version.into()) + .with_subscriber(origin.clone()) + .accept( + std::time::Instant::now(), + moq_tokio::transport::Session::new(session, moq_net::transport::Binding::Uri), + ) + .await + .expect("server handshake"); + tokio::spawn(moq_net::time::run(driver)); + let remote = room(&origin).await; + let mut low = subscribe(&remote, "low", 1).await; + let mut high = subscribe(&remote, "high", 200).await; + + // Wait until three quarters of the window holds a frame that can never complete. + commands.write_all(b"low\n").await.unwrap(); + let mut partial = partial_frame(&mut low).await; + read_partial(&mut partial, WINDOW * 3 / 4).await; + + // Half the window, which only fits once the low frame lets go. + commands.write_all(b"high\n").await.unwrap(); + assert_eq!(next_frame(&mut high).await, vec![2u8; WINDOW / 2], "{version}"); + loop { + let chunk = tokio::time::timeout(TIMEOUT, partial.read_chunk()) + .await + .expect("stop timed out"); + match chunk { + Ok(Some(_)) => {} + Ok(None) => panic!("{version}: a frame larger than the window completed"), + Err(_) => break, + } + } + + // The session survives, and the low track keeps delivering what fits. + commands.write_all(b"after\n").await.unwrap(); + assert_eq!(next_frame(&mut low).await, &b"after"[..], "{version}"); + + drop(commands); + let status = tokio::time::timeout(TIMEOUT, child.wait()) + .await + .expect("sender exit timed out") + .expect("sender"); + assert!(status.success(), "{version}: JS sender failed"); + drop(session); +} + +#[tokio::test] +#[ignore = "requires Bun; run by just test receive-credit in the interop workflow"] +async fn javascript_partial_frame_gives_way_tokio() { + for version in ["moq-lite-06", "moq-transport-16"] { + javascript_partial_frame_gives_way(version).await; + } +} + +/// moq-tokio's default connection receive window. +const STOCK_WINDOW: u64 = 128 << 20; +/// Bytes each benchmark run delivers. +const BENCH_BYTES: usize = 256 << 20; + +/// What one benchmark run measured. +struct Measured { + /// The most the receiver held plus received without reading, in bytes. + retained: u64, + /// Payload bytes delivered per second. + throughput: f64, + /// How long frames took from the write to the receiver completing them, sorted. + latency: Vec, +} + +/// Deliver [`BENCH_BYTES`] over `streams` tracks of one-frame groups of `frame` bytes, +/// each track writing its next group once the last arrived, so `streams` streams are +/// open at once. `follow` gives each stream the whole connection window; otherwise it +/// keeps the driver's finite default. +async fn measure(streams: usize, frame: usize, follow: bool) -> Measured { + let mut transport = transport(STOCK_WINDOW); + let max = driver::VarInt::from_u64(moq_tokio::quic::DEFAULT_MAX_STREAMS).unwrap(); + transport.max_concurrent_uni_streams(max); + transport.max_concurrent_bidi_streams(max); + if follow { + transport.stream_receive_window(driver::VarInt::from_u64(STOCK_WINDOW).unwrap()); + } + let setup = setup("moq-lite-06", transport).await; + let mut tracks = Vec::with_capacity(streams); + for index in 0..streams { + tracks.push(setup.subscribe(&index.to_string(), 0).await); + } + + let done = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let sampler = tokio::spawn({ + let conn = setup.conn.clone(); + let done = done.clone(); + async move { + let mut peak = 0; + while !done.load(std::sync::atomic::Ordering::Relaxed) { + let credit = conn.stats().credit; + peak = peak.max(credit.held + credit.unread); + tokio::time::sleep(Duration::from_millis(1)).await; + } + peak + } + }); + + let rounds = (BENCH_BYTES / frame / streams).max(1); + let payload = Bytes::from(vec![7u8; frame]); + let start = std::time::Instant::now(); + let tasks: Vec<_> = tracks + .into_iter() + .map(|(track, mut sub)| { + let payload = payload.clone(); + tokio::spawn(async move { + let mut latency = Vec::with_capacity(rounds); + for _ in 0..rounds { + let sent = std::time::Instant::now(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + group.finish().unwrap(); + assert_eq!(next_frame(&mut sub).await.len(), payload.len()); + latency.push(sent.elapsed()); + } + latency + }) + }) + .collect(); + let mut latency = Vec::with_capacity(rounds * streams); + for task in tasks { + latency.extend(task.await.unwrap()); + } + let elapsed = start.elapsed(); + done.store(true, std::sync::atomic::Ordering::Relaxed); + latency.sort(); + Measured { + retained: sampler.await.unwrap(), + throughput: (rounds * streams * frame) as f64 / elapsed.as_secs_f64(), + latency, + } +} + +/// What bounded receive memory costs: a finite stream window against one that follows the +/// connection window, swept over concurrent streams and frame sizes on separate axes so +/// a cost that grows with either shows up as a slope. Reports the peak retained bytes +/// (held plus unread credit), throughput, and frame completion latency, each the median +/// of the runs. The publisher is a stock server, so its send window bounds what is in +/// flight as it would in production. Run by nightly. +#[tokio::test(flavor = "multi_thread")] +#[ignore = "release benchmark, run by nightly"] +async fn benchmark_receive_windows() { + const RUNS: usize = 3; + const FRAME: usize = 256 * 1024; + const STREAMS: usize = 16; + let cells = [1, 16, 256] + .map(|streams| ("streams", streams, FRAME)) + .into_iter() + .chain([16 * 1024, 256 * 1024, 4 << 20].map(|frame| ("frame", STREAMS, frame))); + println!("axis,streams,frame,stream_window,retained_mib,mib_per_s,p50_ms,p99_ms"); + for (axis, streams, frame) in cells { + let mut runs: [Vec; 2] = Default::default(); + // Alternate the modes so drift in the host hits both alike. + for _ in 0..RUNS { + for (mode, follow) in [false, true].into_iter().enumerate() { + runs[mode].push(measure(streams, frame, follow).await); + } + } + for (runs, mode) in runs.iter().zip(["finite", "follow"]) { + let median = |mut values: Vec| { + values.sort_by(f64::total_cmp); + values[values.len() / 2] + }; + let percentile = |p: usize| { + median( + runs.iter() + .map(|run| run.latency[(run.latency.len() - 1) * p / 100].as_secs_f64() * 1e3) + .collect(), + ) + }; + let retained = median(runs.iter().map(|run| run.retained as f64).collect()); + let throughput = median(runs.iter().map(|run| run.throughput).collect()); + println!( + "{axis},{streams},{frame},{mode},{:.1},{:.0},{:.2},{:.2}", + retained / (1 << 20) as f64, + throughput / (1 << 20) as f64, + percentile(50), + percentile(99), + ); + } + } +} + /// Accepts any server certificate: the test dials its own self-signed server. #[derive(Debug)] struct AnyCert; diff --git a/rs/moq-uring/tests/integration/receive_credit.rs b/rs/moq-uring/tests/integration/receive_credit.rs index e0d952f71a6..8a842807181 100644 --- a/rs/moq-uring/tests/integration/receive_credit.rs +++ b/rs/moq-uring/tests/integration/receive_credit.rs @@ -4,7 +4,9 @@ //! //! The subscriber's connection window is far smaller than a moq session would allow, //! which keeps the payloads small. Without recovery the high-priority frame never -//! arrives: the window is spent on a low-priority frame that can never complete. +//! arrives: the window is spent on a low-priority frame that can never complete. The +//! JavaScript arm serves WebTransport to a Bun publisher; it needs Bun and runs in the +//! interop workflow. //! //! Unsupported local kernels skip loudly; MOQ_STRICT makes CI refuse them. @@ -221,3 +223,151 @@ fn partial_frame_gives_way() { drop(sub_origin); origins.join().expect("origin drivers"); } + +/// A Bun publisher (`test/receive-credit/sender.ts`) dials a WebTransport endpoint whose +/// connection window is [`WINDOW`] and sends a low-priority frame twice that size. The +/// subscriber stops it, a high-priority frame completes, and a later low group arrives. +/// The worker speaks moq-lite only. +#[test] +#[ignore = "requires Bun; run by just test receive-credit in the interop workflow"] +fn javascript_partial_frame_gives_way() { + use std::io::Write; + + let Some(mut worker) = worker() else { return }; + let handle = worker.handle(); + + let (sub_origin, sub_driver) = origin::Producer::new(origin::Config::default()); + let origins = std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .expect("tokio runtime"); + rt.block_on(moq_net::time::run(sub_driver)); + }); + + let certs = support::certs().expect("certificates"); + let mut server = quic::server::Config::new( + quic::Identity::open(&certs.cert, &certs.key) + .expect("identity") + .server(&["h3"], quic::server::ClientAuth::None) + .expect("tls"), + ); + server.transport.receive_window = Some(WINDOW as u64); + let sock = handle + .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) + .expect("socket"); + let endpoint = quic::Endpoint::new(sock, quic::endpoint::Config::default().with_server(server)).expect("endpoint"); + + let script = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../test/receive-credit/sender.ts"); + let mut child = std::process::Command::new("bun") + .arg(script) + .arg(format!("https://{}/", endpoint.local_addr())) + .arg(ALPN) + .arg(&certs.cert) + .arg(WINDOW.to_string()) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::inherit()) + .spawn() + .expect("spawn bun"); + let mut commands = child.stdin.take().unwrap(); + // Waiting would block the worker, so a thread waits and reports back. + let (exited, exit) = tokio::sync::oneshot::channel(); + let waiter = std::thread::spawn(move || { + let _ = exited.send(child.wait()); + }); + + let sub = sub_origin.clone(); + worker + .block_on(async move { + let conn = within(&handle, "the connection", endpoint.accept()) + .await + .expect("accept"); + let request = within(&handle, "the CONNECT", quic::web::Request::accept(conn)) + .await + .expect("handshake"); + let session = request + .respond(quic::web::Response::default().with_protocol(ALPN)) + .await + .expect("respond"); + let (session, driver) = moq_net::Server::new() + .with_subscriber(sub.clone()) + .accept_lite( + std::time::Instant::now(), + moq_uring::transport::Session::new(session, moq_net::transport::Binding::Uri), + ) + .await + .expect("accept"); + let task_handle = handle.clone(); + handle.spawn(async move { + let _ = task_handle.run(driver).await; + }); + + let consumer = sub.consume(); + within(&handle, "the route", consumer.routed("room")) + .await + .expect("routed"); + let remote = within(&handle, "the broadcast", consumer.request_broadcast("room", None)) + .await + .expect("broadcast resolves"); + let subscribe = |name: &str, priority: u8| { + let track = remote.track(name).expect("track"); + async move { + track + .subscribe(Subscription::default().with_priority(priority)) + .await + .expect("subscribe") + } + }; + let mut low = within(&handle, "the low subscription", subscribe("low", 1)).await; + let mut high = within(&handle, "the high subscription", subscribe("high", 200)).await; + + // Wait until three quarters of the window holds a frame that can never complete. + commands.write_all(b"low\n").expect("command"); + let mut partial = within(&handle, "the low group", low.recv_group()) + .await + .expect("group") + .expect("track ended"); + let mut partial = within(&handle, "the low frame header", partial.next_frame()) + .await + .expect("frame") + .expect("group ended"); + let mut seen = 0; + while seen < WINDOW * 3 / 4 { + seen += within(&handle, "the partial payload", partial.read_chunk()) + .await + .expect("chunk") + .expect("frame ended") + .len(); + } + + // Half the window, which only fits once the low frame lets go. + commands.write_all(b"high\n").expect("command"); + assert_eq!(next_frame(&handle, &mut high).await, vec![2u8; WINDOW / 2]); + loop { + match within(&handle, "the low stop", partial.read_chunk()).await { + Ok(Some(_)) => {} + Ok(None) => panic!("a frame larger than the window completed"), + Err(_) => break, + } + } + + // The session survives, and the low track keeps delivering what fits. + commands.write_all(b"after\n").expect("command"); + assert_eq!(next_frame(&handle, &mut low).await, &b"after"[..]); + + drop(commands); + let status = within(&handle, "the sender to exit", exit) + .await + .expect("waiter") + .expect("sender"); + assert!(status.success(), "JS sender failed"); + drop(session); + }) + .expect("worker"); + + waiter.join().expect("waiter"); + drop(worker); + drop(sub_origin); + origins.join().expect("origin driver"); +} diff --git a/sh/gh/nightly.sh b/sh/gh/nightly.sh index 84c4f3a7c74..a49bcc08589 100755 --- a/sh/gh/nightly.sh +++ b/sh/gh/nightly.sh @@ -26,6 +26,7 @@ case "${1:-}" in run just rs test -p moq-relay -E 'test(benchmark_stun_sources)' --run-ignored only --success-output immediate run just rs test -p moq-quic --release -E 'test(benchmark_receive_credit)' --run-ignored only --success-output immediate run just rs test -p moq-tokio --release -E 'test(benchmark_held_reads)' --run-ignored only --success-output immediate + run just rs test -p moq-tokio --release -E 'test(benchmark_receive_windows)' --run-ignored only --success-output immediate run just rs test -p moq-net --release -E 'test(benchmark_victim)' --run-ignored only --success-output immediate # Fuzz exposes moq-net's codec benches; it is not a feature of the other crates. run cargo bench --locked --no-fail-fast -p moq-net --features fuzz --bench '*' -- --test diff --git a/test/justfile b/test/justfile index 07f610765b3..1b9b5163bbb 100644 --- a/test/justfile +++ b/test/justfile @@ -182,6 +182,11 @@ video: max-age: cargo nextest run --locked -p moq-tokio --test integration -E 'test(/^max_age::/)' --run-ignored all +# A JavaScript sender exhausts each native receiver's connection credit, which recovers. +receive-credit: + cargo nextest run --locked -p moq-tokio --test integration -E 'test(/^receive_credit::javascript_/)' --run-ignored only + cargo nextest run --locked -p moq-uring --test integration -E 'test(/^receive_credit::javascript_/)' --run-ignored only + # Cross-language subscription termination over in-memory transports. bare-fin: just rs test -p moq-net --lib bare_fin_interop --run-ignored all diff --git a/test/receive-credit/package.json b/test/receive-credit/package.json new file mode 100644 index 00000000000..c9b094ac6bf --- /dev/null +++ b/test/receive-credit/package.json @@ -0,0 +1,12 @@ +{ + "name": "@moq/receive-credit-test", + "private": true, + "type": "module", + "scripts": { + "check": "tsc --noEmit" + }, + "dependencies": { + "@moq/net": "workspace:*", + "@moq/web-transport": "^0.1.5" + } +} diff --git a/test/receive-credit/sender.ts b/test/receive-credit/sender.ts new file mode 100644 index 00000000000..362a41a8559 --- /dev/null +++ b/test/receive-credit/sender.ts @@ -0,0 +1,78 @@ +/** + * A JavaScript publisher that drives a native receiver into receive-credit exhaustion. + * + * Dials the Rust test's WebTransport server through the `@moq/web-transport` polyfill, + * pinned to its certificate, and announces broadcast `room` with tracks `low` and `high`. + * Each stdin line asks for one group, so the receiver decides when each one is written: + * + * - `low`: a frame twice the window on `low`, which can never complete. + * - `high`: half a window on `high`, which only fits once the receiver stops `low`. + * - `after`: a small group on `low`, which has to arrive after that stream is stopped. + * + * bun sender.ts https://127.0.0.1:PORT/ ALPN CERT.pem WINDOW + * + * Exits 0 once stdin closes. + * + * @module + */ +import * as Moq from "@moq/net"; +import { install } from "@moq/web-transport"; + +// stdout is unused, but keep library debug logs off it all the same. +console.debug = console.error; + +// Bun has no native WebTransport; @moq/net reads the global at connect time. +install(); + +const [address, protocol, certificate, windowArg] = process.argv.slice(2); +const window = Number(windowArg); +if (!address || !protocol || !certificate || !Number.isInteger(window) || window <= 0) { + console.error("usage: sender.ts URL ALPN CERT.pem WINDOW"); + process.exit(2); +} + +const origin = new Moq.Origin.Producer(); +const broadcast = origin.createBroadcast(Moq.Path.from("room")); +const low = broadcast.createTrack("low"); +const high = broadcast.createTrack("high"); +broadcast.announce(); + +const session = await Moq.Connection.connect({ + url: new URL(address), + publish: origin.consume(), + webtransport: { protocols: [protocol], serverCertificate: await Bun.file(certificate).text() }, + // The point is the native QUIC receiver; a WebSocket fallback would skip it. + websocket: { enabled: false }, +}); + +// Written once and left open: the receiver has to give up on it. +let partial: Moq.Group.Producer | undefined; + +try { + for await (const line of console) { + const command = line.trim(); + // Bun yields a trailing empty line at EOF. + if (command === "") continue; + if (command === "low") { + partial = low.appendGroup(); + partial.writeFrame({ payload: new Uint8Array(2 * window).fill(1) }); + } else if (command === "high") { + const group = high.appendGroup(); + group.writeFrame({ payload: new Uint8Array(window / 2).fill(2) }); + group.close(); + } else if (command === "after") { + const group = low.appendGroup(); + group.writeFrame({ payload: new TextEncoder().encode("after") }); + group.close(); + } else { + throw new Error(`unknown command: ${command}`); + } + } +} finally { + // End the live tracks, or the close waits out its deadline for them. + partial?.close(); + low.close(); + high.close(); + await session.close(); + origin.close(); +} diff --git a/test/receive-credit/tsconfig.json b/test/receive-credit/tsconfig.json new file mode 100644 index 00000000000..9f2c971fb7e --- /dev/null +++ b/test/receive-credit/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../js/tsconfig.json", + "compilerOptions": { "types": ["bun"], "noEmit": true }, + "include": ["*.ts"] +}