From 2af23198aacbce5f98528b666ba4e4d66cf46514 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 11:31:33 -0700 Subject: [PATCH 1/3] quest: claim m1/quic/receive-memory-coverage Co-Authored-By: Claude Opus 5.5 From 5e206471ab2977de8a95013520cece42b3765fa1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 12:12:05 -0700 Subject: [PATCH 2/3] test(quic): prove receive credit recovery with a JavaScript sender and benchmark it A Bun publisher over @moq/web-transport fills a test-only WebTransport server's connection window, on tokio's raw driver and on an io_uring endpoint, with a frame that can never complete. The receiver stops it, a high-priority frame completes, and a later low group arrives. The interop workflow runs it, since normal CI has no Bun. benchmark_receive_windows sweeps concurrent streams and frame sizes on separate axes, comparing the driver's finite stream window with one that follows the connection window, and reports retained bytes, throughput, and frame completion latency. Nightly runs it next to benchmark_held_reads. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/interop.yml | 11 + bun.lock | 9 + package.json | 1 + rs/moq-tokio/tests/receive_credit.rs | 363 +++++++++++++++++++++++---- rs/moq-uring/tests/receive_credit.rs | 152 ++++++++++- sh/gh/nightly.sh | 1 + test/justfile | 5 + test/receive-credit/package.json | 12 + test/receive-credit/sender.ts | 78 ++++++ test/receive-credit/tsconfig.json | 5 + 10 files changed, 585 insertions(+), 52 deletions(-) create mode 100644 test/receive-credit/package.json create mode 100644 test/receive-credit/sender.ts create mode 100644 test/receive-credit/tsconfig.json diff --git a/.github/workflows/interop.yml b/.github/workflows/interop.yml index 676469e81a..d4946e312e 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/max_age.rs" + - "test/receive-credit/**" + - "rs/moq-tokio/tests/receive_credit.rs" + - "rs/moq-uring/tests/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" @@ -94,6 +97,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 e4e72f2270..b97f55dd49 100644 --- a/bun.lock +++ b/bun.lock @@ -434,6 +434,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": { @@ -754,6 +761,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 078b408bb8..4b17431a00 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/rs/moq-tokio/tests/receive_credit.rs b/rs/moq-tokio/tests/receive_credit.rs index 4bd54161e7..8864ce7c33 100644 --- a/rs/moq-tokio/tests/receive_credit.rs +++ b/rs/moq-tokio/tests/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/receive_credit.rs b/rs/moq-uring/tests/receive_credit.rs index 92246d4198..cdd044f0a6 100644 --- a/rs/moq-uring/tests/receive_credit.rs +++ b/rs/moq-uring/tests/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. @@ -222,3 +224,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 2e552ce1a7..e1ec62b5c1 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 # 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 # Each kio routine once, including the contended bench's worker threads. diff --git a/test/justfile b/test/justfile index 0313be1c2d..1ef7cc6abb 100644 --- a/test/justfile +++ b/test/justfile @@ -186,6 +186,11 @@ video: max-age: cargo nextest run --locked -p moq-tokio --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 receive_credit --run-ignored only javascript + cargo nextest run --locked -p moq-uring --test receive_credit --run-ignored only javascript + # 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 0000000000..c9b094ac6b --- /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 0000000000..362a41a855 --- /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 0000000000..9f2c971fb7 --- /dev/null +++ b/test/receive-credit/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../js/tsconfig.json", + "compilerOptions": { "types": ["bun"], "noEmit": true }, + "include": ["*.ts"] +} From e6e1be3a18472c1094490cecf069c3587255f932 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 12:12:59 -0700 Subject: [PATCH 3/3] quest: finish m1/quic/receive-memory-coverage Co-Authored-By: Claude Opus 5.5 --- quest/m1/quic/README.md | 1 - quest/m1/quic/receive-memory-coverage.md | 43 ------------------------ quest/m1/quic/receive-memory.md | 5 +-- 3 files changed, 1 insertion(+), 48 deletions(-) delete mode 100644 quest/m1/quic/receive-memory-coverage.md diff --git a/quest/m1/quic/README.md b/quest/m1/quic/README.md index db5c05223e..90542b510a 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 - [Bound message assembly](/quest/m1/quic/receive-memory.md) - complete headers and frames return credit, and receiver priority cancellation restores progress -- [Prove receive memory](/quest/m1/quic/receive-memory-coverage.md) - a JavaScript sender exhausts both native receivers and they recover; nightly measures the cost - [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - `RESET_STREAM_AT`, so a reset WebTransport stream still delivers its header - [Hierarchical stream scheduling](/quest/m1/quic/scheduler.md) - strict diff --git a/quest/m1/quic/receive-memory-coverage.md b/quest/m1/quic/receive-memory-coverage.md deleted file mode 100644 index 417990398d..0000000000 --- a/quest/m1/quic/receive-memory-coverage.md +++ /dev/null @@ -1,43 +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](/quest/m1/quic/receive-memory.md) -work, 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. - -## Required - -- [Bound message assembly](/quest/m1/quic/receive-memory.md) - the native credit, recovery, and window validation this exercises diff --git a/quest/m1/quic/receive-memory.md b/quest/m1/quic/receive-memory.md index afab88facb..6f3c0a29ce 100644 --- a/quest/m1/quic/receive-memory.md +++ b/quest/m1/quic/receive-memory.md @@ -85,9 +85,7 @@ shared allowance, a finite stream window smaller than a frame, the largest supported encoded unit, read-ahead across unit boundaries, cancellation and late reset/data races, unidentified and control-only pressure, and continued progress after selecting victims, and run all-language interop. Wire checks -into CI; no wall-clock sleeps in unit tests. The JavaScript sender and the -retained-bytes, throughput, and latency sweep moved to -[their own quest](/quest/m1/quic/receive-memory-coverage.md). +into CI; no wall-clock sleeps in unit tests. Public API: native receive-credit integration and validation of the existing window configuration; no new header-specific quota. Wire: existing QUIC @@ -98,7 +96,6 @@ separate guide quest is needed. ## Related -- [Prove receive memory](/quest/m1/quic/receive-memory-coverage.md) - a JavaScript sender against both native receivers, and the nightly benchmark sweep - [Scope track priority](/quest/m1/track-priority-scope.md) - broadcast fairness domains and subscription priority semantics - [Hierarchical stream scheduling](/quest/m1/quic/scheduler.md) - share priority semantics without requiring sender-side recovery - [QUIC bitrate caps](/quest/m2/rate-quic.md) - pace credit grants inside the independently bounded memory allowance