From 7fce76cf23b90eca694db628bc29e5cdffea90f6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 13:30:55 -0800 Subject: [PATCH 1/5] chore(quest): claim quest/m1/mock-clock-tests/rs-transport.md Quest-Claim: a5c23dc35ee65845 From f4f126e18e3618148d993d8f174ab196e352e32a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 14:06:00 -0800 Subject: [PATCH 2/5] test: transport crate tests stop waiting on real time Co-Authored-By: Claude Opus 5.5 --- rs/justfile | 2 +- rs/moq-quic/src/connection/mtud.rs | 34 +++---- .../src/tests/characterize/endpoint.rs | 22 ++--- rs/moq-sock/src/shard.rs | 22 +++-- rs/moq-sock/tests/udp/main.rs | 92 ++++++------------ rs/moq-tokio/src/jemalloc.rs | 92 ++++++++++-------- rs/moq-tokio/src/mdns.rs | 38 ++------ rs/moq-tokio/src/quic/driver/tests.rs | 96 ++++++++++--------- rs/moq-tokio/tests/integration/broadcast.rs | 79 +++++++-------- rs/moq-tokio/tests/integration/reconnect.rs | 44 +-------- .../integration/subscription_end_integrity.rs | 27 +++--- rs/moq-uring/src/worker.rs | 13 ++- .../tests/integration/socket_queues.rs | 29 ++++-- 13 files changed, 267 insertions(+), 323 deletions(-) diff --git a/rs/justfile b/rs/justfile index 6b3f5b8e152..e4e43acfaa9 100644 --- a/rs/justfile +++ b/rs/justfile @@ -450,7 +450,7 @@ render-sw: # Run the ignored mDNS tests, which multicast on the host's network interfaces. multicast: - just rs test -p moq-tokio --features mdns --lib --run-ignored only -E 'test(=mdns::tests::apps_partition_discovery) | test(/^mdns::tests::daemon_/)' + just rs test -p moq-tokio --features mdns --lib --run-ignored only -E 'test(=mdns::tests::same_app_peers_find_each_other) | test(/^mdns::tests::daemon_/)' just rs test -p moq-relay --features cluster-lan --run-ignored only -E 'test(/^lan_mesh::/)' # Run the long TS recovery arms; a failing arm keeps its export and ledgers under $HARNESS_RUNS. diff --git a/rs/moq-quic/src/connection/mtud.rs b/rs/moq-quic/src/connection/mtud.rs index fac3b48bf7b..350e9ee3363 100644 --- a/rs/moq-quic/src/connection/mtud.rs +++ b/rs/moq-quic/src/connection/mtud.rs @@ -549,7 +549,7 @@ mod tests { assert_eq!(mtud.black_hole_detector.suspicious_loss_burst_count(), 0); mtud.on_non_probe_lost(4, 800); - assert!(!mtud.black_hole_detected(Instant::now())); + assert!(!mtud.black_hole_detected(Instant::ZERO)); assert_eq!(mtud.black_hole_detector.largest_non_probe_lost(), None); assert_eq!(mtud.black_hole_detector.suspicious_loss_burst_count(), 0); } @@ -562,7 +562,7 @@ mod tests { assert_eq!(mtud.black_hole_detector.largest_non_probe_lost(), Some(3)); assert_eq!(mtud.black_hole_detector.suspicious_loss_burst_count(), 0); - assert!(!mtud.black_hole_detected(Instant::now())); + assert!(!mtud.black_hole_detected(Instant::ZERO)); assert_eq!(mtud.black_hole_detector.largest_non_probe_lost(), None); assert_eq!(mtud.black_hole_detector.suspicious_loss_burst_count(), 1); } @@ -570,21 +570,21 @@ mod tests { #[test] fn black_hole_detector_ignores_empty_burst() { let mut mtud = default_mtud(); - assert!(!mtud.black_hole_detected(Instant::now())); + assert!(!mtud.black_hole_detected(Instant::ZERO)); assert_eq!(mtud.black_hole_detector.suspicious_loss_burst_count(), 0); } #[test] fn mtu_discovery_disabled_does_nothing() { let mut mtud = MtuDiscovery::disabled(1_200, 1_200); - let probe_size = mtud.poll_transmit(Instant::now(), 0); + let probe_size = mtud.poll_transmit(Instant::ZERO, 0); assert_eq!(probe_size, None); } #[test] fn mtu_discovery_disabled_lost_four_packet_bursts_triggers_black_hole_detection() { let mut mtud = MtuDiscovery::disabled(1_400, 1_250); - let now = Instant::now(); + let now = Instant::ZERO; for i in 0..4 { // The packets are never contiguous, so each one has its own burst @@ -599,7 +599,7 @@ mod tests { #[test] fn mtu_discovery_lost_two_packet_bursts_does_not_trigger_black_hole_detection() { let mut mtud = default_mtud(); - let now = Instant::now(); + let now = Instant::ZERO; for i in 0..2 { mtud.on_non_probe_lost(i, 1300); @@ -610,7 +610,7 @@ mod tests { #[test] fn mtu_discovery_lost_four_packet_bursts_triggers_black_hole_detection_and_resets_timer() { let mut mtud = default_mtud(); - let now = Instant::now(); + let now = Instant::ZERO; for i in 0..4 { // The packets are never contiguous, so each one has its own burst @@ -633,7 +633,7 @@ mod tests { ..mtu::Config::default() }; let mut mtud = MtuDiscovery::new(1_200, 1_200, config); - let now = Instant::now(); + let now = Instant::ZERO; drive_to_completion(&mut mtud, now, 1_500); // Polling right after completion does not cause new packets to be sent @@ -660,7 +660,7 @@ mod tests { let mut mtud = default_mtud(); let mut probe_sizes = (0..4).map(|i| { - let probe_size = mtud.poll_transmit(Instant::now(), i); + let probe_size = mtud.poll_transmit(Instant::ZERO, i); assert!(probe_size.is_some(), "no probe returned for packet {i}"); mtud.on_probe_lost(); @@ -684,7 +684,7 @@ mod tests { let mut mtud = default_mtud(); mtud.on_peer_max_udp_payload_size_received(1300); - let probed_sizes = drive_to_completion(&mut mtud, Instant::now(), 1500); + let probed_sizes = drive_to_completion(&mut mtud, Instant::ZERO, 1500); assert_eq!(mtud.state.as_ref().unwrap().peer_max_udp_payload_size, 1300); assert_eq!(mtud.current_mtu, 1300); @@ -701,7 +701,7 @@ mod tests { assert_eq!(mtud.current_mtu, 1400); assert_eq!(mtud.state.as_ref().unwrap().peer_max_udp_payload_size, 1400); - let probed_sizes = drive_to_completion(&mut mtud, Instant::now(), 1500); + let probed_sizes = drive_to_completion(&mut mtud, Instant::ZERO, 1500); assert_eq!(mtud.current_mtu, 1400); assert!(probed_sizes.is_empty()); @@ -713,7 +713,7 @@ mod tests { #[should_panic(expected = "Transport parameters received after MTU probing started")] fn mtu_discovery_with_peer_max_udp_payload_size_during_search_panics() { let mut mtud = default_mtud(); - assert!(mtud.poll_transmit(Instant::now(), 0).is_some()); + assert!(mtud.poll_transmit(Instant::ZERO, 0).is_some()); assert!(matches!(mtud.state.as_ref().unwrap().phase, Phase::Searching(_))); mtud.on_peer_max_udp_payload_size_received(1300); } @@ -722,7 +722,7 @@ mod tests { fn mtu_discovery_with_1500_limit() { let mut mtud = default_mtud(); - let probed_sizes = drive_to_completion(&mut mtud, Instant::now(), 1500); + let probed_sizes = drive_to_completion(&mut mtud, Instant::ZERO, 1500); let expected_probed_sizes = &[1326, 1389, 1420, 1452]; assert_eq!(probed_sizes, expected_probed_sizes); @@ -738,7 +738,7 @@ mod tests { }; let mut mtud = MtuDiscovery::new(1_200, 1_200, config); - let probed_sizes = drive_to_completion(&mut mtud, Instant::now(), 1500); + let probed_sizes = drive_to_completion(&mut mtud, Instant::ZERO, 1500); let expected_probed_sizes = &[ 5600, 5600, 5600, 3399, 3399, 3399, 2299, 2299, 2299, 1749, 1749, 1749, 1474, 1611, 1611, 1611, 1542, 1542, @@ -757,7 +757,7 @@ mod tests { }; let mut mtud = MtuDiscovery::new(1200, 1200, config); - drive_to_completion(&mut mtud, Instant::now(), u16::MAX); + drive_to_completion(&mut mtud, Instant::ZERO, u16::MAX); assert_eq!(mtud.current_mtu, 65527); assert!(completed(&mtud)); @@ -771,7 +771,7 @@ mod tests { }; let mut mtud = MtuDiscovery::new(1200, 1200, config); - let now = Instant::now(); + let now = Instant::ZERO; let mut iterations = 0; for i in 1..100 { iterations += 1; @@ -779,7 +779,7 @@ mod tests { let probe_pn = i * 2 - 1; let other_pn = i * 2; - let result = mtud.poll_transmit(Instant::now(), probe_pn); + let result = mtud.poll_transmit(Instant::ZERO, probe_pn); if completed(&mtud) { break; diff --git a/rs/moq-quic/src/tests/characterize/endpoint.rs b/rs/moq-quic/src/tests/characterize/endpoint.rs index 0b870926c8e..85c0cdeaf7a 100644 --- a/rs/moq-quic/src/tests/characterize/endpoint.rs +++ b/rs/moq-quic/src/tests/characterize/endpoint.rs @@ -107,7 +107,7 @@ impl Answer { fn unknown_cid_gets_stateless_reset() { let _guard = subscribe(); let mut server = endpoint(true); - let now = Instant::now(); + let now = Instant::ZERO; let cid = valid_cid(); let packet = short(&cid, 100); @@ -126,7 +126,7 @@ fn unknown_cid_gets_stateless_reset() { #[test] fn small_unknown_packet_gets_no_reset() { let _guard = subscribe(); - let now = Instant::now(); + let now = Instant::ZERO; let cid = valid_cid(); assert_eq!(answer(&mut endpoint(true), now, &short(&cid, 21)), Answer::Nothing); let Answer::Response(reset) = answer(&mut endpoint(true), now, &short(&cid, 22)) else { @@ -139,7 +139,7 @@ fn small_unknown_packet_gets_no_reset() { #[test] fn invalid_cid_is_dropped() { let _guard = subscribe(); - let now = Instant::now(); + let now = Instant::ZERO; assert_eq!( answer(&mut endpoint(true), now, &short(&[0xab; 8], 100)), Answer::Nothing @@ -154,14 +154,14 @@ fn zero_length_cid_short_packet_is_dropped() { ..crate::endpoint::Config::default() }; let mut server = Endpoint::new(config.with_server(server_config()), true).unwrap(); - assert_eq!(answer(&mut server, Instant::now(), &short(&[], 100)), Answer::Nothing); + assert_eq!(answer(&mut server, Instant::ZERO, &short(&[], 100)), Answer::Nothing); } /// Only Initials can start a connection, so another long header for an unknown CID is dropped #[test] fn long_header_for_unknown_connection_is_dropped() { let _guard = subscribe(); - let now = Instant::now(); + let now = Instant::ZERO; let packet = long(2, Version::V1.wire(), &valid_cid(), 1200); assert_eq!(answer(&mut endpoint(true), now, &packet), Answer::Nothing); } @@ -172,7 +172,7 @@ fn long_header_for_unknown_connection_is_dropped() { fn short_initial_is_dropped() { let _guard = subscribe(); let packet = long(0, Version::V1.wire(), &[0x11; 8], 1199); - assert_eq!(answer(&mut endpoint(true), Instant::now(), &packet), Answer::Nothing); + assert_eq!(answer(&mut endpoint(true), Instant::ZERO, &packet), Answer::Nothing); let mut pair = pair_with(|_| {}); pair.begin_connect(client_with(|_| {})); @@ -188,7 +188,7 @@ fn short_initial_is_dropped() { fn initial_with_short_cid_is_refused() { let _guard = subscribe(); let packet = long(0, Version::V1.wire(), &[0x11; 7], 1200); - let answer = answer(&mut endpoint(true), Instant::now(), &packet); + let answer = answer(&mut endpoint(true), Instant::ZERO, &packet); let Answer::Response(close) = answer else { panic!("expected an Initial close, got {answer:?}"); }; @@ -202,7 +202,7 @@ fn initial_with_short_cid_is_refused() { fn undecryptable_initial() { let _guard = subscribe(); let mut server = endpoint(true); - let now = Instant::now(); + let now = Instant::ZERO; let mut packet = long(0, Version::V1.wire(), &[0x11; 8], 1200); assert_eq!(answer(&mut server, now, &packet), Answer::Nothing); @@ -238,7 +238,7 @@ fn undecryptable_initial() { #[test] fn undecodable_datagram_is_dropped() { let _guard = subscribe(); - let now = Instant::now(); + let now = Instant::ZERO; // A long header cut off inside its version assert_eq!(answer(&mut endpoint(true), now, &[0xc0, 0, 0]), Answer::Nothing); // An Initial whose two-byte length field, after its 20 bytes of header, runs past the @@ -252,7 +252,7 @@ fn undecodable_datagram_is_dropped() { #[test] fn wrong_version_initial() { let _guard = subscribe(); - let now = Instant::now(); + let now = Instant::ZERO; let packet = long(0, 0x0a1a_2a3a, &[0x11; 8], 1200); assert!(matches!(answer(&mut endpoint(true), now, &packet), Answer::Response(_))); assert_eq!(answer(&mut endpoint(false), now, &packet), Answer::Nothing); @@ -265,7 +265,7 @@ fn wrong_version_initial() { #[test] fn client_endpoint_sends_stateless_reset() { let _guard = subscribe(); - let now = Instant::now(); + let now = Instant::ZERO; let cid = valid_cid(); let mut client = endpoint(false); assert!(answer(&mut client, now, &short(&cid, 100)).is_reset_for(&client, cid)); diff --git a/rs/moq-sock/src/shard.rs b/rs/moq-sock/src/shard.rs index 3a91e888175..47896662fc3 100644 --- a/rs/moq-sock/src/shard.rs +++ b/rs/moq-sock/src/shard.rs @@ -687,6 +687,20 @@ mod tests { /// Which member received something, given every send targets exactly one. fn receiver(group: &[UdpSocket]) -> Option { + use std::os::fd::AsRawFd; + + // Block until a member is readable: the filter runs on the send, but the + // datagram lands in the member's queue asynchronously. + let mut fds: Vec<_> = group + .iter() + .map(|socket| libc::pollfd { + fd: socket.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }) + .collect(); + let n = unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as libc::nfds_t, 5_000) }; + assert!(n >= 0, "poll failed: {}", std::io::Error::last_os_error()); let mut buf = [0u8; 64]; group.iter().position(|socket| socket.recv_from(&mut buf).is_ok()) } @@ -712,14 +726,6 @@ mod tests { // can be steering this. let sender = crate::bind::udp(crate::bind::Udp::new("127.0.0.1:0".parse().unwrap())).unwrap(); sender.send_to(packet, addr).unwrap(); - - // The send is local and the filter is synchronous, but the receive - // still has to be given a moment to land. - #[expect( - clippy::disallowed_methods, - reason = "a sync test with no runtime waits on a real kernel socket" - )] - std::thread::sleep(std::time::Duration::from_millis(50)); assert_eq!( receiver(&sockets), Some(index), diff --git a/rs/moq-sock/tests/udp/main.rs b/rs/moq-sock/tests/udp/main.rs index 61dbf5cb87f..30539463f45 100644 --- a/rs/moq-sock/tests/udp/main.rs +++ b/rs/moq-sock/tests/udp/main.rs @@ -791,24 +791,11 @@ fn recv_transport_error() { ) .unwrap(); - let mut received = None; - for _ in 0..100 { - match state.recv_transport_error((&sock).into()) { - Ok(Some(err)) => { - received = Some(err); - break; - } - _ => { - #[expect( - clippy::disallowed_methods, - reason = "a sync test with no runtime polls a real kernel socket" - )] - std::thread::sleep(Duration::from_millis(10)); - } - } - } - - let err = received.expect("ICMP Port Unreachable was not received"); + wait_for_error(&sock); + let err = state + .recv_transport_error((&sock).into()) + .expect("failed to read the error queue") + .expect("ICMP Port Unreachable was not received"); assert!( matches!(err.payload, moq_sock::udp::TransportErrorPayload::Unreachable), @@ -855,21 +842,11 @@ fn recv_transport_error_ipv6() { ) .unwrap(); - let mut received = None; - - for _ in 0..100 { - if let Ok(Some(err)) = state.recv_transport_error((&sock).into()) { - received = Some(err); - break; - } - #[expect( - clippy::disallowed_methods, - reason = "a sync test with no runtime polls a real kernel socket" - )] - std::thread::sleep(Duration::from_millis(10)); - } - - let err = received.expect("ICMPv6 Port Unreachable not received"); + wait_for_error(&sock); + let err = state + .recv_transport_error((&sock).into()) + .expect("failed to read the error queue") + .expect("ICMPv6 Port Unreachable not received"); assert!(matches!(err.payload, moq_sock::udp::TransportErrorPayload::Unreachable)); @@ -916,31 +893,15 @@ fn draining_error_queue_before_send_prevents_absorption() { ) .unwrap(); - #[expect( - clippy::disallowed_methods, - reason = "a sync test with no runtime polls a real kernel socket" - )] - std::thread::sleep(Duration::from_millis(20)); - // Drain the error queue before sending - this is the intended usage pattern - let mut got_error = false; - for _ in 0..10 { - match state.recv_transport_error((&sock).into()) { - Ok(Some(_)) => { - got_error = true; - break; - } - #[expect( - clippy::disallowed_methods, - reason = "a sync test with no runtime polls a real kernel socket" - )] - _ => std::thread::sleep(Duration::from_millis(10)), - } - } - assert!(got_error, "expected one queued error"); + wait_for_error(&sock); + state + .recv_transport_error((&sock).into()) + .expect("failed to read the error queue") + .expect("expected one queued error"); let receiver = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); - receiver.set_nonblocking(true).unwrap(); + receiver.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); state .try_send( @@ -955,15 +916,24 @@ fn draining_error_queue_before_send_prevents_absorption() { ) .expect("send failed after draining the error queue"); - #[expect( - clippy::disallowed_methods, - reason = "a sync test with no runtime polls a real kernel socket" - )] - std::thread::sleep(Duration::from_millis(20)); - let mut buf = [0u8; 64]; let n = receiver .recv(&mut buf) .expect("packet was not delivered after draining the error queue"); assert_eq!(&buf[..n], b"connection close"); } + +/// Block until `sock` has an error queued, which the kernel reports as `POLLERR`. +#[cfg(any(target_os = "linux", target_os = "android"))] +fn wait_for_error(sock: &Socket) { + use std::os::fd::AsRawFd; + + let mut pfd = libc::pollfd { + fd: sock.as_raw_fd(), + events: 0, + revents: 0, + }; + let n = unsafe { libc::poll(&mut pfd, 1, 5_000) }; + assert_eq!(n, 1, "no error was queued"); + assert_ne!(pfd.revents & libc::POLLERR, 0, "revents {:#x}", pfd.revents); +} diff --git a/rs/moq-tokio/src/jemalloc.rs b/rs/moq-tokio/src/jemalloc.rs index 971b44b5c56..abc4c5c605f 100644 --- a/rs/moq-tokio/src/jemalloc.rs +++ b/rs/moq-tokio/src/jemalloc.rs @@ -25,6 +25,13 @@ fn signal_ready() -> &'static tokio::sync::Notify { READY.get_or_init(tokio::sync::Notify::new) } +/// Notified once each SIGUSR1 has been handled, so a test needn't poll for its effect. +#[cfg(test)] +fn signal_handled() -> &'static tokio::sync::Notify { + static HANDLED: std::sync::OnceLock = std::sync::OnceLock::new(); + HANDLED.get_or_init(tokio::sync::Notify::new) +} + /// Whether jemalloc built its profiling backend, which is what makes `prof.active` writable. /// /// Fixed at startup by `MALLOC_CONF=prof:true`; there is no way to turn it on later, which is the @@ -62,42 +69,49 @@ pub async fn run() -> crate::Result<()> { loop { sig.recv().await; - if !ready { - tracing::warn!( - "heap profile requested, but jemalloc profiling was not enabled at startup (MALLOC_CONF=prof:true)" - ); - continue; - } + dump(ready).await; + #[cfg(test)] + signal_handled().notify_one(); + } + }) + .await +} - // Start sampling on the first request rather than at startup, so an unprofiled process - // carries none of the cost. Only allocations from here on are sampled: whatever the heap - // already holds is not attributed retroactively, so a dump taken to explain an - // already-large heap wants `prof_active:true` from the start instead. - if !sampling() { - match blocking(|| unsafe { raw::write(b"prof.active\0", true) }).await { - Ok(Ok(())) => { - tracing::info!("jemalloc heap profiling started; this dump covers only allocations from now on") - } - Ok(Err(err)) => { - tracing::error!(%err, "failed to start heap profiling"); - continue; - } - Err(err) => { - tracing::error!(%err, "start-profiling task failed"); - continue; - } - } - } +/// Answer one SIGUSR1: start sampling if it is off, then dump a profile. +async fn dump(ready: bool) { + if !ready { + tracing::warn!( + "heap profile requested, but jemalloc profiling was not enabled at startup (MALLOC_CONF=prof:true)" + ); + return; + } - // Null pointer tells jemalloc to use prof_prefix from MALLOC_CONF. - match blocking(|| unsafe { raw::write(b"prof.dump\0", std::ptr::null::()) }).await { - Ok(Ok(())) => tracing::info!("heap profile dumped"), - Ok(Err(err)) => tracing::error!(%err, "failed to dump heap profile"), - Err(err) => tracing::error!(%err, "heap profile task failed"), + // Start sampling on the first request rather than at startup, so an unprofiled process + // carries none of the cost. Only allocations from here on are sampled: whatever the heap + // already holds is not attributed retroactively, so a dump taken to explain an + // already-large heap wants `prof_active:true` from the start instead. + if !sampling() { + match blocking(|| unsafe { raw::write(b"prof.active\0", true) }).await { + Ok(Ok(())) => { + tracing::info!("jemalloc heap profiling started; this dump covers only allocations from now on") + } + Ok(Err(err)) => { + tracing::error!(%err, "failed to start heap profiling"); + return; + } + Err(err) => { + tracing::error!(%err, "start-profiling task failed"); + return; } } - }) - .await + } + + // Null pointer tells jemalloc to use prof_prefix from MALLOC_CONF. + match blocking(|| unsafe { raw::write(b"prof.dump\0", std::ptr::null::()) }).await { + Ok(Ok(())) => tracing::info!("heap profile dumped"), + Ok(Err(err)) => tracing::error!(%err, "failed to dump heap profile"), + Err(err) => tracing::error!(%err, "heap profile task failed"), + } } #[cfg(test)] @@ -182,15 +196,9 @@ mod tests { super::signal_ready().notified().await; unsafe { libc::raise(libc::SIGUSR1) }; - // Flipping `prof.active` hops to a blocking thread, so poll rather than assume it - // has landed by the next yield. - for _ in 0..200 { - if super::sampling() { - break; - } - kio::clock::sleep(std::time::Duration::from_millis(10)).await; - } - + // Flipping `prof.active` hops to a blocking thread, so wait for the signal to be + // handled rather than assume it has landed by the next yield. + super::signal_handled().notified().await; assert!(super::sampling(), "SIGUSR1 did not start sampling"); assert!(!runner.is_finished()); runner.abort(); @@ -215,7 +223,7 @@ mod tests { let runner = tokio::spawn(super::run()); super::signal_ready().notified().await; unsafe { libc::raise(libc::SIGUSR1) }; - tokio::task::yield_now().await; + super::signal_handled().notified().await; assert!(!runner.is_finished()); runner.abort(); }); diff --git a/rs/moq-tokio/src/mdns.rs b/rs/moq-tokio/src/mdns.rs index 6a36ffec435..a29434b0634 100644 --- a/rs/moq-tokio/src/mdns.rs +++ b/rs/moq-tokio/src/mdns.rs @@ -1199,7 +1199,7 @@ mod tests { /// has interfaces that cannot carry mDNS (a down VPN adapter, a container /// bridge with multicast off). Treating the first error as fatal would refuse /// to start a process whose real NIC was fine. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn announce_on_any_interface_is_enough() { let (tx, rx) = flume::unbounded(); tx.send(mdns_sd::DaemonEvent::Error(mdns_sd::Error::Msg("eth1 is down".into()))) @@ -1336,20 +1336,19 @@ mod tests { stopped(monitor).await; } - /// Two processes under the same app and secret find each other over real mDNS; - /// two under different apps never do, even with the same secret. + /// Two processes under the same app and secret find each other over real mDNS. /// - /// The same-app pair is the positive control that multicast works. Once it - /// has resolved, a foreign-app record would already have arrived if the - /// subtype browse leaked, so a short wait is enough to assert isolation - /// without treating "the packet never came" as success. + /// Different apps never do, even with the same secret, but no wait on the + /// network can prove a record never came: every record is checked against + /// the app it arrived under, and `a_stolen_proof_does_not_transfer` proves a + /// record from another app fails that check. /// /// Ignored because it multicasts on the host network; nightly runs it with /// `just rs multicast`. A host where no interface can announce skips, unless /// CI_STRICT makes that a failure. #[tokio::test] #[ignore = "needs multicast on the host network; run by just rs multicast"] - async fn apps_partition_discovery() { + async fn same_app_peers_find_each_other() { let shared = unique_app(); let mut same_a = match discovery(shared.clone(), Some(KEY_A), "same-a").await { Ok(discovery) => discovery, @@ -1362,9 +1361,6 @@ mod tests { let mut same_b = discovery(shared, Some(KEY_A), "same-b") .await .expect("a second advertiser on a working host"); - let mut foreign = discovery(unique_app(), Some(KEY_A), "foreign") - .await - .expect("a foreign-app advertiser on a working host"); let (on_a, on_b) = tokio::join!(find(&mut same_a, "same-b"), find(&mut same_b, "same-a")); assert!(same_b.verify_credential(&on_a.credential)); @@ -1375,25 +1371,5 @@ mod tests { same_b.should_dial(&on_b.id), "exactly one side of the pair dials" ); - - let leaked = kio::clock::timeout(std::time::Duration::from_secs(2), async { - loop { - match foreign.recv().await { - Some(Event::Found(peer)) - if peer.fingerprint.as_deref() == Some("same-a") - || peer.fingerprint.as_deref() == Some("same-b") => - { - return true; - } - Some(_) => continue, - None => return false, - } - } - }) - .await; - assert!( - !matches!(leaked, Ok(true)), - "a different app must not resolve this application's records" - ); } } diff --git a/rs/moq-tokio/src/quic/driver/tests.rs b/rs/moq-tokio/src/quic/driver/tests.rs index c05c7c72a1f..f2d266d0d25 100755 --- a/rs/moq-tokio/src/quic/driver/tests.rs +++ b/rs/moq-tokio/src/quic/driver/tests.rs @@ -17,7 +17,7 @@ use std::{ }; use crate::quic::driver::runtime::{AsyncUdpSocket, wrap_udp_socket}; -use crate::quic::driver::{Duration, Instant}; +use crate::quic::driver::Duration; use bytes::{Buf, Bytes}; use kio::clock::{sleep, timeout}; use moq_quic::RandomConnectionIdGenerator; @@ -42,14 +42,11 @@ use super::{ recv, send, server, transport, }; -#[test] -fn handshake_timeout() { +/// Paused: nobody answers the dial, so only the handshake timer can end it. +#[tokio::test(start_paused = true)] +async fn handshake_timeout() { let _guard = subscribe(); - let runtime = rt_threaded(); - let client = { - let _guard = runtime.enter(); - client_endpoint(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)).unwrap() - }; + let client = client_endpoint(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)).unwrap(); // Avoid NoRootAnchors error let cert = (|| -> std::result::Result<_, rcgen::Error> { @@ -71,24 +68,22 @@ fn handshake_timeout() { transport_config.initial_rtt = Duration::from_millis(10); client_config.transport = transport_config; - let start = Instant::now(); - runtime.block_on(async move { - match client - .connect( - &client_config, - SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1), - "localhost", - ) - .unwrap() - .await - { - Err(crate::quic::driver::connection::Error::HandshakeTimeout) => {} - Err(e) => panic!("unexpected error: {e:?}"), - Ok(_) => panic!("unexpected success"), - } - }); + let start = kio::clock::now(); + match client + .connect( + &client_config, + SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1), + "localhost", + ) + .unwrap() + .await + { + Err(crate::quic::driver::connection::Error::HandshakeTimeout) => {} + Err(e) => panic!("unexpected error: {e:?}"), + Ok(_) => panic!("unexpected success"), + } let dt = start.elapsed(); - assert!(dt > IDLE_TIMEOUT && dt < 2 * IDLE_TIMEOUT); + assert!(dt >= IDLE_TIMEOUT && dt < 2 * IDLE_TIMEOUT, "timed out after {dt:?}"); } #[tokio::test] @@ -208,7 +203,8 @@ fn read_after_close() { .unwrap() .await .expect("connect"); - sleep(Duration::from_millis(100)).await; + // The peer closes once its stream is acknowledged; read only after that close. + let _ = wait_closed(&new_conn).await; let mut stream = new_conn.accept_uni().await.expect("incoming streams"); let msg = stream.read_to_end(usize::MAX).await.expect("read_to_end"); assert_eq!(msg, MSG); @@ -574,8 +570,14 @@ async fn await_connection( timeout(Duration::from_secs(5), task).await.unwrap().unwrap() } -async fn assert_wait_idle_pending(endpoint: &Endpoint) { - assert!(timeout(Duration::from_millis(50), endpoint.wait_idle()).await.is_err()); +/// Idleness is the endpoint's own state, so one poll reads it. +fn assert_wait_idle_pending(endpoint: &Endpoint) { + let (waker, _) = new_count_waker(); + assert!( + pin!(endpoint.wait_idle()) + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); } async fn wait_idle(endpoint: &Endpoint) { @@ -632,32 +634,34 @@ fn wait_idle_returns_when_the_driver_is_dropped() { }); let _server_connection = accepted.join().expect("accept thread"); - // Park `wait_idle`; it blocks while that connection is live. - let (done_tx, done_rx) = std::sync::mpsc::channel(); - let waiter = { - let handle = endpoint_handle.clone(); - let endpoint = endpoint.clone(); - std::thread::spawn(move || { - handle.block_on(endpoint.wait_idle()); - let _ = done_tx.send(()); - }) - }; - - // Otherwise a pass below would be vacuous. + // Park `wait_idle`; it blocks while that connection is live. Otherwise a pass + // below would be vacuous. + struct Signal(Mutex>); + impl std::task::Wake for Signal { + fn wake(self: Arc) { + let _ = self.0.lock().unwrap().send(()); + } + } + let (woken_tx, woken) = std::sync::mpsc::channel(); + let waker = Waker::from(Arc::new(Signal(Mutex::new(woken_tx)))); + let mut idle = pin!(endpoint.wait_idle()); assert!( - done_rx.recv_timeout(Duration::from_millis(200)).is_err(), + idle.as_mut().poll(&mut Context::from_waker(&waker)).is_pending(), "wait_idle returned before the driver was dropped" ); // Dropping the runtime drops the endpoint driver. endpoint_runtime.shutdown_background(); - let woke = done_rx.recv_timeout(Duration::from_secs(10)).is_ok(); + let woke = woken.recv_timeout(Duration::from_secs(10)).is_ok(); drop(connection); drop(peer); drop(peer_runtime); - assert!(woke, "wait_idle never returned after the endpoint driver was dropped"); - let _ = waiter.join(); + assert!(woke, "wait_idle was never woken after the endpoint driver was dropped"); + assert!( + idle.as_mut().poll(&mut Context::from_waker(&waker)).is_ready(), + "wait_idle still pending after the endpoint driver was dropped" + ); } async fn wait_closed(conn: &crate::quic::driver::connection::Connection) -> crate::quic::driver::connection::Error { @@ -1065,7 +1069,7 @@ async fn split_accept_wait_idle_and_open_connections() { wait_for_blocked_handshake(blocker.clone()).await; assert_eq!(server.open_connections(), 0); - assert_wait_idle_pending(&server).await; + assert_wait_idle_pending(&server); blocker.release(); @@ -1135,7 +1139,7 @@ async fn split_accept_close_during_pending_accept() { wait_for_blocked_handshake(blocker.clone()).await; drop(server.close(0u32.into(), "closing")); - assert_wait_idle_pending(&server).await; + assert_wait_idle_pending(&server); blocker.release(); diff --git a/rs/moq-tokio/tests/integration/broadcast.rs b/rs/moq-tokio/tests/integration/broadcast.rs index 80194a7b08c..6ac37ba7b05 100644 --- a/rs/moq-tokio/tests/integration/broadcast.rs +++ b/rs/moq-tokio/tests/integration/broadcast.rs @@ -2328,10 +2328,12 @@ async fn resubscribe_keeps_flowing_moq_lite_03() { drop(g); drop(sub1); - // Yield a few times so the subscriber task can observe the demand going away - // and send the FIN. A small real sleep also makes the test less - // scheduler-dependent across runtimes. - kio::clock::sleep(Duration::from_millis(20)).await; + // Resubscribe only once the publisher has seen the cancel, so the second + // subscription is a fresh SUBSCRIBE rather than the one being torn down. + kio::clock::timeout(TIMEOUT, crate::unwanted(track.demand())) + .await + .expect("upstream never canceled") + .expect("track open"); let mut sub2 = bc.track("video").unwrap().subscribe(None).await.expect("subscribe2"); @@ -2633,7 +2635,7 @@ async fn one_shot_connect_surfaces_the_session_close() { let mut client_config = moq_tokio::connect::Config::default(); client_config.tls.insecure = Some(true); - // A tiny backoff so a buggy redial happens well within the sleep below. + // A tiny backoff, so a buggy redial lands well before the close times out. client_config.backoff.initial = Some((Duration::from_millis(10)).into()); let client = client_config .init(Default::default()) @@ -2652,13 +2654,10 @@ async fn one_shot_connect_surfaces_the_session_close() { .expect_err("a severed session must surface as an error"); assert!(!connection.connected()); - // Give a buggy reconnect loop ample time to redial before counting accepts. - kio::clock::sleep(Duration::from_millis(200)).await; - assert_eq!( - accepts.load(Ordering::SeqCst), - 1, - "one-shot mode must dial exactly once" - ); + // `closed` resolved, so the loop has stopped: its own count is final, and a + // redial that landed before then would also have reached the server. + assert_eq!(connection.connects(), 1, "one-shot mode must dial exactly once"); + assert_eq!(accepts.load(Ordering::SeqCst), 1, "one-shot mode must dial exactly once"); drop(connection); server_handle.abort(); @@ -2781,8 +2780,7 @@ async fn announce_interest_unauthorized_keeps_session_alive() { let session = request.with_publisher(publish).ok().await?; let _broadcast = broadcast; let _track = track; - let _ = session.closed().await; - Ok::<_, anyhow::Error>(()) + Ok::<_, anyhow::Error>(session.closed().await) }); let client = client.with_subscriber(consume); @@ -2799,19 +2797,20 @@ async fn announce_interest_unauthorized_keeps_session_alive() { assert_eq!(update.prefix.as_str(), "allowed/test"); assert!(active, "expected announce, got retraction"); - // The unauthorized "denied" interest must not have torn down the session. - assert!( - kio::clock::timeout(Duration::from_millis(200), connection.closed()) - .await - .is_err(), - "session closed after unauthorized announce interest", - ); - - drop(connection); - server_handle + // Leave cleanly. Had the unauthorized "denied" interest torn the session down, + // the server would see that failure rather than our clean close. + kio::clock::timeout(TIMEOUT, connection.close()) + .await + .expect("close timed out") + .expect("close failed"); + let reason = server_handle .await .expect("server task panicked") .expect("server task failed"); + assert!( + matches!(reason, moq_net::Error::Cancel), + "session closed after unauthorized announce interest: {reason}" + ); } /// A wildcard grant on both ends of a session. The subscriber asks for the grant's @@ -2895,12 +2894,6 @@ async fn wildcard_scope_test(version: &str, server_scope: &str) { Some(vec!["alice".parse::().unwrap()]) ); assert!(active); - assert!( - kio::clock::timeout(Duration::from_millis(200), next_update(&mut announcements)) - .await - .is_err(), - "a path outside one of the grants was announced" - ); // The granted path serves; the excluded ones are refused before the wire. let bc = kio::clock::timeout(TIMEOUT, sub_consumer.request("room/alice/chat", None)) @@ -2931,7 +2924,20 @@ async fn wildcard_scope_test(version: &str, server_scope: &str) { ); } + // Ending the session retracts alice's chat after every announcement the session + // delivered, so anything outside the grants shows up before that retraction. drop(connection); + loop { + let (update, active) = next_announce(&mut announcements).await; + assert_eq!( + update.prefix.as_str(), + "room/alice/chat", + "a path outside one of the grants was announced" + ); + if !active { + break; + } + } server_handle .await .expect("server task panicked") @@ -3023,14 +3029,9 @@ async fn publish_only_client_to_subscribe_only_server() { .expect("group closed prematurely"); assert_eq!(&frame.payload[..], b"hello"); - // The disjoint "denied" interest must not have torn down the session. - assert!( - kio::clock::timeout(Duration::from_millis(200), session.closed()) - .await - .is_err(), - "server session closed after unauthorized announce interest", - ); - + // A clean close ends a live session; had the disjoint "denied" interest torn + // it down, this returns that failure instead. + session.close().await?; Ok::<_, anyhow::Error>(()) }); @@ -3065,7 +3066,7 @@ async fn publish_only_client_to_subscribe_only_server() { server_handle .await .expect("server task panicked") - .expect("server task failed"); + .expect("server session failed after unauthorized announce interest"); drop(connection); drop(track); diff --git a/rs/moq-tokio/tests/integration/reconnect.rs b/rs/moq-tokio/tests/integration/reconnect.rs index fa1007adc22..6450fe9fccf 100644 --- a/rs/moq-tokio/tests/integration/reconnect.rs +++ b/rs/moq-tokio/tests/integration/reconnect.rs @@ -1,8 +1,7 @@ //! What ends the reconnect loop, and where a peer's GOAWAY can send it. //! -//! Dials over plain TCP (`tcp://`), which fails fast and locally: no TLS material, no QUIC -//! handshake, no server. That keeps the assertion about the *budget* rather than about how long a -//! particular backend takes to give up. +//! Dials over plain TCP (`tcp://`): no TLS material and no QUIC handshake, so the assertions are +//! about the loop rather than about how a particular backend connects. #![cfg(feature = "tcp")] @@ -17,43 +16,6 @@ fn client(backoff: moq_tokio::Backoff) -> moq_tokio::Client { config.init(Default::default()).expect("failed to init client") } -/// A transient failure is retried, escalating, until the budget runs out. The give-up error names -/// the underlying cause so an operator sees why rather than just "timed out". -#[tokio::test] -async fn a_transient_failure_retries_until_the_budget_runs_out() { - let mut backoff = moq_tokio::Backoff::default(); - backoff.initial = Some((Duration::from_millis(20)).into()); - backoff.max = Some((Duration::from_millis(40)).into()); - backoff.timeout = Some((Duration::from_millis(200)).into()); - - // Nothing listens on port 1, so every attempt is refused: transient as far as this layer knows. - let url: url::Url = "tcp://127.0.0.1:1".parse().expect("failed to parse url"); - let started = kio::clock::now(); - let reconnect = client(backoff).connect(url); - - let err = kio::clock::timeout(Duration::from_secs(10), reconnect.closed()) - .await - .expect("reconnect loop never gave up") - .expect_err("reconnect loop stopped without an error"); - - assert!( - matches!(err, moq_tokio::Error::Reconnect(_)), - "stopped with {err} rather than exhausting the budget" - ); - assert_ne!( - err.to_string(), - "reconnect timed out after 200ms", - "give-up error lost the underlying cause" - ); - // The budget is spent on sleeping between attempts, so reaching it takes at least most of it. - // Jitter draws each delay from the top half of its window, hence half rather than the whole. - assert!( - started.elapsed() >= Duration::from_millis(100), - "gave up after {:?} without retrying", - started.elapsed() - ); -} - /// The retry window decides whether a new attempt starts, not how long one may take: a handshake /// slower than the window, but inside the connect timeout, lands on the first dial. Cutting it at /// the window would only fail it, and every redial the same way. @@ -141,7 +103,7 @@ async fn the_connect_count_advances_on_reconnect() { kio::clock::timeout(Duration::from_secs(10), async { while connection.connects() < 2 { - kio::clock::sleep(Duration::from_millis(10)).await; + connection.status().await.expect("status failed"); } }) .await diff --git a/rs/moq-tokio/tests/integration/subscription_end_integrity.rs b/rs/moq-tokio/tests/integration/subscription_end_integrity.rs index 69fc24f3b8d..e6e7268854d 100644 --- a/rs/moq-tokio/tests/integration/subscription_end_integrity.rs +++ b/rs/moq-tokio/tests/integration/subscription_end_integrity.rs @@ -14,9 +14,9 @@ //! cuts the final group's stream mid-flight. //! //! The subscriber drains a head group first (so the subscription is established and -//! only the tail is at stake), then stops reading for [`STALL`] while the publisher -//! writes a 4 MB final group, finishes the track and disconnects, so the group is still -//! queued behind the connection's flow-control window when the session goes away. The +//! only the tail is at stake), then stops reading while the publisher writes a 4 MB +//! final group, finishes the track and disconnects, so the group is still queued +//! behind the connection's flow-control window when the session goes away. The //! subscriber must then either see every frame or end with an error; it must not end //! `Ok(None)` short of the final group. //! @@ -40,9 +40,6 @@ const FRAMES: usize = 10; /// close, the publisher would only ever lose its FIN, and the test would not cover /// a group cut mid-flight. const PADDING: usize = 400_000; -/// How long the subscriber stops reading after the head group, so the final group is -/// still queued in the publisher when the session goes away. -const STALL: Duration = Duration::from_millis(300); /// One round: a head group the subscriber drains, then a final group, `finish()`, and /// the publisher's session dropped with no await in between. Returns how many frames @@ -70,8 +67,11 @@ async fn round(drop_session: bool) -> (usize, Option) { // The publisher only learns of a subscription once the subscriber polls it, so the // reader runs concurrently. It tells the publisher when it has drained the head - // group, so the final group is written against an established subscription. + // group, so the final group is written against an established subscription, and + // stops reading until the publisher is done, so that group is still queued behind + // the flow-control window when the session goes away. let (drained_tx, drained_rx) = tokio::sync::oneshot::channel(); + let (resume_tx, resume_rx) = tokio::sync::oneshot::channel::<()>(); let reader = tokio::spawn(async move { let mut client_config = moq_tokio::connect::Config::default(); @@ -98,7 +98,7 @@ async fn round(drop_session: bool) -> (usize, Option) { let mut sub = remote.track("video").unwrap().subscribe(None).await.expect("subscribe"); let mut got = 0; - let mut drained = Some(drained_tx); + let mut drained = Some((drained_tx, resume_rx)); loop { let mut group = match sub.recv_group().await { Ok(Some(group)) => group, @@ -112,12 +112,10 @@ async fn round(drop_session: bool) -> (usize, Option) { Err(err) => return (got, Some(err)), } } - // The head group is in: from here on, only the tail can be lost. Stop reading - // for a moment, so the publisher's final group is still queued behind the - // connection's flow-control window when it disconnects. - if let Some(tx) = drained.take() { - let _ = tx.send(()); - kio::clock::sleep(STALL).await; + // The head group is in: from here on, only the tail can be lost. + if let Some((drained, resume)) = drained.take() { + let _ = drained.send(()); + let _ = resume.await; } } }); @@ -156,6 +154,7 @@ async fn round(drop_session: bool) -> (usize, Option) { } else { Some(session) }; + let _ = resume_tx.send(()); let result = kio::clock::timeout(TIMEOUT, reader) .await diff --git a/rs/moq-uring/src/worker.rs b/rs/moq-uring/src/worker.rs index 3a934c09f00..bcb8dbdec40 100644 --- a/rs/moq-uring/src/worker.rs +++ b/rs/moq-uring/src/worker.rs @@ -1201,7 +1201,7 @@ mod tests { #[test] fn spawned_tasks_run() { - let Some(mut worker) = worker() else { return }; + let Some((mut worker, clock)) = manual() else { return }; let handle = worker.handle(); let flag = Rc::new(std::cell::Cell::new(0)); @@ -1211,13 +1211,22 @@ mod tests { flag.set(flag.get() + index + 1); }); } - // Spawned tasks run even while the main future pends on a timer. + // Spawned tasks run even while the main future pends on a timer, which only + // fires once the worker has parked on it and the clock is advanced. + let parks = handle.metrics(); + let advancer = std::thread::spawn(move || { + while parks.snapshot().parks == 0 { + std::hint::spin_loop(); + } + clock.advance(Duration::from_millis(10)); + }); let handle2 = handle.clone(); worker .block_on(async move { Deadline::after(&handle2, Duration::from_millis(10)).wait().await; }) .unwrap(); + advancer.join().unwrap(); assert_eq!(flag.get(), 6); } diff --git a/rs/moq-uring/tests/integration/socket_queues.rs b/rs/moq-uring/tests/integration/socket_queues.rs index 073e776002e..87b703086a3 100644 --- a/rs/moq-uring/tests/integration/socket_queues.rs +++ b/rs/moq-uring/tests/integration/socket_queues.rs @@ -403,6 +403,16 @@ fn drops(sockets: &[UdpSocket]) -> io::Result> { .collect() } +/// Whether any member still has datagrams waiting in its queue. +fn queued(sockets: &[UdpSocket]) -> io::Result { + for socket in sockets { + if meminfo(socket.as_raw_fd())?[SK_MEMINFO_RMEM_ALLOC] != 0 { + return Ok(true); + } + } + Ok(false) +} + /// Datagrams the host dropped before any socket saw them, because a CPU's /// input backlog (`net.core.netdev_max_backlog`) was full: the second column /// of `/proc/net/softnet_stat`, summed over CPUs. Loopback delivers through @@ -762,18 +772,16 @@ fn measure(path: Path, cell: Cell) -> anyhow::Result { let snmp_before = snmp_rcvbuf_errors(); let (sent, send_dropped) = send(cell, target)?; - // Drain: wait out the path's queue, then until nothing more arrives. + // Drain: wait out the path's queue, which the shaper releases on the host clock, + // then until every member's queue is empty. The workers have read everything + // then, so stopping them below accounts for every datagram they took. if cell.pattern == Pattern::Shaped { std::thread::sleep(PATH_QUEUE * 2); } - let mut last = u64::MAX; - loop { - std::thread::sleep(Duration::from_millis(100) + cell.stall); - let now = delivered.load(Ordering::Relaxed); - if now == last { - break; - } - last = now; + let deadline = now_ns() + Duration::from_secs(60).as_nanos() as u64; + while queued(&monitors)? { + anyhow::ensure!(now_ns() < deadline, "the workers never drained their queues"); + std::thread::sleep(Duration::from_millis(1)); } sampling.store(false, Ordering::Relaxed); let peak_queue = sampler.join().expect("sampler"); @@ -800,7 +808,6 @@ fn measure(path: Path, cell: Cell) -> anyhow::Result { rcvbuf, sent, send_dropped, - delivered: delivered.load(Ordering::Relaxed), socket_drops, backlog_drops: backlog, path_drops, @@ -821,6 +828,8 @@ fn measure(path: Path, cell: Cell) -> anyhow::Result { result.rx_cpu += worker.cpu; result.enobufs += worker.enobufs; } + // Read once every worker has stopped, so a batch still being processed counts. + result.delivered = delivered.load(Ordering::Relaxed); contention.store(true, Ordering::Relaxed); for spinner in spinners { spinner.join().expect("spinner"); From c2de4b21d7eb8fa00aed56c613c446e2788c4db0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 14:38:05 -0800 Subject: [PATCH 3/5] test: format Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/src/quic/driver/tests.rs | 2 +- rs/moq-tokio/tests/integration/broadcast.rs | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/rs/moq-tokio/src/quic/driver/tests.rs b/rs/moq-tokio/src/quic/driver/tests.rs index f2d266d0d25..bf107022d8b 100755 --- a/rs/moq-tokio/src/quic/driver/tests.rs +++ b/rs/moq-tokio/src/quic/driver/tests.rs @@ -16,8 +16,8 @@ use std::{ task::{Context, Poll, RawWaker, RawWakerVTable, Waker}, }; -use crate::quic::driver::runtime::{AsyncUdpSocket, wrap_udp_socket}; use crate::quic::driver::Duration; +use crate::quic::driver::runtime::{AsyncUdpSocket, wrap_udp_socket}; use bytes::{Buf, Bytes}; use kio::clock::{sleep, timeout}; use moq_quic::RandomConnectionIdGenerator; diff --git a/rs/moq-tokio/tests/integration/broadcast.rs b/rs/moq-tokio/tests/integration/broadcast.rs index 6ac37ba7b05..5508bbfe111 100644 --- a/rs/moq-tokio/tests/integration/broadcast.rs +++ b/rs/moq-tokio/tests/integration/broadcast.rs @@ -2657,7 +2657,11 @@ async fn one_shot_connect_surfaces_the_session_close() { // `closed` resolved, so the loop has stopped: its own count is final, and a // redial that landed before then would also have reached the server. assert_eq!(connection.connects(), 1, "one-shot mode must dial exactly once"); - assert_eq!(accepts.load(Ordering::SeqCst), 1, "one-shot mode must dial exactly once"); + assert_eq!( + accepts.load(Ordering::SeqCst), + 1, + "one-shot mode must dial exactly once" + ); drop(connection); server_handle.abort(); From 412e3ef7b27a6b437826f2a4ba083b212f9c5360 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 14:45:46 -0800 Subject: [PATCH 4/5] quest: finish the transport crates' mocked-clock tests Co-Authored-By: Claude Opus 5.5 --- quest/m1/mock-clock-tests/README.md | 1 - quest/m1/mock-clock-tests/rs-transport.md | 24 ----------------------- 2 files changed, 25 deletions(-) delete mode 100644 quest/m1/mock-clock-tests/rs-transport.md diff --git a/quest/m1/mock-clock-tests/README.md b/quest/m1/mock-clock-tests/README.md index f1df1479e8a..f2998d1f7f2 100644 --- a/quest/m1/mock-clock-tests/README.md +++ b/quest/m1/mock-clock-tests/README.md @@ -41,7 +41,6 @@ before starting, since tests land and move. - [js/net tests run on a mocked clock](/quest/m1/mock-clock-tests/js-net.md) - about 120 js/net tests stop waiting on real time, through their shared helpers - [Other JS tests run on a mocked clock](/quest/m1/mock-clock-tests/js-rest.md) - hang's `settle(20)`, publish's fanout helpers, and the rest of JS - [moq-relay tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-relay.md) - the relay's `timeout(..).is_err()` session proofs and settle sleeps -- [Transport crate tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-transport.md) - moq-tokio, moq-quic, moq-sock, and moq-uring - [Media crate tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-media.md) - moq-audio, moq-video, moq-hls, moq-mux, and the bridges - [Binding and CLI tests run on a mocked clock](/quest/m1/mock-clock-tests/rs-bindings.md) - moq-ffi, moq-c, moq-cli, and kio - [Go and Python tests run on a mocked clock](/quest/m1/mock-clock-tests/go-py.md) - the Go wrapper's and Python packages' real-time waits diff --git a/quest/m1/mock-clock-tests/rs-transport.md b/quest/m1/mock-clock-tests/rs-transport.md deleted file mode 100644 index 52024340e54..00000000000 --- a/quest/m1/mock-clock-tests/rs-transport.md +++ /dev/null @@ -1,24 +0,0 @@ -# [M] Transport crate tests run on a mocked clock - -## Goal - -The transport crates' tests that wait on real time move to a paused or -simulated clock, or an awaited event. See the -[epic](/quest/m1/mock-clock-tests/README.md) for the rules. - -## Plan - -As of 2026-11-06: - -- moq-tokio, 24: `tests/integration/broadcast.rs` 9, `src/mdns.rs` 4, - `tests/integration/reconnect.rs` 3, `src/quic/driver/tests.rs` 2, - `tests/integration/subscription_end_integrity.rs` 2, and one each in - `src/jemalloc.rs`, `src/server.rs`, `src/tcp.rs`, and - `tests/integration/backend.rs`. -- moq-quic, 6: `src/connection/mtud.rs` 4 and - `src/tests/characterize/endpoint.rs` 2. -- moq-sock, 4: `tests/udp/main.rs` 3 and `src/shard.rs`. -- moq-uring, 3: `src/worker.rs` 2 and - `tests/integration/socket_queues.rs`. - -Public API: none. Wire: none. From 73dd59d8832bd8ecfecb099807efe2e288845c60 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 14:50:51 -0800 Subject: [PATCH 5/5] test(tokio): wait on monitor revisions for the connect count Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/tests/integration/reconnect.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/rs/moq-tokio/tests/integration/reconnect.rs b/rs/moq-tokio/tests/integration/reconnect.rs index 6450fe9fccf..3487d794152 100644 --- a/rs/moq-tokio/tests/integration/reconnect.rs +++ b/rs/moq-tokio/tests/integration/reconnect.rs @@ -101,9 +101,13 @@ async fn the_connect_count_advances_on_reconnect() { let session = sessions.recv().await.expect("server stopped accepting"); drop(session); + // Revisions, not `status()`: a drop and redial between polls would leave the status + // reading `Connected`, unchanged, with nothing left to wake it. + let monitor = connection.monitor(); kio::clock::timeout(Duration::from_secs(10), async { - while connection.connects() < 2 { - connection.status().await.expect("status failed"); + let mut state = monitor.state(); + while state.connects < 2 { + state = monitor.state_changed(state.revision).await.expect("monitor stopped"); } }) .await