From 01c8024351e8a7a4a1cb9e69593414b96a0f3437 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 22:17:32 -0700 Subject: [PATCH 01/15] chore(quest): start test stability questline Co-Authored-By: GPT-6 From 8aa60171f285431949f073514424cebe08a2657d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 15:29:42 -0700 Subject: [PATCH 02/15] test(tokio): make broadcast transport race deterministic (#4664) Co-authored-by: GPT-6 --- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/broadcast-race.md | 22 ----- rs/moq-tokio/src/client.rs | 108 +++++++++++++++++++++++ rs/moq-tokio/tests/broadcast.rs | 82 +---------------- rs/moq-tokio/tests/worker.rs | 28 ++---- 5 files changed, 116 insertions(+), 125 deletions(-) delete mode 100644 quest/m1/test-flakes-2/broadcast-race.md diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index c48c60452b..94b6950a7a 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -32,7 +32,6 @@ Public API: none. Wire: none. - [moq-cli tests on a paused clock](/quest/m1/test-flakes-2/cli-paused-clock.md) - the fetch timeout and completion tests stop racing wall-clock budgets - [Subscription cut by disconnect](/quest/m1/test-flakes-2/subscription-cut.md) - a publisher disconnect never ends a subscription clean -- [Broadcast race](/quest/m1/test-flakes-2/broadcast-race.md) - the QUIC-wins race test binds no shared port and has a deterministic winner - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed - [Shaper virtual time](/quest/m1/test-flakes-2/shaper-virtual-time.md) - `moq-shaper` tests judge seeded decisions on paused time, not on wall-clock delivery under load - [Scoped WARN capture](/quest/m1/test-flakes-2/warn-capture.md) - the drop-unfinished tests count only their own WARNs diff --git a/quest/m1/test-flakes-2/broadcast-race.md b/quest/m1/test-flakes-2/broadcast-race.md deleted file mode 100644 index 36b0ddf185..0000000000 --- a/quest/m1/test-flakes-2/broadcast-race.md +++ /dev/null @@ -1,22 +0,0 @@ -# [S] Broadcast race without a shared port - -## Goal - -moq-tokio `broadcast_race_quic_wins` binds no port it didn't reserve and -has a deterministic winner. - -## Plan - -It binds TCP `:0` and then UDP on the same number, which nothing reserves: -the collision [#4084](https://github.com/moq-dev/moq/pull/4084) removed from -its sibling after [#4055](https://github.com/moq-dev/moq/pull/4055) papered -over it with a retry. - -The test shares one port only so both transports sit behind one URL. -Separate `:0` ports fix the bind collision but not the race itself: with -`websocket.delay = 0` either arm can legitimately win under load. Make the -order deterministic the way #4084 did, holding the WebSocket arm until QUIC -has connected, rather than asserting on a real race; no retry. #4084's -follow-ups (`tests/reconnect.rs` `spawn_server`, `tests/worker.rs` -`free_udp_port`) are the same probe-and-rebind pattern; fix them here if -cheap. diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index d7c517d258..654e2f2513 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -1332,6 +1332,114 @@ mod tests { accepted.await.unwrap().expect("server handshake failed"); } + /// QUIC wins while the fallback handshake is held until QUIC connects. + /// Each transport owns its ephemeral port; this exercises the same race as + /// connect_inner without depending on which real handshake runs faster. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tracing_test::traced_test] + #[tokio::test] + async fn broadcast_race_quic_wins() { + let pub_origin = crate::origin::spawn(); + let broadcast = pub_origin.create_broadcast("test").expect("failed to create broadcast"); + broadcast + .announce(Default::default()) + .expect("failed to create broadcast"); + let track = broadcast.create_track("video", None).expect("failed to create track"); + let mut group = track.append_group().expect("failed to append group"); + group + .write_frame(crate::moq_net::Timestamp::ZERO, b"hello".as_ref()) + .expect("failed to write frame"); + group.finish().expect("failed to finish group"); + + let ws_listener = crate::websocket::Listener::bind("[::]:0".parse().unwrap()) + .await + .expect("failed to bind WebSocket listener"); + let ws_port = ws_listener.local_addr().expect("failed to get ws addr").port(); + let (connected, quic_connected) = tokio::sync::oneshot::channel(); + let websocket_handle = tokio::spawn(async move { + quic_connected.await.expect("QUIC connected"); + ws_listener.accept().await + }); + + let mut server_config = crate::listen::Config { + bind: Some("[::]:0".parse().unwrap()), + ..Default::default() + }; + server_config.tls.generate = vec!["localhost".into()]; + let server = server_config.init(Default::default()).expect("failed to init server"); + let mut server = server.listen().await.expect("failed to listen"); + let quic_port = server.local_addr().expect("failed to get QUIC addr").port(); + + let sub_origin = crate::origin::spawn(); + let sub_consumer = sub_origin.consume(); + let mut announcements = sub_consumer.announced(); + let mut client_config = crate::connect::Config::default(); + client_config.tls.insecure = Some(true); + // Dial both arms immediately; the fallback handshake is gated by QUIC. + client_config.websocket.delay = std::time::Duration::ZERO; + + let client = client_config.init(Default::default()).expect("failed to init client"); + let quic_addr: crate::connect::Addr = Url::parse(&format!("https://localhost:{quic_port}")).unwrap().into(); + let ws_addr: crate::connect::Addr = Url::parse(&format!("http://localhost:{ws_port}")).unwrap().into(); + + // Keep this aligned with the newest default Lite version, as in tests/broadcast.rs. + let expected_version: moq_net::Version = "moq-lite-06".parse().expect("invalid version"); + + let server_handle = tokio::spawn(async move { + let request = server.accept().await.expect("no incoming connection"); + assert_eq!( + request.transport(), + crate::server::Transport::Quic, + "expected the QUIC listener", + ); + let session = request.with_publisher(&pub_origin).ok().await?; + assert_eq!(session.version(), expected_version, "server negotiated stale version"); + let _broadcast = broadcast; + let _track = track; + let _ = session.closed().await; + Ok::<_, anyhow::Error>(()) + }); + + let client = client.with_subscriber(sub_origin); + let noq = client.noq.as_ref().expect("QUIC backend"); + let quic = async { + let session = noq.connect(&client.tls, quic_addr, &client.versions).await?; + connected.send(()).expect("fallback listener alive"); + Ok::<_, crate::Error>(crate::transport::Session::new(session)) + }; + let cc = client + .race_moq_connect(&client.moq, ws_addr, quic) + .await + .expect("client connect failed"); + + assert_eq!(cc.version(), expected_version, "client negotiated stale version"); + websocket_handle.abort(); + let _ = websocket_handle.await; + + let update = announcements.next().await.expect("origin closed"); + assert_eq!(update.prefix.as_str(), "test"); + assert!(update.kind.is_active(), "expected announcement"); + let broadcast = sub_consumer + .request_broadcast("test") + .await + .expect("broadcast resolves"); + let mut track = broadcast + .track("video") + .unwrap() + .subscribe(None) + .await + .expect("subscribe"); + let mut group = track.recv_group().await.expect("receive group").expect("track open"); + let frame = group.read_frame().await.expect("read frame").expect("group open"); + assert_eq!(&frame.payload[..], b"hello"); + + drop(cc); + server_handle + .await + .expect("server task panicked") + .expect("server task failed"); + } + #[cfg(all(feature = "websocket", feature = "noq"))] #[tokio::test] async fn race_transport_connect_reports_auth_when_both_refuse() { diff --git a/rs/moq-tokio/tests/broadcast.rs b/rs/moq-tokio/tests/broadcast.rs index 2ab11f6844..79460ac601 100644 --- a/rs/moq-tokio/tests/broadcast.rs +++ b/rs/moq-tokio/tests/broadcast.rs @@ -372,7 +372,7 @@ async fn lite05_fetch_roundtrip(scheme: &str) { #[tokio::test] async fn broadcast_moq_lite_05_fetch_webtransport() { // Exercises the WebTransport path; lite-05 is forced via config on both ends. - // The raw-QUIC ALPN path is covered by broadcast_race_quic_wins. + // The raw-QUIC ALPN path is covered by client::tests::broadcast_race_quic_wins. lite05_fetch_roundtrip("https").await; } @@ -2362,86 +2362,6 @@ async fn broadcast_websocket_uses_newest_version() { .expect("server task failed"); } -/// Regression guard for the QUIC vs WebSocket race. With both transports -/// reachable at the same URL, QUIC must win, since it's lower-latency and -/// has direct ALPN negotiation. A WebSocket win here means QUIC silently -/// regressed (and would also tend to drag the version down to Lite02 on -/// older relays). We bind WebSocket TCP and QUIC UDP to the same port, -/// then disable the head start so the race is genuine. -#[tracing_test::traced_test] -#[tokio::test] -async fn broadcast_race_quic_wins() { - let pub_origin = moq_tokio::origin::spawn(); - let broadcast = pub_origin.create_broadcast("test").expect("failed to create broadcast"); - broadcast - .announce(Default::default()) - .expect("failed to create broadcast"); - let track = broadcast.create_track("video", None).expect("failed to create track"); - let mut group = track.append_group().expect("failed to append group"); - group - .write_frame(moq_tokio::moq_net::Timestamp::ZERO, b"hello".as_ref()) - .expect("failed to write frame"); - group.finish().expect("failed to finish group"); - - // Bind WebSocket TCP first to pick a random port, then bind QUIC UDP to - // the same port. UDP and TCP live in separate kernel namespaces, so this - // works on every supported platform. - let ws_listener = moq_tokio::websocket::Listener::bind("[::]:0".parse().unwrap()) - .await - .expect("failed to bind WebSocket listener"); - let port = ws_listener.local_addr().expect("failed to get ws addr").port(); - - let mut server_config = moq_tokio::listen::Config::default(); - server_config.bind = Some(format!("[::]:{port}").parse().unwrap()); - server_config.tls.generate = vec!["localhost".into()]; - - let mut config = moq_tokio::server::Config::default(); - config.listen = server_config; - config.websocket = Some(ws_listener); - let server = config.init().expect("failed to init server"); - let mut server = server.listen().await.expect("failed to listen"); - - let sub_origin = moq_tokio::origin::spawn(); - let mut client_config = moq_tokio::connect::Config::default(); - client_config.tls.insecure = Some(true); - // Zero head start: QUIC has to win on its own merit, not by penalising WS. - client_config.websocket.delay = Duration::ZERO; - - let client = client_config.init(Default::default()).expect("failed to init client"); - let url: url::Url = format!("https://localhost:{port}").parse().unwrap(); - - let expected_version: moq_net::Version = NEWEST_LITE.parse().expect("invalid version"); - - let server_handle = tokio::spawn(async move { - let request = server.accept().await.expect("no incoming connection"); - assert_eq!( - request.transport(), - moq_tokio::server::Transport::Quic, - "QUIC lost the race to WebSocket with both reachable", - ); - let session = request.with_publisher(&pub_origin).ok().await?; - assert_eq!(session.version(), expected_version, "server negotiated stale version"); - let _broadcast = broadcast; - let _track = track; - let _ = session.closed().await; - Ok::<_, anyhow::Error>(()) - }); - - let client = client.with_subscriber(sub_origin); - let (_client, cc) = tokio::time::timeout(TIMEOUT, connect_once(client, url)) - .await - .expect("client connect timed out") - .expect("client connect failed"); - - assert_eq!(cc.version(), Some(expected_version), "client negotiated stale version"); - - drop(cc); - server_handle - .await - .expect("server task panicked") - .expect("server task failed"); -} - /// Regression guard: a Tokio runtime's session machine must run under the span /// active at accept time (e.g. a relay's /// per-connection span), or every log line moq-net emits from inside it diff --git a/rs/moq-tokio/tests/worker.rs b/rs/moq-tokio/tests/worker.rs index 811b060d58..cdf2348d37 100644 --- a/rs/moq-tokio/tests/worker.rs +++ b/rs/moq-tokio/tests/worker.rs @@ -11,17 +11,6 @@ use moq_tokio::worker::{self, Workers}; const WORKERS: u16 = 4; -/// A UDP port nothing is bound to. -/// -/// Only for the port-lock tests, where the first group names its port. -/// Everything else binds `:0` and reads it back. -fn free_udp_port() -> u16 { - let probe = UdpSocket::bind("127.0.0.1:0").expect("bind probe"); - let port = probe.local_addr().expect("local addr").port(); - drop(probe); - port -} - /// The socket inodes this process holds bound to `addr`, of which there must be /// at least one: every UDP socket, but only a TCP listener. /// @@ -330,10 +319,9 @@ async fn an_occupied_port_is_refused() { let dir = tempfile::tempdir().expect("tempdir"); let (cert, key) = certificate(dir.path()); - let port = free_udp_port(); - let workers = - bind_workers(listen_config(&cert, &key, port), Default::default(), config(WORKERS)).expect("bind workers"); + bind_workers(listen_config(&cert, &key, 0), Default::default(), config(WORKERS)).expect("bind workers"); + let port = workers.local_addr().port(); let err = bind_workers(listen_config(&cert, &key, port), Default::default(), config(WORKERS)) .expect_err("a second group must not join the first"); @@ -360,11 +348,10 @@ async fn the_other_wildcard_spelling_is_refused() { let dir = tempfile::tempdir().expect("tempdir"); let (cert, key) = certificate(dir.path()); - let port = free_udp_port(); - - let mut v4 = listen_config(&cert, &key, port); - v4.bind = Some(format!("0.0.0.0:{port}").parse().unwrap()); + let mut v4 = listen_config(&cert, &key, 0); + v4.bind = Some("0.0.0.0:0".parse().unwrap()); let workers = bind_workers(v4, Default::default(), config(WORKERS)).expect("bind v4 wildcard workers"); + let port = workers.local_addr().port(); let mut v6 = listen_config(&cert, &key, port); v6.bind = Some(format!("[::]:{port}").parse().unwrap()); @@ -390,10 +377,9 @@ async fn a_shared_port_is_refused_across_addresses() { let dir = tempfile::tempdir().expect("tempdir"); let (cert, key) = certificate(dir.path()); - let port = free_udp_port(); - let workers = - bind_workers(listen_config(&cert, &key, port), Default::default(), config(WORKERS)).expect("bind workers"); + bind_workers(listen_config(&cert, &key, 0), Default::default(), config(WORKERS)).expect("bind workers"); + let port = workers.local_addr().port(); // A distinct loopback address: no bind conflict with 127.0.0.1, only the // shared port. From baeaf3ec38904e0a36fc6f4f71e4fde8dd689528 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:07:30 -0700 Subject: [PATCH 03/15] test(shaper): run the unit tests on paused time over in-memory sockets (#4682) Co-authored-by: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/shaper-virtual-time.md | 18 ---- rs/moq-shaper/src/lib.rs | 49 +++++----- rs/moq-shaper/src/mem.rs | 93 +++++++++++++++++++ 4 files changed, 120 insertions(+), 41 deletions(-) delete mode 100644 quest/m1/test-flakes-2/shaper-virtual-time.md create mode 100644 rs/moq-shaper/src/mem.rs diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 94b6950a7a..80843a5165 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -33,7 +33,6 @@ Public API: none. Wire: none. - [moq-cli tests on a paused clock](/quest/m1/test-flakes-2/cli-paused-clock.md) - the fetch timeout and completion tests stop racing wall-clock budgets - [Subscription cut by disconnect](/quest/m1/test-flakes-2/subscription-cut.md) - a publisher disconnect never ends a subscription clean - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed -- [Shaper virtual time](/quest/m1/test-flakes-2/shaper-virtual-time.md) - `moq-shaper` tests judge seeded decisions on paused time, not on wall-clock delivery under load - [Scoped WARN capture](/quest/m1/test-flakes-2/warn-capture.md) - the drop-unfinished tests count only their own WARNs - [moq-mux debounce clock](/quest/m1/test-flakes-2/mux-debounce-clock.md) - the TS export debounce test advances on the paused clock - [js/publish audio clock](/quest/m1/test-flakes-2/publish-audio-clock.md) - the audio encoder delay test runs on mock time diff --git a/quest/m1/test-flakes-2/shaper-virtual-time.md b/quest/m1/test-flakes-2/shaper-virtual-time.md deleted file mode 100644 index 3fea8a4fb4..0000000000 --- a/quest/m1/test-flakes-2/shaper-virtual-time.md +++ /dev/null @@ -1,18 +0,0 @@ -# [S] moq-shaper tests on virtual time - -## Goal - -The `moq-shaper` unit tests pass under any host load: their verdicts come -from the seeded decisions, not from how fast the kernel delivers datagrams. -`reorder_and_jitter_overtake` and `the_seed_reproduces_the_losses` saw every -datagram lost during a slowed full-workspace run, because `round_trip` stops -at the first 200 ms of silence on a real socket. - -## Plan - -- Drive the tests on paused tokio time (or an injected clock) so delay, - jitter, and the read deadline advance deterministically. -- Watch out: paused time auto-advances while the runtime idles, which can - fire a timer before a real UDP datagram lands. If the real sockets fight - the paused clock, separate the scheduling logic from the socket I/O and test - it directly, keeping one socket smoke test with a generous deadline. diff --git a/rs/moq-shaper/src/lib.rs b/rs/moq-shaper/src/lib.rs index 4e9ca01e5a..66e359f033 100644 --- a/rs/moq-shaper/src/lib.rs +++ b/rs/moq-shaper/src/lib.rs @@ -30,14 +30,20 @@ use std::{ time::Duration, }; +#[cfg(test)] +mod mem; mod preset; pub use preset::Preset; use anyhow::Context; +#[cfg(test)] +use mem::UdpSocket; use rand::{RngExt, SeedableRng, rngs::Xoshiro256PlusPlus}; use serde::{Deserialize, Serialize}; -use tokio::{net::UdpSocket, sync::mpsc, task::JoinSet, time::Instant}; +#[cfg(not(test))] +use tokio::net::UdpSocket; +use tokio::{sync::mpsc, task::JoinSet, time::Instant}; /// How one direction of the path treats each datagram. /// @@ -995,7 +1001,7 @@ mod tests { got } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn forwards_both_ways_untouched() { let (shaper, client) = setup(1, Profile::default(), Profile::default()).await; let got = round_trip(&client, 50).await; @@ -1022,7 +1028,7 @@ mod tests { assert_eq!(start.elapsed(), Duration::ZERO); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_seed_reproduces_the_losses() { let lossy = Profile { loss: 0.3, @@ -1042,7 +1048,7 @@ mod tests { assert_eq!(first_got.len() as u64, 200 - first.stats().up.lost); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn reorder_and_jitter_overtake() { let shuffled = Profile { delay: Duration::from_millis(20), @@ -1062,7 +1068,7 @@ mod tests { assert!(stats.up.reordered > 0, "{stats}"); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_rate_limit_queues_then_drops() { // 100 datagrams of 4 bytes is 3200 bits: at 8 kbit/s they need 400ms, // and the queue only holds 100ms of it. @@ -1085,7 +1091,7 @@ mod tests { assert!(got.windows(2).all(|pair| pair[0] < pair[1]), "{got:?}"); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_rate_limit_wider_than_the_traffic_passes_everything() { let wide = Profile { rate: Some(Rate { @@ -1102,7 +1108,7 @@ mod tests { assert_eq!(shaper.verify().unwrap().up.overflowed, 0); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_burst_counts_the_datagram_itself() { // A burst of one datagram lets the first out at once and holds the second. let one = Profile { @@ -1184,7 +1190,7 @@ mod tests { assert_eq!(unapplied(&phases, &stats), ["loss"]); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn an_invalid_profile_is_refused() { let refused = |bad: Profile, why: &'static str| async move { let err = Shaper::bind(Config { @@ -1268,7 +1274,7 @@ mod tests { steps: Vec::new(), }; - #[tokio::test] + #[tokio::test(start_paused = true)] async fn only_the_gaussian_model_keeps_the_order() { let jittery = Profile { delay: Duration::from_millis(20), @@ -1387,7 +1393,7 @@ mod tests { assert!(first_counters.lost > 0 && first_counters.reordered > 0); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_gaussian_sigma_may_exceed_the_delay() { // The uniform model refuses this; a gaussian clamps its draw at zero. let wide = Profile { @@ -1457,7 +1463,7 @@ mod tests { got } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_shared_path_keeps_the_order_across_clients() { let jittery = Profile { delay: Duration::from_millis(20), @@ -1530,7 +1536,7 @@ mod tests { } } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_batch_releases_on_count_and_on_the_window() { let socket = Arc::new(UdpSocket::bind(LOCALHOST).await.unwrap()); let ms = |ms| Duration::from_millis(ms); @@ -1609,7 +1615,7 @@ mod tests { assert_eq!(tally.batched.load(Ordering::Relaxed), 0); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_batch_releases_datagrams_together() { let (shaper, client) = shaped(|config| Setup { up: batched(7, Duration::from_millis(160)), @@ -1644,7 +1650,7 @@ mod tests { ); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_window_releases_a_batch_that_never_fills() { let (shaper, client) = shaped(|config| Setup { up: batched(100, Duration::from_millis(80)), @@ -1659,7 +1665,7 @@ mod tests { assert_eq!(shaper.verify().unwrap().up.delayed, 3); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_shared_batch_fills_from_every_client() { // Three datagrams from each of two clients, into batches of six: only a // shared path fills one, and separate links wait out the window. @@ -1684,7 +1690,7 @@ mod tests { } } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_batch_needs_a_count_and_a_window() { let refused = |options: Options, why: &'static str| async move { let config = Config { @@ -1812,7 +1818,7 @@ mod tests { assert_eq!(lost(start + secs(60)), 0, "the step back never cleared"); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_step_gets_worse_part_way_through() { let ms = Duration::from_millis; let (shaper, client) = shaped(|config| Setup { @@ -1849,7 +1855,7 @@ mod tests { shaper.verify().unwrap(); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_step_the_run_never_reached_is_unapplied() { let (shaper, client) = shaped(|config| Setup { down: stepped(vec![Step { @@ -1866,8 +1872,7 @@ mod tests { assert!(format!("{err:#}").contains("the down step at 60s"), "{err:#}"); } - // Real sockets, so the wall clock: a paused one could jump past a step while a datagram is still in the kernel. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_step_no_datagram_saw_is_unapplied() { let ms = Duration::from_millis; let (shaper, client) = shaped(|config| Setup { @@ -1909,7 +1914,7 @@ mod tests { assert!(!err.contains("the up step at 200ms"), "{err}"); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_profile_no_datagram_saw_before_its_step_is_unapplied() { let ms = Duration::from_millis; let (shaper, client) = shaped(|config| Setup { @@ -1954,7 +1959,7 @@ mod tests { assert_eq!(owed(&mut link, start + Duration::from_secs(1), &[16]), [Duration::ZERO]); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn steps_that_would_be_skipped_or_do_nothing_are_refused() { let refused = |steps: Vec, why: &'static str| async move { let config = Config { diff --git a/rs/moq-shaper/src/mem.rs b/rs/moq-shaper/src/mem.rs new file mode 100644 index 0000000000..bf7e896436 --- /dev/null +++ b/rs/moq-shaper/src/mem.rs @@ -0,0 +1,93 @@ +//! An in-memory stand-in for [`tokio::net::UdpSocket`], so the unit tests run +//! on a paused clock. A paused clock jumps to the next timer whenever the +//! runtime idles, which would fire timeouts and steps while a real datagram +//! sat in the kernel; a datagram here is delivered inside the runtime, so the +//! clock only moves once nothing is in flight. `tests/cli.rs` covers real sockets. + +use std::{ + collections::HashMap, + io, + net::SocketAddr, + sync::{ + LazyLock, Mutex, OnceLock, + atomic::{AtomicU16, Ordering}, + }, +}; + +use tokio::sync::mpsc; + +type Datagram = (SocketAddr, Vec); + +/// Every bound socket by port, which is the whole address on this network. +static PORTS: LazyLock>>> = LazyLock::new(Default::default); +static NEXT_PORT: AtomicU16 = AtomicU16::new(1); + +pub struct UdpSocket { + addr: SocketAddr, + peer: OnceLock, + queue: tokio::sync::Mutex>, +} + +impl UdpSocket { + pub async fn bind(mut addr: SocketAddr) -> io::Result { + if addr.port() == 0 { + addr.set_port(NEXT_PORT.fetch_add(1, Ordering::Relaxed)); + } + let (send, queue) = mpsc::unbounded_channel(); + let mut ports = PORTS.lock().unwrap(); + if ports.contains_key(&addr.port()) { + return Err(io::ErrorKind::AddrInUse.into()); + } + ports.insert(addr.port(), send); + Ok(Self { + addr, + peer: OnceLock::new(), + queue: tokio::sync::Mutex::new(queue), + }) + } + + pub fn local_addr(&self) -> io::Result { + Ok(self.addr) + } + + pub async fn connect(&self, peer: SocketAddr) -> io::Result<()> { + self.peer.set(peer).map_err(|_| io::ErrorKind::AlreadyExists.into()) + } + + /// Like UDP, a datagram to a port nobody bound is lost without an error. + pub async fn send_to(&self, buf: &[u8], dest: SocketAddr) -> io::Result { + if let Some(port) = PORTS.lock().unwrap().get(&dest.port()) { + let _ = port.send((self.addr, buf.to_vec())); + } + Ok(buf.len()) + } + + pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { + // The port's sender lives until this socket drops, so the queue never ends. + let (from, datagram) = self.queue.lock().await.recv().await.expect("bound port"); + let size = datagram.len().min(buf.len()); + buf[..size].copy_from_slice(&datagram[..size]); + Ok((size, from)) + } + + pub async fn send(&self, buf: &[u8]) -> io::Result { + self.send_to(buf, *self.peer.get().expect("not connected")).await + } + + /// Only what the connected peer sends, as a connected UDP socket filters. + pub async fn recv(&self, buf: &mut [u8]) -> io::Result { + let peer = *self.peer.get().expect("not connected"); + loop { + let (size, from) = self.recv_from(buf).await?; + if from == peer { + return Ok(size); + } + } + } +} + +impl Drop for UdpSocket { + fn drop(&mut self) { + PORTS.lock().unwrap().remove(&self.addr.port()); + } +} From c63c01eba7520095f27e739d48c2645bc683190f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:51:52 -0700 Subject: [PATCH 04/15] quest: claim test-flakes-2/warn-capture Co-Authored-By: Claude Opus 5.5 From 113bb17931b0031a5bb5b275983b4f3f20107c04 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:51:56 -0700 Subject: [PATCH 05/15] quest: claim mux-debounce-clock From d965302e692ec1f4492aef27cafc504b5c72307c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:03:55 -0700 Subject: [PATCH 06/15] moq-mux: run the TS SI debounce on the tokio clock The SI capture's cut debounce read crate::Clock, which samples std::time::Instant, so debounce_opens_without_a_media_clock slept 1.2 s of real time despite start_paused. Track the last cut as a web_async Instant (tokio's on native, so the paused clock drives it; wasmtimer in the browser) and advance the paused clock in the test instead. Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/mux-debounce-clock.md | 15 --------------- rs/moq-mux/src/container/ts/export_test.rs | 8 ++++---- rs/moq-mux/src/container/ts/si.rs | 15 +++++++-------- 4 files changed, 11 insertions(+), 28 deletions(-) delete mode 100644 quest/m1/test-flakes-2/mux-debounce-clock.md diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 80843a5165..2e7936f95b 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -34,7 +34,6 @@ Public API: none. Wire: none. - [Subscription cut by disconnect](/quest/m1/test-flakes-2/subscription-cut.md) - a publisher disconnect never ends a subscription clean - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed - [Scoped WARN capture](/quest/m1/test-flakes-2/warn-capture.md) - the drop-unfinished tests count only their own WARNs -- [moq-mux debounce clock](/quest/m1/test-flakes-2/mux-debounce-clock.md) - the TS export debounce test advances on the paused clock - [js/publish audio clock](/quest/m1/test-flakes-2/publish-audio-clock.md) - the audio encoder delay test runs on mock time - [WebSocket paused TLS dial](/quest/m1/test-flakes-2/websocket-paused-tls.md) - the fixed-address WebSocket tests stop pausing the clock over a real dial diff --git a/quest/m1/test-flakes-2/mux-debounce-clock.md b/quest/m1/test-flakes-2/mux-debounce-clock.md deleted file mode 100644 index e29f84934b..0000000000 --- a/quest/m1/test-flakes-2/mux-debounce-clock.md +++ /dev/null @@ -1,15 +0,0 @@ -# [XS] moq-mux debounce test on the paused clock - -## Goal - -moq-mux `container::ts::export_test::debounce_opens_without_a_media_clock` -advances on the paused clock and sleeps no real time. - -## Plan - -It died with SIGTERM once in a combined run. It is marked -`start_paused = true` but sleeps 1.2 s of real time, because the debounce -reads `crate::Clock`, which uses `std::time::Instant`, so the paused tokio -clock never reaches it. Let the test drive `crate::Clock`'s time (a tokio -`Instant` under test, or an injected source) so the window passes on the -paused clock, and drop the real sleep. diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index a79f3dcec3..6924c99062 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -5016,8 +5016,8 @@ async fn contiguous_same_version_commit_replaces_stale_sections() { } /// The cut debounce runs on the host clock, so a revision publishes even when no -/// media ever advances a PTS (an audio-only or SI-only input). This test spends -/// real wall time on the debounce window; media timestamps stay pinned at zero +/// media ever advances a PTS (an audio-only or SI-only input). The paused tokio +/// clock steps over the debounce window; media timestamps stay pinned at zero /// throughout, which is exactly the case a media-clock debounce wedges on. #[tokio::test(start_paused = true)] async fn debounce_opens_without_a_media_clock() { @@ -5038,8 +5038,8 @@ async fn debounce_opens_without_a_media_clock() { .unwrap(); assert_eq!(track.latest(), Some(0), "a revision inside the window is held"); - // ...and publishes once the window passes in *real* time, no finish, no PTS. - std::thread::sleep(Duration::from_millis(1200)); + // ...and publishes once the window passes on the host clock, no finish, no PTS. + tokio::time::advance(Duration::from_millis(1200)).await; rig.import .decode(&BytesMut::from(&si_packet_cc(0x0011, &sdt(1, 0xbb), 3)[..])) .unwrap(); diff --git a/rs/moq-mux/src/container/ts/si.rs b/rs/moq-mux/src/container/ts/si.rs index b00f611ee4..f11f2602ac 100644 --- a/rs/moq-mux/src/container/ts/si.rs +++ b/rs/moq-mux/src/container/ts/si.rs @@ -29,6 +29,8 @@ use std::collections::{BTreeMap, HashMap}; use std::time::Duration; +use web_async::time::Instant; + use bytes::Bytes; use moq_net::Timestamp; @@ -213,8 +215,8 @@ struct Entry { /// The entry appears in the catalog's `mpegts.si` map: exactly while it is /// `populated`, so a subscriber never finds an empty track. advertised: bool, - /// Host-clock micros of the last cut ([`DEBOUNCE`]); `None` until the first. - last_cut: Option, + /// Host-clock instant of the last cut ([`DEBOUNCE`]); `None` until the first. + last_cut: Option, } impl Entry { @@ -345,7 +347,7 @@ impl Entry { broadcast: &moq_net::broadcast::Producer, (pid, table_id): (u16, u8), pts: Timestamp, - now: u128, + now: Instant, ) -> anyhow::Result<()> { let frames: Vec> = self .active @@ -389,8 +391,6 @@ pub(super) struct Capture { broadcast: moq_net::broadcast::Producer, catalog: crate::catalog::Producer, entries: BTreeMap<(u16, u8), Entry>, - /// Host clock driving the cut debounce (see [`DEBOUNCE`]). - clock: crate::Clock, /// The selected program_number, which is the service_id of the only service /// carried; `None` captures every service. program: Option, @@ -402,7 +402,6 @@ impl Capture { broadcast, catalog, entries: BTreeMap::new(), - clock: crate::Clock::new(), program: None, } } @@ -467,14 +466,14 @@ impl Capture { /// the host clock ([`DEBOUNCE`]), and a first snapshot is never delayed. `force` /// overrides the debounce, for end of stream. pub fn flush(&mut self, pts: Timestamp, force: bool) -> anyhow::Result<()> { - let now = self.clock.now().as_micros(); + let now = Instant::now(); for (key, entry) in self.entries.iter_mut() { if !entry.dirty { continue; } if !force && let Some(last) = entry.last_cut - && now.saturating_sub(last) < DEBOUNCE.as_micros() + && now.saturating_duration_since(last) < DEBOUNCE { continue; } From c8db5fb6d4887f0fe5e38e44ca9f777a52779832 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:12:27 -0700 Subject: [PATCH 07/15] moq-net: count drop WARNs through a global subscriber, per thread A scoped with_default subscriber let a parallel test reach the WARN callsite first on a thread with no subscriber, caching its interest as never for the whole process, so drop_unfinished_warns saw 0 WARNs. Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/warn-capture.md | 14 --- rs/moq-net/src/model/group.rs | 2 +- rs/moq-net/src/model/test_tracing.rs | 136 +++++++++++++++---------- rs/moq-net/src/model/track.rs | 2 +- 5 files changed, 87 insertions(+), 68 deletions(-) delete mode 100644 quest/m1/test-flakes-2/warn-capture.md diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 80843a5165..bfe1ae5b56 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -33,7 +33,6 @@ Public API: none. Wire: none. - [moq-cli tests on a paused clock](/quest/m1/test-flakes-2/cli-paused-clock.md) - the fetch timeout and completion tests stop racing wall-clock budgets - [Subscription cut by disconnect](/quest/m1/test-flakes-2/subscription-cut.md) - a publisher disconnect never ends a subscription clean - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed -- [Scoped WARN capture](/quest/m1/test-flakes-2/warn-capture.md) - the drop-unfinished tests count only their own WARNs - [moq-mux debounce clock](/quest/m1/test-flakes-2/mux-debounce-clock.md) - the TS export debounce test advances on the paused clock - [js/publish audio clock](/quest/m1/test-flakes-2/publish-audio-clock.md) - the audio encoder delay test runs on mock time - [WebSocket paused TLS dial](/quest/m1/test-flakes-2/websocket-paused-tls.md) - the fixed-address WebSocket tests stop pausing the clock over a real dial diff --git a/quest/m1/test-flakes-2/warn-capture.md b/quest/m1/test-flakes-2/warn-capture.md deleted file mode 100644 index d6f928e0c4..0000000000 --- a/quest/m1/test-flakes-2/warn-capture.md +++ /dev/null @@ -1,14 +0,0 @@ -# [XS] Scoped WARN capture - -## Goal - -moq-net `model::group::test::drop_unfinished_warns` and the `model::track` -test of the same name count only the WARNs their own code emits. - -## Plan - -Both count WARNs through a global tracing capture, so another test's WARN, -or a missed one, changes the count -([#4104](https://github.com/moq-dev/moq/pull/4104)). Capture per test, with a -scoped subscriber or a filter on the test's own span, instead of a -process-global count. diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index bb725dcba1..5dfa179541 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -2035,7 +2035,7 @@ mod test { drop(writer); drop(producer); }); - assert!(warns >= 1, "unfinished drop must emit unfinished-producer WARN"); + assert_eq!(warns, 1, "unfinished drop must emit one unfinished-producer WARN"); } #[test] diff --git a/rs/moq-net/src/model/test_tracing.rs b/rs/moq-net/src/model/test_tracing.rs index 63caa9d852..37fcdaec49 100644 --- a/rs/moq-net/src/model/test_tracing.rs +++ b/rs/moq-net/src/model/test_tracing.rs @@ -1,80 +1,114 @@ //! Tracing helpers shared by model tests. -use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; +use std::cell::RefCell; +use std::sync::Once; use tracing::field::{Field, Visit}; use tracing::span::{Attributes, Id, Record}; use tracing::{Event, Level, Metadata, Subscriber}; -/// Count WARN events whose `message` field contains `expected_message` while running `f`. +/// The capture active on this thread: the message to match and how many WARNs matched. +struct Capture { + expected: String, + hits: usize, +} + +thread_local! { + static CAPTURE: RefCell> = const { RefCell::new(None) }; +} + +/// Count WARN events emitted on this thread whose `message` field contains `expected_message` +/// while running `f`. /// -/// Uses only the existing `tracing` dependency (no tracing-subscriber). +/// The subscriber is process-global rather than scoped with `with_default`: tracing caches each +/// callsite's interest process-wide, so a parallel test reaching the WARN first on a thread +/// without a scoped subscriber disables it for every thread. Counting per thread keeps another +/// test's WARNs out. pub(crate) fn count_drop_warnings(expected_message: &str, f: impl FnOnce()) -> usize { - struct Count { - hits: Arc, - expected: String, - } + static INSTALL: Once = Once::new(); + INSTALL.call_once(|| tracing::subscriber::set_global_default(Warns).expect("no other global subscriber")); - struct Msg<'a> { - expected: &'a str, - matched: bool, - } + let capture = Capture { + expected: expected_message.to_owned(), + hits: 0, + }; + let prev = CAPTURE.replace(Some(capture)); + assert!(prev.is_none(), "count_drop_warnings does not nest"); - impl Visit for Msg<'_> { - fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { - if field.name() == "message" { - let s = format!("{value:?}"); - if s.contains(self.expected) { - self.matched = true; - } - } - } + f(); - fn record_str(&mut self, field: &Field, value: &str) { - if field.name() == "message" && value.contains(self.expected) { - self.matched = true; - } - } - } + CAPTURE.take().expect("capture still installed").hits +} - impl Subscriber for Count { - fn enabled(&self, metadata: &Metadata<'_>) -> bool { - *metadata.level() == Level::WARN - } +/// Routes every WARN to the capture on the emitting thread, if any. +struct Warns; - fn new_span(&self, _span: &Attributes<'_>) -> Id { - Id::from_u64(1) - } +impl Subscriber for Warns { + fn enabled(&self, metadata: &Metadata<'_>) -> bool { + *metadata.level() == Level::WARN + } - fn record(&self, _span: &Id, _values: &Record<'_>) {} + fn new_span(&self, _span: &Attributes<'_>) -> Id { + Id::from_u64(1) + } + + fn record(&self, _span: &Id, _values: &Record<'_>) {} - fn record_follows_from(&self, _span: &Id, _follows: &Id) {} + fn record_follows_from(&self, _span: &Id, _follows: &Id) {} - fn event(&self, event: &Event<'_>) { + fn event(&self, event: &Event<'_>) { + CAPTURE.with_borrow_mut(|capture| { + let Some(capture) = capture else { return }; let mut msg = Msg { - expected: &self.expected, + expected: &capture.expected, matched: false, }; event.record(&mut msg); if msg.matched { - self.hits.fetch_add(1, AtomicOrdering::SeqCst); + capture.hits += 1; } + }); + } + + fn enter(&self, _span: &Id) {} + + fn exit(&self, _span: &Id) {} +} + +struct Msg<'a> { + expected: &'a str, + matched: bool, +} + +impl Visit for Msg<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" && format!("{value:?}").contains(self.expected) { + self.matched = true; } + } - fn enter(&self, _span: &Id) {} + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() == "message" && value.contains(self.expected) { + self.matched = true; + } + } +} + +mod tests { + use super::count_drop_warnings; - fn exit(&self, _span: &Id) {} + fn probe() { + tracing::warn!("test_tracing probe"); } - let hits = Arc::new(AtomicUsize::new(0)); - let expected = expected_message.to_owned(); - tracing::subscriber::with_default( - Count { - hits: hits.clone(), - expected, - }, - f, - ); - hits.load(AtomicOrdering::SeqCst) + /// Another thread without a capture reaching the WARN first must neither disable it for this + /// thread nor be counted here. + #[test] + fn counts_only_this_threads_warns() { + let warns = count_drop_warnings("test_tracing probe", || { + std::thread::spawn(probe).join().unwrap(); + probe(); + }); + assert_eq!(warns, 1); + } } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 097b162a4b..3eb9bc8fa8 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -6841,7 +6841,7 @@ mod test { drop(writer); drop(producer); }); - assert!(warns >= 1, "unfinished drop must emit unfinished-producer WARN"); + assert_eq!(warns, 1, "unfinished drop must emit one unfinished-producer WARN"); } #[tokio::test] From 1265ae09f385a991e0d35dd7836f9bada5bd2a3f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:17:18 -0700 Subject: [PATCH 08/15] quest: plan moq-mux live import clock and tracing capture audit Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 2 ++ .../m1/test-flakes-2/mux-live-import-clock.md | 28 +++++++++++++++++++ quest/m1/test-flakes-2/trace-capture-audit.md | 24 ++++++++++++++++ 3 files changed, 54 insertions(+) create mode 100644 quest/m1/test-flakes-2/mux-live-import-clock.md create mode 100644 quest/m1/test-flakes-2/trace-capture-audit.md diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 80843a5165..04e35825c2 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -37,6 +37,8 @@ Public API: none. Wire: none. - [moq-mux debounce clock](/quest/m1/test-flakes-2/mux-debounce-clock.md) - the TS export debounce test advances on the paused clock - [js/publish audio clock](/quest/m1/test-flakes-2/publish-audio-clock.md) - the audio encoder delay test runs on mock time - [WebSocket paused TLS dial](/quest/m1/test-flakes-2/websocket-paused-tls.md) - the fixed-address WebSocket tests stop pausing the clock over a real dial +- [moq-mux live import clock](/quest/m1/test-flakes-2/mux-live-import-clock.md) - the live import restart tests advance a paused clock instead of sleeping +- [Tracing capture audit](/quest/m1/test-flakes-2/trace-capture-audit.md) - no test misses tracing events because a parallel test silenced the call site ## Related diff --git a/quest/m1/test-flakes-2/mux-live-import-clock.md b/quest/m1/test-flakes-2/mux-live-import-clock.md new file mode 100644 index 0000000000..c468e4258a --- /dev/null +++ b/quest/m1/test-flakes-2/mux-live-import-clock.md @@ -0,0 +1,28 @@ +# [S] moq-mux live import tests on the paused clock + +## Goal + +The moq-mux live import restart tests (`live_import_restarts_forward_after_idle` +for ts, fmp4, and flv) advance a paused clock across their idle gap and sleep +no real time. + +## Plan + +Facts (`origin/main`, 2026-10-01): the helpers (`live_import` in +`container/{ts,fmp4}/import_test.rs`, `import` in `flv/import_test.rs`) call +`std::thread::sleep(idle)` with a 300 ms idle. The gap is measured through +`clock::Anchor`, which reads `crate::Clock::now()`, and that is +`std::time::Instant::elapsed()`, so a paused tokio clock can't move it. + +Decided (2026-10-01): `crate::Clock`'s monotonic epoch becomes a +`web_async::time::Instant`. That is tokio's clock on native, so the tests start +paused and call `tokio::time::advance`, and it takes `std::time::Instant` off the +wasm path, as #4687 did for the TS SI debounce. The wall mapping stays +`SystemTime`. Rejected: injecting a `now` into the importers through +`translate_at`, which is plumbing only tests want. + +Update `test_util::late_clock` to match, and check other `crate::Clock` users +(including `Clock::at` callers) still build on wasm. + +Public API: check whether `Clock::at` is exported; if its `Instant` type +changes, report it. Wire: none. diff --git a/quest/m1/test-flakes-2/trace-capture-audit.md b/quest/m1/test-flakes-2/trace-capture-audit.md new file mode 100644 index 0000000000..b6be476abd --- /dev/null +++ b/quest/m1/test-flakes-2/trace-capture-audit.md @@ -0,0 +1,24 @@ +# [XS] Tracing captures survive the call-site cache + +## Goal + +No test in the workspace counts or asserts tracing events through a +thread-scoped subscriber (`with_default`, `set_default`) that a parallel +test can silence under a shared-process `cargo test`. + +## Plan + +#4690 found that tracing caches, per call site and for the whole process, +whether any subscriber wants its events. With only scoped subscribers, the +first thread to hit a call site decides for everyone, so a parallel test on a +thread with no subscriber turns the event off and the capture misses it. +Nextest runs each test in its own process and can't hit it, but `cargo test` +does. + +Grep the Rust crates for tests that capture tracing output with scoped +subscribers. Move each one to the helper #4690 added in +`rs/moq-net/src/model/test_tracing.rs` (one process-wide subscriber, events +routed to a per-thread counter), or a shared copy of it if other crates need +it. If none turn up, delete this quest. + +Public API: none. Wire: none. From ac25d88ab58eb1aba8e064630148bb2d313d33b6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:30:38 -0700 Subject: [PATCH 09/15] quest: address review on live import clock and tracing capture audit Co-Authored-By: Claude Opus 5.5 --- .../m1/test-flakes-2/mux-live-import-clock.md | 24 ++++++++++++++----- quest/m1/test-flakes-2/trace-capture-audit.md | 11 ++++++--- 2 files changed, 26 insertions(+), 9 deletions(-) diff --git a/quest/m1/test-flakes-2/mux-live-import-clock.md b/quest/m1/test-flakes-2/mux-live-import-clock.md index c468e4258a..5d5375a721 100644 --- a/quest/m1/test-flakes-2/mux-live-import-clock.md +++ b/quest/m1/test-flakes-2/mux-live-import-clock.md @@ -17,12 +17,24 @@ Facts (`origin/main`, 2026-10-01): the helpers (`live_import` in Decided (2026-10-01): `crate::Clock`'s monotonic epoch becomes a `web_async::time::Instant`. That is tokio's clock on native, so the tests start paused and call `tokio::time::advance`, and it takes `std::time::Instant` off the -wasm path, as #4687 did for the TS SI debounce. The wall mapping stays -`SystemTime`. Rejected: injecting a `now` into the importers through +wasm path. #4687 used the same `Instant` type, but only for the private SI +debounce after dropping its `crate::Clock`; `Clock` itself still reads +`std::time::Instant`. The wall mapping stays `SystemTime`. Rejected: injecting a `now` into the importers through `translate_at`, which is plumbing only tests want. -Update `test_util::late_clock` to match, and check other `crate::Clock` users -(including `Clock::at` callers) still build on wasm. +`std::time::Instant` is public at the boundary: `Clock::at(epoch, wall)` is +re-exported from `moq-mux`, and the json and binary timed writes +(`json.rs` `update`/`append`, and their `binary.rs` twins) take +`Timed<_, Instant>` through the crate-private `Clock::stamp`/`capture`. +`moq_net::Timed`'s docs say moq-mux uses `std::time::Instant`. Callers pass +`std::time::Instant` from `test_util::late_clock`, moq-cli publish, moq-hls +export, moq-audio and moq-video capture, and the catalog producer tests. -Public API: check whether `Clock::at` is exported; if its `Instant` type -changes, report it. Wire: none. +Open (maintainer, ask before starting): keep the public inputs on +`std::time::Instant` and convert at the boundary (native only; additive, stays +on `main`), or switch them to `web_async::time::Instant` as a typed break on +`dev` and update every caller and doc. Recommended: convert at the boundary, +so only the private epoch moves. + +Public API: none if converted at the boundary; otherwise `Clock::at` and the +timed writes change type. Wire: none. diff --git a/quest/m1/test-flakes-2/trace-capture-audit.md b/quest/m1/test-flakes-2/trace-capture-audit.md index b6be476abd..7dca6d7ea8 100644 --- a/quest/m1/test-flakes-2/trace-capture-audit.md +++ b/quest/m1/test-flakes-2/trace-capture-audit.md @@ -15,10 +15,15 @@ thread with no subscriber turns the event off and the capture misses it. Nextest runs each test in its own process and can't hit it, but `cargo test` does. +Start after #4690 lands, with its open review finding on the subscriber +installation window resolved, so the race isn't copied into other crates. + Grep the Rust crates for tests that capture tracing output with scoped -subscribers. Move each one to the helper #4690 added in +subscribers. Move each one onto #4690's pattern in `rs/moq-net/src/model/test_tracing.rs` (one process-wide subscriber, events -routed to a per-thread counter), or a shared copy of it if other crates need -it. If none turn up, delete this quest. +routed to a per-thread capture). Its helper only counts WARN events by +message, so a capture that asserts something else needs the helper +generalized, or a shared copy if other crates need it. If none turn up, delete +this quest. Public API: none. Wire: none. From 7a934933f201df8f8a4ae7371c903c1a4943898c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:45:05 -0700 Subject: [PATCH 10/15] moq-net: rebuild callsite interest at each WARN capture Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/test_tracing.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/rs/moq-net/src/model/test_tracing.rs b/rs/moq-net/src/model/test_tracing.rs index 37fcdaec49..c1d315f103 100644 --- a/rs/moq-net/src/model/test_tracing.rs +++ b/rs/moq-net/src/model/test_tracing.rs @@ -27,6 +27,10 @@ thread_local! { pub(crate) fn count_drop_warnings(expected_message: &str, f: impl FnOnce()) -> usize { static INSTALL: Once = Once::new(); INSTALL.call_once(|| tracing::subscriber::set_global_default(Warns).expect("no other global subscriber")); + // A callsite first reached on another thread while the install above was still publishing + // sees no subscriber and is cached as disabled. Every capture starts after the install has + // completed, so rebuilding here re-enables it against the now-global subscriber. + tracing::callsite::rebuild_interest_cache(); let capture = Capture { expected: expected_message.to_owned(), From 3873e5ddeb6cac2a4e9480de81d32cb384371a87 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 20:06:57 -0700 Subject: [PATCH 11/15] quest: settle the moq-mux Clock boundary Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/mux-live-import-clock.md | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/quest/m1/test-flakes-2/mux-live-import-clock.md b/quest/m1/test-flakes-2/mux-live-import-clock.md index 5d5375a721..fffbc557b8 100644 --- a/quest/m1/test-flakes-2/mux-live-import-clock.md +++ b/quest/m1/test-flakes-2/mux-live-import-clock.md @@ -30,11 +30,9 @@ re-exported from `moq-mux`, and the json and binary timed writes `std::time::Instant` from `test_util::late_clock`, moq-cli publish, moq-hls export, moq-audio and moq-video capture, and the catalog producer tests. -Open (maintainer, ask before starting): keep the public inputs on -`std::time::Instant` and convert at the boundary (native only; additive, stays -on `main`), or switch them to `web_async::time::Instant` as a typed break on -`dev` and update every caller and doc. Recommended: convert at the boundary, -so only the private epoch moves. - -Public API: none if converted at the boundary; otherwise `Clock::at` and the -timed writes change type. Wire: none. +Decided (maintainer, 2026-10-01): convert at the boundary. Public inputs stay +`std::time::Instant` and convert to the private epoch internally (native only), +so this is additive and stays on `main`. Rejected: switching the public types to +`web_async::time::Instant` as a break on `dev`. + +Public API: none. Wire: none. From 126c6a8eb8b94feabf8741578e9106be41999f42 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 2 Oct 2026 07:32:21 -0700 Subject: [PATCH 12/15] chore: claim media late-join quest From 70321a0d143bae2d64f4c83fbffc4d9125e0bab3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 2 Oct 2026 07:46:35 -0700 Subject: [PATCH 13/15] test(media): measure late join against the published GOP Co-Authored-By: GPT-6 --- test/interop/README.md | 5 ++- test/interop/clients/js/harness.ts | 13 ++++--- test/interop/clients/js/media.ts | 20 +++++------ test/interop/clients/js/src/contract.test.ts | 37 +++++++++++++++++++- test/interop/clients/js/src/contract.ts | 33 +++++++++++++++++ test/interop/clients/js/src/fixture.ts | 18 +++++++--- test/interop/clients/js/src/setup.ts | 4 +++ test/justfile | 2 +- 8 files changed, 108 insertions(+), 24 deletions(-) diff --git a/test/interop/README.md b/test/interop/README.md index 8d862ba42e..35f8a1efa9 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -126,7 +126,10 @@ Each run covers, against a real local relay: The fake device is not physical hardware, and the headless permission decision is not a person clicking a browser prompt. - **pause and resume**, **unsubscribe and rejoin**, **detach and reattach**, - **publisher stop and same-path republish**, and **late join**. + **publisher stop and same-path republish**, and **late join**. The late join + must present the newest published GOP or a newer one. Its lower bound is the + encoded keyframe timestamp sampled before the existing viewer closes, since + painting the canvas does not mean capture and encoding have finished. - **resources return to baseline** - the page wraps `WebTransport`, `WebSocket`, `AudioContext`, and `Worker` to count live instances, so a detach that leaks a session is visible rather than merely invisible. diff --git a/test/interop/clients/js/harness.ts b/test/interop/clients/js/harness.ts index bd59af1602..9b79020537 100644 --- a/test/interop/clients/js/harness.ts +++ b/test/interop/clients/js/harness.ts @@ -256,15 +256,18 @@ export async function readFixtureState(page: Page): Promise { } /** Invoke one of the page's {@link InteropControl} commands. */ -export async function command(page: Page, name: keyof InteropControl): Promise { - await page.evaluate( +export async function command( + page: Page, + name: K, +): Promise>> { + return (await page.evaluate( ([key, fn]) => { - const control = (window as unknown as Record void> | undefined>)[key]; + const control = (window as unknown as Record unknown> | undefined>)[key]; if (!control?.[fn]) throw new Error(`the page exposes no ${fn} command`); - control[fn](); + return control[fn](); }, [CONTROL, name] as const, - ); + )) as Awaited>; } /** Read the page's live resource counts, which outlive the player element. */ diff --git a/test/interop/clients/js/media.ts b/test/interop/clients/js/media.ts index f4e9577d3a..d73259a50e 100644 --- a/test/interop/clients/js/media.ts +++ b/test/interop/clients/js/media.ts @@ -47,7 +47,7 @@ import { import { type CaptureState, FAULTS, - KEYFRAME_INTERVAL_MS, + lateJoinStartsLive, leakedPlayerStarted, SAMPLE_MS, SAMPLE_RATE, @@ -115,9 +115,6 @@ const MIN_RATE = 0.5; */ const MAX_SKEW_STEPS = 1; -/** The first decodable frame may start at the current GOP's keyframe, but never in older history. */ -const MAX_LATE_JOIN_LAG_FRAMES = Math.ceil((Pattern.FPS * KEYFRAME_INTERVAL_MS) / 1000); - const percentile = (values: number[], p: number) => { if (values.length === 0) return Number.NaN; const sorted = [...values].sort((a, b) => a - b); @@ -660,8 +657,8 @@ try { // at the live edge rather than replay what it missed. if (wants("late-join")) { console.error("=== late join ==="); + const live = await command(publisher, "liveGop"); await player.close(); - const live = await readFixtureState(publisher); [player, playerErrors] = await subscriber(broadcast, "latecomer"); await gesture(player); const joined = await waitForState(player, playerErrors, { @@ -670,17 +667,16 @@ try { description: "the latecomer to present the fixture", predicate: (state) => state.frameId !== undefined && state.audioContext === "running", }); - // The fixture sample names the frame painted immediately before the page opens. The first - // decodable frame can be the keyframe at the start of the current GOP, so require it to be - // within that GOP rather than requiring an impossible zero-frame capture/encode delay. - const lag = live.frameId - (joined.frameId ?? 0); + // requestFrame() is asynchronous, and capture timestamps decide keyframes. The painted + // counter can therefore be over 15 ticks ahead of a still-current encoded GOP under load. + // Compare against its actual published keyframe, sampled while the old viewer held demand. check( - lag <= MAX_LATE_JOIN_LAG_FRAMES, + lateJoinStartsLive(live, joined.videoTimestamp), "late join starts live", () => - `joined at frame ${joined.frameId}, ${lag} frames behind the ${live.frameId} already published when it opened (one GOP is ${MAX_LATE_JOIN_LAG_FRAMES})`, + `joined at frame ${joined.frameId}, timestamp ${joined.videoTimestamp}ms before the current GOP's ${live.timestamp}ms keyframe`, ); - console.error(` joined at frame ${joined.frameId}, live edge was ${live.frameId}`); + console.error(` joined at frame ${joined.frameId}, timestamp ${joined.videoTimestamp}ms, current GOP began at ${live.timestamp}ms`); assertMedia(await collect(player, playerErrors, WINDOW_MS), "late join"); } diff --git a/test/interop/clients/js/src/contract.test.ts b/test/interop/clients/js/src/contract.test.ts index b36189ed58..755ffa9ce3 100644 --- a/test/interop/clients/js/src/contract.test.ts +++ b/test/interop/clients/js/src/contract.test.ts @@ -1,5 +1,6 @@ import { expect, test } from "bun:test"; -import { leakedPlayerStarted, type Resources } from "./contract"; +import * as Moq from "@moq/net"; +import { lateJoinStartsLive, leakedPlayerStarted, readLiveGop, type Resources } from "./contract"; const playing: Resources = { transports: 1, sockets: 0, audioContexts: 1, workers: 0 }; @@ -17,3 +18,37 @@ test("a leaked player is visible when it reuses a pooled websocket fallback", () test("unchanged counts are not a leak start", () => { expect(leakedPlayerStarted(playing, playing)).toBe(false); }); + + +test("late join uses the published GOP when painting is ahead of capture", async () => { + const broadcast = new Moq.Broadcast.Producer(); + const track = broadcast.createTrack("video"); + try { + const old = track.appendGroup(); + old.writeFrame({ payload: new Uint8Array([1]), timestamp: Moq.Time.Timestamp.fromMillis(3000) }); + old.close(); + const current = track.appendGroup(); + current.writeFrame({ payload: new Uint8Array([2]), timestamp: Moq.Time.Timestamp.fromMillis(3700) }); + current.writeFrame({ payload: new Uint8Array([3]), timestamp: Moq.Time.Timestamp.fromMillis(4199) }); + + // Painted counter 127 can precede capture while GOP 111 remains current. The old + // 127 - 111 <= 15 check rejects a valid join; an earlier published GOP must fail. + const viewer = track.subscribe(); + const gop = await readLiveGop(broadcast.track("video")); + expect(gop.timestamp).toBe(3700); + expect(lateJoinStartsLive(gop, 3700)).toBe(true); + expect(lateJoinStartsLive(gop, 4199)).toBe(true); + expect(lateJoinStartsLive(gop, 3000)).toBe(false); + expect(lateJoinStartsLive(gop, undefined)).toBe(false); + expect(track.used.peek()).toBe(true); + const viewed = await viewer.recvGroup(); + expect((await viewed?.readFrame())?.payload).toEqual(new Uint8Array([2])); + expect((await viewed?.readFrame())?.payload).toEqual(new Uint8Array([3])); + viewed?.close(); + viewer.close(); + await track.unused(); + expect(track.used.peek()).toBe(false); + } finally { + broadcast.close(); + } +}); diff --git a/test/interop/clients/js/src/contract.ts b/test/interop/clients/js/src/contract.ts index 00c8803e94..431c8e44e7 100644 --- a/test/interop/clients/js/src/contract.ts +++ b/test/interop/clients/js/src/contract.ts @@ -8,6 +8,8 @@ * @module */ +import type * as Moq from "@moq/net"; + // ── the deterministic publisher ───────────────────────────────────────────── /** A deliberate defect, used to prove an assertion can fail. */ @@ -43,6 +45,35 @@ export type FixtureState = { encodedFrames: number; }; +/** The newest published video's GOP, sampled independently of canvas capture and encoding. */ +export type LiveGop = { + /** The timestamp of its keyframe, in milliseconds on the publisher's media clock. */ + timestamp: number; +}; + +/** Read the newest published GOP without retaining a subscription or changing its frames. */ +export async function readLiveGop(track: Moq.Track.Consumer): Promise { + const subscriber = track.subscribe(); + try { + const group = await subscriber.recvGroup(); + if (!group) throw new Error("the fixture video has no published GOP"); + try { + const frame = await group.readFrame(); + if (!frame) throw new Error("the fixture GOP has no keyframe"); + return { timestamp: frame.timestamp.asMillis() }; + } finally { + group.close(); + } + } finally { + subscriber.close(); + } +} + +/** True when the presented frame belongs to the sampled GOP or a newer one. */ +export function lateJoinStartsLive(gop: LiveGop, timestamp: number | undefined): boolean { + return timestamp !== undefined && timestamp >= gop.timestamp; +} + /** The camera publisher state mirrored onto its element for Playwright. */ export type CaptureState = { videoError?: string; @@ -161,6 +192,8 @@ export type Sample = { * owns them rather than the driver reaching in. */ export type InteropControl = { + /** Read the current video GOP while an existing viewer is still pulling media. */ + liveGop(): Promise; /** Stop the fixture publisher, releasing its session. */ stop(): void; /** Start (or restart) the fixture publisher on the same broadcast path. */ diff --git a/test/interop/clients/js/src/fixture.ts b/test/interop/clients/js/src/fixture.ts index 17f6faef55..a3d3ff1a78 100644 --- a/test/interop/clients/js/src/fixture.ts +++ b/test/interop/clients/js/src/fixture.ts @@ -17,7 +17,7 @@ import { Time } from "@moq/net"; import * as Publish from "@moq/publish"; import { Effect, Signal } from "@moq/signals"; import type { Fault, FixtureState } from "./contract"; -import { KEYFRAME_INTERVAL_MS, OFFSET_STEPS, SAMPLE_RATE } from "./contract"; +import { KEYFRAME_INTERVAL_MS, OFFSET_STEPS, readLiveGop, SAMPLE_RATE } from "./contract"; import * as Pattern from "./pattern"; /** Cap the encoder rather than letting it track a bandwidth estimate, so runs are comparable. */ @@ -47,6 +47,7 @@ export class Fixture { readonly host: HTMLElement; readonly #signals = new Effect(); + readonly #broadcast: Publish.Broadcast; readonly #audio: AudioContext; readonly #frameId = new Signal(-1); readonly #audioState = new Signal("suspended"); @@ -95,6 +96,8 @@ export class Fixture { }); this.#signals.cleanup(() => broadcast.close()); + this.#broadcast = broadcast; + const video = new Publish.Video.Encoder("video", { broadcast, capture, @@ -183,8 +186,8 @@ export class Fixture { this.#signals.interval(schedule, (SCHEDULE_AHEAD * 1000) / 4); } - // Paint and emit one frame per pattern tick. The counter comes from the audio clock rather than - // the timer, so a late or coalesced tick skips a frame instead of drifting the two apart. + // Paint and request capture per pattern tick. Capture and encoding can finish later. The counter + // comes from the audio clock, so a coalesced tick skips a frame instead of drifting the two apart. #runPicture(ctx: CanvasRenderingContext2D, track: CanvasSource, start: number, fault: Fault): void { let painted = -1; @@ -202,7 +205,7 @@ export class Fixture { }; // Twice the frame rate: a tick that lands between frames is a no-op, and one that lands late - // still emits the frame the clock is on. + // still requests capture of the frame the clock is on. this.#timer = self.setInterval(tick, 1000 / Pattern.FPS / 2); this.#signals.cleanup(() => { if (this.#timer !== undefined) self.clearInterval(this.#timer); @@ -210,6 +213,13 @@ export class Fixture { }); } + /** Sample the current published keyframe before the viewer closes its subscription. */ + async liveGop() { + const broadcast = this.#broadcast.net.peek(); + if (!broadcast) throw new Error("the fixture is not publishing"); + return readLiveGop(broadcast.track("video")); + } + /** Stop publishing and release the capture, audio graph, and session. */ close(): void { this.#signals.close(); diff --git a/test/interop/clients/js/src/setup.ts b/test/interop/clients/js/src/setup.ts index aeb4177abb..1cd5275a10 100644 --- a/test/interop/clients/js/src/setup.ts +++ b/test/interop/clients/js/src/setup.ts @@ -55,6 +55,10 @@ if (role === "publish") { // The driver stops and restarts the publisher in place to exercise a same-path republish, which // has to reuse this page so the audio context keeps its user activation. publish({ + liveGop: () => { + if (!fixture) throw new Error("the fixture is stopped"); + return fixture.liveGop(); + }, stop: () => { fixture?.close(); fixture = undefined; diff --git a/test/justfile b/test/justfile index 1be6619e2c..fecc03f330 100644 --- a/test/justfile +++ b/test/justfile @@ -24,7 +24,7 @@ harness: ./lib/harness.test.sh cd .. && bun install --frozen-lockfile cd interop/clients/js && bunx playwright install chromium - cd interop/clients/js && bun test ./harness.browser.ts + cd interop/clients/js && bun test ./harness.browser.ts ./src/contract.test.ts # Cross-language media interop test, built from this checkout. Stands up a # relay and runs the publisher x subscriber matrix. Default: rust only (fast From 0165839e73627701d45b1ea4cea832759e16f67d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 17:22:26 -0700 Subject: [PATCH 14/15] quest: claim media-late-join Co-Authored-By: Claude Opus 5.5 From b4b39cda756cc326301b57452e235b75119bf4e0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 18:00:40 -0700 Subject: [PATCH 15/15] test(media): late join asserts a prompt start and a bounded catch-up The relay may hand a latecomer the GOP from before a demand gap, and the maintainer accepted showing it and then fast-forwarding. The check now asserts what a viewer sees instead of the first frame's keyframe: video within 5s of the page loading, then within 1.5s of the fixture's painted frame within 2s. Bounds are about twice the worst of 45 looped runs under ambient and stress-ng CPU load. Drops the liveGop command, readLiveGop, and lateJoinStartsLive, which only the old check used, and deletes the completed quest. Co-Authored-By: Claude Opus 5.5 --- quest/m1/media-audio-tone.md | 4 +- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/media-late-join.md | 54 ------------ test/interop/README.md | 9 +- test/interop/clients/js/media.ts | 89 ++++++++++++++++---- test/interop/clients/js/src/contract.test.ts | 36 +------- test/interop/clients/js/src/contract.ts | 34 -------- test/interop/clients/js/src/fixture.ts | 12 +-- test/interop/clients/js/src/setup.ts | 4 - 9 files changed, 80 insertions(+), 163 deletions(-) delete mode 100644 quest/m1/test-flakes-2/media-late-join.md diff --git a/quest/m1/media-audio-tone.md b/quest/m1/media-audio-tone.md index c1b8f1bc51..553eb78554 100644 --- a/quest/m1/media-audio-tone.md +++ b/quest/m1/media-audio-tone.md @@ -4,7 +4,8 @@ The `just test media` audio-tone check passes under load, fixed at its cause. It failed with 66 of 75 samples audible in #4719's runs and 72 of 81 in -the 2026-10-06 late-join loops (#4914), and passed on rerun. +the 2026-10-06 late-join loops (#4914), 68 of 81 once in 45 late-join runs +on 2026-10-09 under `stress-ng --cpu 20`, and passed on rerun. ## Plan @@ -20,5 +21,4 @@ loaded check. ## Related - [More tests under load](/quest/m1/test-flakes-2/README.md) - shares the loaded-check validation -- [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - shares the media fixture, but tracks a separate failure - [Forced channel count](/quest/m1/publish-audio-channel-count.md) - another source of audio gaps on the JS publish path; rule it in or out first diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index b0fc70f158..f5acdc531f 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -32,7 +32,6 @@ Public API: none. Wire: none. ## Required -- [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - a late joiner shows video promptly and catches up to live, and the check asserts that under load - [Import catalog finish](/quest/m1/test-flakes-2/import-catalog-finish.md) - `moq-cli`'s subprocess EOF catalog-finish test holds up under load with event-based fixture coordination - [Relay restart rebind](/quest/m1/test-flakes-2/relay-restart-rebind.md) - the crash drill restarts on its original UDP address under concurrent load - [Impaired handshake](/quest/m1/test-flakes-2/impaired-handshake.md) - the impaired cluster drills' clients never time out while connecting diff --git a/quest/m1/test-flakes-2/media-late-join.md b/quest/m1/test-flakes-2/media-late-join.md deleted file mode 100644 index e22bb0d9d8..0000000000 --- a/quest/m1/test-flakes-2/media-late-join.md +++ /dev/null @@ -1,54 +0,0 @@ -# [S] Media late join catches up to live - -## Goal - -A viewer joining a broadcast through a relay after a demand gap shows video -promptly and reaches the live edge within a bound, and the late-join check -asserts exactly that and passes under load. - -## Plan - -It failed once after [#4181](https://github.com/moq-dev/moq/pull/4181) ("16 -frames behind" a 15-frame GOP), and the interop browser check failed the -same way on `main`. `just test media` is that browser lane through a Rust -`moq-relay`; there is no separate Rust media test. -[#4719](https://github.com/moq-dev/moq/pull/4719) now compares the -latecomer against the published keyframe instead of the painted counter. - -It is real, not fixture sampling: the latecomer is handed a GOP from before -the demand gap. Reproduced by looping `bun media.ts --cases late-join` -against a local relay with debug logs. In the fixture, the old viewer -leaves, the publisher pauses (closing its open GOP and writing an empty -marker group), and the latecomer joins about a second later. - -[#4914](https://github.com/moq-dev/moq/pull/4914) fixed two relay windows, -each with a deterministic test: - -- A received group stays hidden from readers until its first frame lands or - its stream ends, and an idle copy goes live only once the cache shows the - route's answer (`rejoin_waits_for_the_answers_first_frame`). -- A rejoin's answer is judged against the groups that survived the leave, - not the idle snapshot, since the cancel resets the group in flight - (`an_answer_past_a_reset_group_skips_the_cache`). - -What remains (1 of 30 runs after both fixes): the latecomer's max delay rises -to its playout delay (about 145ms) right after it subscribes. When the -pre-gap GOP was short, the GOP before it still reaches within that budget of -the marker, so the relay hands it over by design. The player then presents -from that GOP's keyframe, not from its playout position. Relay log: groups -7, 8, 9 served in 1ms with a 145ms budget, and the player shows frames 360ms -behind the published keyframe. - -Decided 2026-10-06 by the maintainer: a joiner showing the reachable pre-gap -GOP at once and then fast-forwarding to live is the wanted behavior, so no -player or consumer change. The check is what's too strict: it asserts the -first presented frame is at or after the newest published keyframe. - -Remaining work: the late-join check (`just test media` and the interop -browser lane) asserts what a viewer cares about instead: video shows within a -start bound, and the player reaches the live edge within a catch-up bound. -Set both bounds from measurements, looped under synthetic CPU load (the -maintainer approved a bounded load generator such as `stress-ng` for these -runs), and keep the relay fixes' deterministic tests as the regression guard. - -Public API: none. Wire: none. diff --git a/test/interop/README.md b/test/interop/README.md index 3359d53444..8fc5d53e08 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -186,10 +186,11 @@ Each run covers, against a real local relay: The fake device is not physical hardware, and the headless permission decision is not a person clicking a browser prompt. - **pause and resume**, **unsubscribe and rejoin**, **detach and reattach**, - **publisher stop and same-path republish**, and **late join**. The late join - must present the newest published GOP or a newer one. Its lower bound is the - encoded keyframe timestamp sampled before the existing viewer closes, since - painting the canvas does not mean capture and encoding have finished. + **publisher stop and same-path republish**, and **late join**. The late + joiner may first show the GOP from before the demand gap, but it must present + a frame within 5s of loading and then come within 1.5s of the fixture's + painted frame within 2s, both bounds about twice the worst seen in looped runs + under CPU load. - **resources return to baseline** - the page wraps `WebTransport`, `WebSocket`, `AudioContext`, and `Worker` to count live instances, so a detach that leaks a session is visible rather than merely invisible. diff --git a/test/interop/clients/js/media.ts b/test/interop/clients/js/media.ts index 2d24122eb4..3570b6ea1b 100644 --- a/test/interop/clients/js/media.ts +++ b/test/interop/clients/js/media.ts @@ -47,7 +47,6 @@ import { import { type CaptureState, FAULTS, - lateJoinStartsLive, leakedPlayerStarted, SAMPLE_MS, SAMPLE_RATE, @@ -166,6 +165,61 @@ async function waitFrozen(page: Page, errors: BrowserErrors, assertion: string, throw new Failure(assertion, `${description}: the presented frame is still advancing, now ${frame}`); } +/** + * A presented frame at most this far behind the fixture's painted counter counts as live. + * + * The lag spans capture, encode, relay, playout delay, and decode. Loaded runs settle at 4 to 24 + * frames with spikes to 33, while a player stuck on the GOP from before a demand gap sits 60 or more + * behind. + */ +const LIVE_LAG_FRAMES = 45; + +/** How long a latecomer may take to present its first frame, from its page loading. Loaded runs take up to 2.6s. */ +const LATE_START_MS = 5000; + +/** How long a latecomer may take to reach live after its first frame. Loaded runs take up to 0.3s. */ +const LATE_CATCHUP_MS = 2000; + +const framesToMs = (frames: number) => Math.round((frames * 1000) / Pattern.FPS); + +/** + * Wait until the player presents a frame within {@link LIVE_LAG_FRAMES} of the one the fixture last + * painted, and report how far behind its first reading was. + * + * Both pages are read at once, in separate browsers, so the lag is what a viewer would see against + * the source. + */ +async function waitForLive( + publisher: Page, + publisherErrors: BrowserErrors, + player: Page, + playerErrors: BrowserErrors, +): Promise<{ first: number; lag: number; after: number }> { + const since = Date.now(); + let first: number | undefined; + let lag: number | undefined; + while (Date.now() < since + LATE_CATCHUP_MS) { + throwPageErrors(publisherErrors); + throwPageErrors(playerErrors); + const [fixture, state] = await Promise.all([readFixtureState(publisher), readPlayerState(player)]).catch( + () => [], + ); + if (fixture && state?.frameId !== undefined) { + lag = fixture.frameId - state.frameId; + first ??= lag; + if (lag <= LIVE_LAG_FRAMES) return { first, lag, after: Date.now() - since }; + } + await sleep(POLL_INTERVAL_MS); + } + throwPageErrors(publisherErrors); + throwPageErrors(playerErrors); + throw new Failure( + "late join reaches live", + `still ${lag === undefined ? "?" : framesToMs(lag)}ms behind the fixture ${LATE_CATCHUP_MS}ms after the first frame, ` + + `against a ${framesToMs(LIVE_LAG_FRAMES)}ms bound`, + ); +} + // One sample as a line: elapsed time, the frame on the canvas, the tone step heard against the one // that frame belongs to, how far the tone stood above the noise floor, and whether audio was // arriving at all, which separates a silent player from a publisher that stopped sending. @@ -691,30 +745,29 @@ try { } // ── late join ──────────────────────────────────────────────────────────── - // A fresh page against a broadcast that has been running for a while: the player has to tune in - // at the live edge rather than replay what it missed. + // A fresh page against a broadcast that has been running for a while, after a demand gap. The + // relay may hand over the GOP from before the gap, so the first frame can be stale; the player + // has to show it promptly and then fast-forward to live rather than replay what it missed. if (wants("late-join")) { console.error("=== late join ==="); - const live = await command(publisher, "liveGop"); await player.close(); [player, playerErrors] = await subscriber(broadcast, "latecomer"); + const joined = Date.now(); await gesture(player); - const joined = await waitForState(player, playerErrors, { - deadline: Date.now() + timeoutMs, - assertion: "late join presents media", - description: "the latecomer to present the fixture", - predicate: (state) => state.frameId !== undefined && state.audioContext === "running", + + const shown = await waitForState(player, playerErrors, { + deadline: joined + LATE_START_MS, + assertion: "late join shows video", + description: `a presented fixture frame within ${LATE_START_MS}ms of the latecomer loading`, + predicate: (state) => state.frameId !== undefined, }); - // requestFrame() is asynchronous, and capture timestamps decide keyframes. The painted - // counter can therefore be over 15 ticks ahead of a still-current encoded GOP under load. - // Compare against its actual published keyframe, sampled while the old viewer held demand. - check( - lateJoinStartsLive(live, joined.videoTimestamp), - "late join starts live", - () => - `joined at frame ${joined.frameId}, timestamp ${joined.videoTimestamp}ms before the current GOP's ${live.timestamp}ms keyframe`, + const startMs = Date.now() - joined; + + const live = await waitForLive(publisher, publisherErrors, player, playerErrors); + console.error( + ` showed frame ${shown.frameId} after ${startMs}ms, ${framesToMs(live.first)}ms behind the fixture; ` + + `live ${live.after}ms later, ${framesToMs(live.lag)}ms behind`, ); - console.error(` joined at frame ${joined.frameId}, timestamp ${joined.videoTimestamp}ms, current GOP began at ${live.timestamp}ms`); assertMedia(await collect(player, playerErrors, WINDOW_MS), "late join"); } diff --git a/test/interop/clients/js/src/contract.test.ts b/test/interop/clients/js/src/contract.test.ts index 33deb26acc..d22d56c5a5 100644 --- a/test/interop/clients/js/src/contract.test.ts +++ b/test/interop/clients/js/src/contract.test.ts @@ -1,6 +1,5 @@ import { expect, test } from "bun:test"; -import * as Moq from "@moq/net"; -import { lateJoinStartsLive, leakedPlayerStarted, readLiveGop, type Resources } from "./contract"; +import { leakedPlayerStarted, type Resources } from "./contract"; const playing: Resources = { transports: 1, sockets: 0, audioContexts: 1, workers: 0 }; @@ -19,36 +18,3 @@ test("unchanged counts are not a leak start", () => { expect(leakedPlayerStarted(playing, playing)).toBe(false); }); - -test("late join uses the published GOP when painting is ahead of capture", async () => { - const broadcast = new Moq.Broadcast.Producer(); - const track = broadcast.createTrack("video", { timescale: Moq.Time.Timescale.MILLI }); - try { - const old = track.appendGroup(); - old.writeFrame({ payload: new Uint8Array([1]), timestamp: Moq.Time.Timestamp.fromMillis(3000) }); - old.close(); - const current = track.appendGroup(); - current.writeFrame({ payload: new Uint8Array([2]), timestamp: Moq.Time.Timestamp.fromMillis(3700) }); - current.writeFrame({ payload: new Uint8Array([3]), timestamp: Moq.Time.Timestamp.fromMillis(4199) }); - - // Painted counter 127 can precede capture while GOP 111 remains current. The old - // 127 - 111 <= 15 check rejects a valid join; an earlier published GOP must fail. - const viewer = track.subscribe(); - const gop = await readLiveGop(broadcast.track("video")); - expect(gop.timestamp).toBe(3700); - expect(lateJoinStartsLive(gop, 3700)).toBe(true); - expect(lateJoinStartsLive(gop, 4199)).toBe(true); - expect(lateJoinStartsLive(gop, 3000)).toBe(false); - expect(lateJoinStartsLive(gop, undefined)).toBe(false); - expect(track.demand().used.peek()).toBe(true); - const viewed = await viewer.recvGroup(); - expect((await viewed?.readFrame())?.payload).toEqual(new Uint8Array([2])); - expect((await viewed?.readFrame())?.payload).toEqual(new Uint8Array([3])); - viewed?.close(); - viewer.close(); - await track.demand().unused(); - expect(track.demand().used.peek()).toBe(false); - } finally { - broadcast.close(); - } -}); diff --git a/test/interop/clients/js/src/contract.ts b/test/interop/clients/js/src/contract.ts index 1211e7d087..cb5562dcb6 100644 --- a/test/interop/clients/js/src/contract.ts +++ b/test/interop/clients/js/src/contract.ts @@ -8,8 +8,6 @@ * @module */ -import type * as Moq from "@moq/net"; - // ── the deterministic publisher ───────────────────────────────────────────── /** A deliberate defect, used to prove an assertion can fail. */ @@ -47,36 +45,6 @@ export type FixtureState = { videoEnabled: boolean; }; -/** The newest published video's GOP, sampled independently of canvas capture and encoding. */ -export type LiveGop = { - /** The timestamp of its keyframe, in milliseconds on the publisher's media clock. */ - timestamp: number; -}; - -/** Read the newest published GOP without retaining a subscription or changing its frames. */ -export async function readLiveGop(track: Moq.Track.Consumer): Promise { - const subscriber = track.subscribe(); - try { - const group = await subscriber.recvGroup(); - if (!group) throw new Error("the fixture video has no published GOP"); - try { - const frame = await group.readFrame(); - if (!frame) throw new Error("the fixture GOP has no keyframe"); - if (!frame.timestamp) throw new Error("the fixture video arrived untimed"); - return { timestamp: frame.timestamp.asMillis() }; - } finally { - group.close(); - } - } finally { - subscriber.close(); - } -} - -/** True when the presented frame belongs to the sampled GOP or a newer one. */ -export function lateJoinStartsLive(gop: LiveGop, timestamp: number | undefined): boolean { - return timestamp !== undefined && timestamp >= gop.timestamp; -} - /** The camera publisher state mirrored onto its element for Playwright. */ export type CaptureState = { videoError?: string; @@ -204,8 +172,6 @@ export type CloseState = * owns them rather than the driver reaching in. */ export type InteropControl = { - /** Read the current video GOP while an existing viewer is still pulling media. */ - liveGop(): Promise; /** Stop the fixture publisher, releasing its session. */ stop(): void; /** Start (or restart) the fixture publisher on the same broadcast path. */ diff --git a/test/interop/clients/js/src/fixture.ts b/test/interop/clients/js/src/fixture.ts index d8f141a78f..dbbe25ab67 100644 --- a/test/interop/clients/js/src/fixture.ts +++ b/test/interop/clients/js/src/fixture.ts @@ -17,7 +17,7 @@ import { Time } from "@moq/net"; import * as Publish from "@moq/publish"; import { Effect, Signal } from "@moq/signals"; import type { Fault, FixtureState } from "./contract"; -import { KEYFRAME_INTERVAL_MS, OFFSET_STEPS, readLiveGop, SAMPLE_RATE } from "./contract"; +import { KEYFRAME_INTERVAL_MS, OFFSET_STEPS, SAMPLE_RATE } from "./contract"; import * as Pattern from "./pattern"; /** Cap the encoder rather than letting it track a bandwidth estimate, so runs are comparable. */ @@ -47,7 +47,6 @@ export class Fixture { readonly host: HTMLElement; readonly #signals = new Effect(); - readonly #broadcast: Publish.Broadcast; readonly #audio: AudioContext; readonly #frameId = new Signal(-1); readonly #audioState = new Signal("suspended"); @@ -97,8 +96,6 @@ export class Fixture { }); this.#signals.cleanup(() => broadcast.close()); - this.#broadcast = broadcast; - const video = new Publish.Video.Encoder("video", { broadcast, capture, @@ -220,13 +217,6 @@ export class Fixture { this.#videoEnabled.set(enabled); } - /** Sample the current published keyframe before the viewer closes its subscription. */ - async liveGop() { - const broadcast = this.#broadcast.net.peek(); - if (!broadcast) throw new Error("the fixture is not publishing"); - return readLiveGop(broadcast.track("video")); - } - /** Stop publishing and release the capture, audio graph, and session. */ close(): void { this.#signals.close(); diff --git a/test/interop/clients/js/src/setup.ts b/test/interop/clients/js/src/setup.ts index c6525e4ae1..46bc44b22f 100644 --- a/test/interop/clients/js/src/setup.ts +++ b/test/interop/clients/js/src/setup.ts @@ -57,10 +57,6 @@ if (role === "publish") { // The driver stops and restarts the publisher in place to exercise a same-path republish, which // has to reuse this page so the audio context keeps its user activation. publish({ - liveGop: () => { - if (!fixture) throw new Error("the fixture is stopped"); - return fixture.liveGop(); - }, stop: () => { fixture?.close(); fixture = undefined;