diff --git a/cpp/obs/test/moq-dock-stop-test.cpp b/cpp/obs/test/moq-dock-stop-test.cpp index 15ecabeb164..cd02ca3aba0 100644 --- a/cpp/obs/test/moq-dock-stop-test.cpp +++ b/cpp/obs/test/moq-dock-stop-test.cpp @@ -6,9 +6,9 @@ #include "moq-dock-stop.h" -#include #include #include +#include #include #define CHECK(cond) \ @@ -36,20 +36,25 @@ int main() { MoQDockStopBridge bridge; CHECK(bridge.begin()); - std::atomic ended{false}; + // Ends only once the close began, so the close returning with nothing active + // means it waited for this end. The worker's deadline only bounds a lost close + // notify, after which it ends anyway, so it can never fail the test. The close's + // deadline is one the test never reaches, so only the end releases it. std::thread worker([&] { - std::this_thread::sleep_for(std::chrono::milliseconds(30)); + { + std::unique_lock lock(bridge.mutex); + bridge.cv.wait_for(lock, std::chrono::seconds(10), [&] { return bridge.closing; }); + } bridge.end(); - ended.store(true); }); - const auto start = std::chrono::steady_clock::now(); - bridge.closeAndWait(std::chrono::milliseconds(500)); - const auto elapsed = std::chrono::steady_clock::now() - start; + bridge.closeAndWait(std::chrono::hours(1)); + bool idle; + { + std::lock_guard lock(bridge.mutex); + idle = bridge.active == 0; + } worker.join(); - CHECK(ended.load()); - CHECK(bridge.active == 0); - CHECK(elapsed >= std::chrono::milliseconds(20)); - CHECK(elapsed < std::chrono::milliseconds(400)); + CHECK(idle); CHECK(!bridge.begin()); } printf("close waits for in-flight end: ok\n"); diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index b8607c2a594..dc0315d9634 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -427,6 +427,9 @@ struct MoQOutputStatsTest { stale.bytes_sent = 123; return output.CommitConnectionStats(session, std::move(stale), accepted); } + // Runs the continuations already queued, the last that can report once the attempt + // is retired. + static void Drain(MoQOutput &output) { ::Drain(output.worker); } static void Disconnect(MoQOutput &output) { std::lock_guard lock(output.mutex); @@ -956,7 +959,8 @@ int main() o.CopyLastFailure(&code, &reason); CHECK(!reason.empty()); CHECK(!o.IsLiveSession()); - std::this_thread::sleep_for(std::chrono::milliseconds(50)); + // The failure retired the attempt, so nothing it queued can report again. + MoQOutputStatsTest::Drain(o); CHECK(signalCount() == 1); } printf("fatal before connect: ok\n"); @@ -1036,7 +1040,8 @@ int main() MoQOutput o(nullptr, OUTPUT); CHECK(o.Start()); o.Stop(); - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + // The stop dropped the connect, so only a result it already queued could still run. + MoQOutputStatsTest::Drain(o); CHECK(signalCount() == 1); CHECK(signalAt(0).code == OBS_OUTPUT_SUCCESS); } @@ -1085,7 +1090,7 @@ int main() CHECK(WaitFor([] { return signalCount() == 2; })); CHECK(signalAt(0).begin_capture_at == 1); CHECK(signalAt(1).begin_capture_at == 2); - std::this_thread::sleep_for(std::chrono::milliseconds(50)); + MoQOutputStatsTest::Drain(o); CHECK(signalCount() == 2); } printf("OBS-driven restart: ok\n"); @@ -1128,7 +1133,8 @@ int main() const size_t starts = round % 2 ? 2 : 1; CHECK(signalCount() <= starts + 1); } - std::this_thread::sleep_for(std::chrono::milliseconds(50)); + // A late report could only come from a continuation, and the destructor joined the + // worker that runs them. CHECK(g_signals_after_destroy == 0); } printf("failures outliving the output: ok\n"); diff --git a/cpp/obs/test/moq-source-test.cpp b/cpp/obs/test/moq-source-test.cpp index 00680f21af5..98b7943c2b2 100644 --- a/cpp/obs/test/moq-source-test.cpp +++ b/cpp/obs/test/moq-source-test.cpp @@ -1803,10 +1803,15 @@ int main() const int blanks = g_blank_calls; g_info.update(source, settingsData()); CHECK(g_blank_calls == blanks + 1); + auto *ctx = static_cast(source); + { + // No connection is left to subscribe with, so later frames have nowhere to land. + std::lock_guard lock(ctx->mutex); + CHECK(!ctx->connection); + } + // A frame the dropped connection already queued finds it gone and outputs nothing. const int frames = g_output_frames; - for (int i = 0; i < 10; i++) - publisher.Write(); - std::this_thread::sleep_for(std::chrono::milliseconds(50)); + Drain(ctx->worker); CHECK(g_output_frames == frames); destroySource(source); } @@ -1903,11 +1908,25 @@ int main() TestRelay relay; reset(relay.Url()); void *source = createSource(); + auto *ctx = static_cast(source); + std::promise offline; { Publisher publisher(relay, "obs/test"); CHECK(publisher.WriteUntil([] { return g_output_frames > 0; })); + // The source waits for the next instance, and re-arms that wait once it sees + // this one go offline, replacing the call it holds now. Observe the replacement, + // keeping the real cancellation. + std::lock_guard lock(ctx->mutex); + auto &call = ctx->connection->available_call; + CHECK(call.has_value()); + auto pending = std::make_shared(std::move(*call)); + call.emplace([pending, &offline] { + pending->cancel(); + offline.set_value(); + }); } - std::this_thread::sleep_for(std::chrono::milliseconds(50)); + CHECK(offline.get_future().wait_for(std::chrono::seconds(10)) == std::future_status::ready); + // On a lost re-arm, the destroy runs the replacement, so `offline` outlives it. destroySource(source); } report("broadcast ends"); diff --git a/cpp/obs/test/moq-test-relay.h b/cpp/obs/test/moq-test-relay.h index 6f1c325d4bc..ba5b206f1a0 100644 --- a/cpp/obs/test/moq-test-relay.h +++ b/cpp/obs/test/moq-test-relay.h @@ -7,6 +7,8 @@ #include +#include "moq-worker.h" + #include #include #include @@ -17,6 +19,7 @@ #include #include #include +#include #include #include #include @@ -54,6 +57,22 @@ inline bool WaitFor(const std::function &done, std::chrono::milliseconds return true; } +// Runs whatever `worker` already queued, so a test checks what an object's dropped calls +// did without waiting out a window. Dropping a moq::Continuation does not recall a result +// whose delivery already began, which can still queue after this returns: a check after +// it can miss a stale continuation, never fail falsely. +inline void Drain(const MoQWorker &worker) +{ + auto drained = std::make_shared>(); + // A stopped worker runs nothing more, which is just as settled. + if (!worker.Executor()([drained] { drained->set_value(); })) + return; + if (drained->get_future().wait_for(std::chrono::seconds(10)) != std::future_status::ready) { + std::fprintf(stderr, "FAIL: the worker never reached the drain\n"); + std::abort(); + } +} + // A moq-ffi server that publishes every broadcast it consumes, so whatever one // session announces, another can subscribe to. // diff --git a/cpp/obs/test/obs-moq-test.cpp b/cpp/obs/test/obs-moq-test.cpp index 73d870dcfc3..b6f3278b742 100644 --- a/cpp/obs/test/obs-moq-test.cpp +++ b/cpp/obs/test/obs-moq-test.cpp @@ -52,7 +52,8 @@ int main() auto server = TestOk(moq::Server::init(config), "server init"); TestOk(server->listen().get(), "listen"); auto pending = server->accept(); - if (pending.wait_for(std::chrono::milliseconds(50)) != std::future_status::timeout) { + // Nothing dials, so only the unload below can end it; read without waiting. + if (pending.wait_for(std::chrono::seconds(0)) != std::future_status::timeout) { fprintf(stderr, "FAIL: accept resolved with no peer\n"); return 1; } diff --git a/quest/m1/README.md b/quest/m1/README.md index df79db13c63..0c21b0beaa8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -41,7 +41,6 @@ moves what remains of them once their claims merge. - [Audio jitter target](/quest/m1/audio-jitter-target/README.md) - the playout target is default in both languages; time-stretched browser playback without steady-state underruns and a native trace replay remain - [Buffered playback subscribes in ascending order](/quest/m1/buffered-ascending.md) - a buffered `` asks for the oldest group first, so a voice reply's opening arrives first, and audio's head hold shrinks to the floor - [Push-to-talk's press-to-audible latency fits a budget on the defaults](/quest/m1/push-to-talk-latency.md) - press to audible on the browser defaults, excluding device output, stays under a kixtop-measured budget that `just test push-to-talk` enforces -- [Tests run on a mocked clock](/quest/m1/mock-clock-tests/README.md) - about 300 tests stop proving nothing happened by waiting, sleeping to settle, or asserting on elapsed time - [Lifecycles are explicit state](/quest/m1/explicit-state/README.md) - every lifecycle the 2026-11-07 audit found implicit is one enum or union, fixing the bugs in the 11 lifecycles it redesigns, with review and agent rules against new ones - [Track drop runs are forgotten below the floor](/quest/m1/drop-runs-prune.md) - a track's dropped-group record stays proportional to its window in Rust and JS - [A macOS camera delivers frames the renderer imports](/quest/m1/camera-nv12-macos.md) - AVFoundation capture asks for NV12, so a camera whose native format is `2vuy` reaches the renderer diff --git a/quest/m1/mock-clock-tests/README.md b/quest/m1/mock-clock-tests/README.md deleted file mode 100644 index 5ae00211c7e..00000000000 --- a/quest/m1/mock-clock-tests/README.md +++ /dev/null @@ -1,63 +0,0 @@ -# [L] Tests run on a mocked clock - -## Goal - -No test proves that nothing happened by waiting, sleeps to let things -settle, or asserts on elapsed real time. Each one runs on a paused, -simulated, or `Manual` clock, or awaits the event it means. Tests that only -wait for progress under a generous timeout, and tests on real sockets, -devices, or processes, stay on the real clock. - -## Plan - -Decided 2026-11-06 (maintainer, /quest-complete, from an audit of every -test's clock): - -- ✅ **Scope: negative proofs, settle sleeps, and elapsed-time - assertions.** Not chosen: only the first two; also the ~530 tests that - only wait for progress under a timeout, which flake little and would cost - a much larger migration. -- ✅ m1, while the guards that stop new ones (the runtime guards and the - `REVIEW.md` rule) landed in m0. - -Rules the remaining work keeps: - -- Fix the shared helper first, then its callers. A helper that resolves - `undefined` when a timeout runs out becomes one that advances a mocked - clock past the deadline, or awaits the event. -- Never replace one real wait with a shorter or longer one, or with a - quiet window. -- Fix a test and its siblings in other languages together. -- An opt-out with a trailing reason comment (`hostTime(); // a real HTTP - server`, `kio::clock::host_time(); // real SRT sockets`) is real I/O and - stays. A bare one marks a test still to convert: remove it with the - conversion, so the `bun test` guard (`js/clock/src/preload.ts`) or - `just rs test`'s host-timer guard covers the test. A test over a real QUIC - endpoint or ICE agent needs none, since those transports allow host timers - themselves. - -What remains, from a scan on 2026-11-07; re-scan before starting, since -tests land and move: - -- **moq-relay's TCP-only tests**, the seven bare `kio::clock::host_time()` - calls in `tests/integration/auth_lifetime.rs` (4), `session_revalidate.rs` - (2), and `socket_auth.rs` (1). They wait out real re-check cadences and the - handshake deadline over TCP, bound a close by a real budget ("well inside - the cadence"), and poll listings on kio sleeps. The relay runs on the test - runtime, so an installed `kio::clock::Manual` can own its timers while the - sockets stay real: a push then proves its close with the clock standing - still, and a cadence passes by `advance`. `wait_listed` and `assert_closed` - need awaited events first. `socket_auth`'s test waits only for progress, - so find which relay timer fires there. -- **moq-ffi's `video_encoder_follows_a_shrinking_grant`**, which polls - `applied_bitrate()` every 10 ms while the follow thread retunes the - encoder. The applied target lives in `VideoProducer` under the rendition's - one lock, so the test needs an event from the follower, not a second - channel beside that lock. -- **The OBS plugin tests** in `cpp/obs/test`, which neither runtime guard - sees. `moq-output-test.cpp` and `moq-source-test.cpp` sleep 50 to 100 ms, - then check that no further signal or frame arrived; `obs-moq-test.cpp` - proves a parked `accept` by waiting 50 ms; `moq-dock-stop-test.cpp` - asserts `20ms <= elapsed < 400ms` around a worker that sleeps 30 ms. Each - needs an event from the plugin, such as its worker joining or the bridge - observing the close, instead of a window. diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index a5c9b68ac3c..0fe67768045 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -7517,7 +7517,6 @@ async fn cancelled_state_wait_does_not_swallow_the_next_transition() { #[cfg(feature = "video")] #[tokio::test] async fn video_encoder_follows_a_shrinking_grant() { - kio::clock::host_time(); use crate::bandwidth::MoqAllocator; use crate::video::*; @@ -7568,18 +7567,8 @@ async fn video_encoder_follows_a_shrinking_grant() { .unwrap(); assert_eq!(reservation.rate(), Some(1_000_000)); - let deadline = kio::clock::Instant::now() + TIMEOUT; - loop { - if video.applied_bitrate() == 1_000_000 { - break; - } - assert!( - kio::clock::Instant::now() < deadline, - "encoder did not follow the shrinking grant, last applied {}", - video.applied_bitrate() - ); - kio::clock::sleep(Duration::from_millis(10)).await; - } + // A drop applies at once, so this waits on the follow thread alone, never a timer. + video.wait_applied(1_000_000); video.finish().unwrap(); broadcast.close().unwrap(); diff --git a/rs/moq-ffi/src/video.rs b/rs/moq-ffi/src/video.rs index 52db4e462f0..8fcee995ecf 100644 --- a/rs/moq-ffi/src/video.rs +++ b/rs/moq-ffi/src/video.rs @@ -360,6 +360,10 @@ struct VideoProducer { /// an allocator. Beside the encoder, so a move to a new track, a finish, and an abort /// change all three together. follow: Option, + /// Notified under this rendition's lock each time the follower retunes, so a test + /// awaits the follower instead of polling `target`. + #[cfg(test)] + retuned: Arc, } /// A rendition's bandwidth claim on its current track, and the thread retuning the encoder @@ -610,13 +614,29 @@ impl MoqVideoProducer { /// The bitrate the encoder produces at, or 0 while disabled. #[cfg(test)] pub(crate) fn applied_bitrate(&self) -> u64 { - match self.inner.lock().unwrap().as_ref().map(|producer| producer.target) { - Some(moq_mux::rate::Target::Enabled(bitrate)) => bitrate.as_bps(), - _ => 0, + applied(self.inner.lock().unwrap().as_ref()) + } + + /// Block until the follower has retuned the encoder to `bitrate`. + #[cfg(test)] + pub(crate) fn wait_applied(&self, bitrate: u64) { + let mut guard = self.inner.lock().unwrap(); + while applied(guard.as_ref()) != bitrate { + let retuned = guard.as_ref().expect("the rendition is open").retuned.clone(); + guard = retuned.wait(guard).unwrap(); } } } +/// The bitrate `producer` encodes at, or 0 while disabled or gone. +#[cfg(test)] +fn applied(producer: Option<&VideoProducer>) -> u64 { + match producer.map(|producer| producer.target) { + Some(moq_mux::rate::Target::Enabled(bitrate)) => bitrate.as_bps(), + _ => 0, + } +} + // Dropped or closed before its finish, the track ends the way a publisher leaving does. impl crate::producer::Child for MoqVideoProducer { fn abort(&self, err: moq_net::Error) { @@ -855,6 +875,8 @@ impl MoqVideoProducer { priority, target: moq_mux::rate::Target::Enabled(ceiling), follow: None, + #[cfg(test)] + retuned: Arc::default(), }))), children: children.clone(), }; @@ -1000,6 +1022,8 @@ async fn follow_reservation(followed: Followed) { let policy = video_policy(*producer.ceiling.read(), below); let before = producer.producer.demand(); let retuned = retune(&mut follower, policy, target, |target| producer.apply(target)); + #[cfg(test)] + producer.retuned.notify_all(); // A disable drains the encoder into the track, and output with a new decoder identity // moves the rendition, so the claim moves with it. The replaced follower is this // thread, which ends here, so it is detached rather than joined. @@ -1662,6 +1686,7 @@ mod finish_tests { ceiling: kio::Producer::new(moq_net::bandwidth::Rate::from_mbps(1)), priority: hang::catalog::PRIORITY.video, follow: None, + retuned: Arc::default(), }; let (video, disabled) = tokio::task::spawn_blocking(move || { video diff --git a/rs/moq-relay/tests/integration/auth_lifetime.rs b/rs/moq-relay/tests/integration/auth_lifetime.rs index b4b4e480811..a164e2df093 100644 --- a/rs/moq-relay/tests/integration/auth_lifetime.rs +++ b/rs/moq-relay/tests/integration/auth_lifetime.rs @@ -32,6 +32,8 @@ fn timed() -> moq_net::track::Info { moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI) } +/// Bounds a wait on the host clock. Under `crate::manual()` it never fires, and nextest's +/// `terminate-after` bounds the test instead. const TIMEOUT: Duration = Duration::from_secs(10); /// What the scripted server answers next, per event. @@ -175,11 +177,14 @@ fn all() -> Patterns { [Pattern::all()].into_iter().collect() } -/// Everything under the dialed path, re-checked every second, good for `expires_in`. +/// How often [`grant`] asks to be re-checked. +const CADENCE: Duration = Duration::from_secs(1); + +/// Everything under the dialed path, re-checked every [`CADENCE`], good for `expires_in`. fn grant(expires_in: Duration) -> Grant { let mut grant = Grant::new(all(), all()); grant.expires = Some(kio::clock::wall() + expires_in); - grant.revalidate = Some(Duration::from_secs(1)); + grant.revalidate = Some(CADENCE); grant } @@ -425,8 +430,8 @@ async fn spawn_quic_relay_with( (addr, handle) } -async fn assert_closed(session: moq_tokio::Connection, within: Duration, what: &str) { - let _ = kio::clock::timeout(within, session.closed()) +async fn assert_closed(session: moq_tokio::Connection, what: &str) { + let _ = kio::clock::timeout(TIMEOUT, session.closed()) .await .unwrap_or_else(|_| panic!("relay should close the {what} session")); } @@ -656,7 +661,7 @@ async fn refusals_and_outages_refuse_at_connect() { /// ones, and media sent afterwards records there in both directions. #[tokio::test] async fn a_moved_tier_retags_the_live_session() { - kio::clock::host_time(); + let clock = crate::manual(); let mut labeled = grant(Duration::from_secs(3600)); labeled.label = Some("project/key/1".into()); let script = Script::new(labeled); @@ -780,10 +785,10 @@ async fn a_moved_tier_retags_the_live_session() { moved_grant.label = Some("project/key/2".into()); script.on_revalidate(Answer::Grant(moved_grant)); - let deadline = kio::clock::now() + TIMEOUT; + clock.advance(CADENCE); + // The stats signal no retag, so yield until both re-checks land; the clock stands still. while active(&moved) < 2 { - assert!(kio::clock::now() < deadline, "the re-checked tier never applied"); - kio::clock::sleep(Duration::from_millis(50)).await; + tokio::task::yield_now().await; } assert_eq!(active(&stats::Tier::default()), 0, "presence left the old tier"); @@ -824,8 +829,8 @@ async fn a_moved_tier_retags_the_live_session() { #[cfg(feature = "websocket")] #[tokio::test] async fn a_rechecked_grant_resizes_live_sessions() { - // Each scheme waits out two re-check cadences, so run them side by side. - futures::future::join_all(["tcp", "ws"].map(|scheme| async move { + let clock = crate::manual(); + for scheme in ["tcp", "ws"] { let script = Script::new(grant(Duration::from_secs(3600))); let auth = build_auth(script.spawn().await); let (port, relay) = match scheme { @@ -910,6 +915,7 @@ async fn a_rechecked_grant_resizes_live_sessions() { let mut narrow = grant(Duration::from_secs(3600)); narrow.subscribe = ["alice/video/**".parse().unwrap()].into_iter().collect(); script.on_revalidate(Answer::Grant(narrow)); + clock.advance(CADENCE); let err = loop { match kio::clock::timeout(TIMEOUT, subs[0].recv_group()) @@ -939,6 +945,7 @@ async fn a_rechecked_grant_resizes_live_sessions() { // Undeafen: the path is announced again and a new subscription flows. script.on_revalidate(Answer::Grant(grant(Duration::from_secs(3600)))); + clock.advance(CADENCE); let broadcast = kio::clock::timeout(TIMEOUT, sub_consumer.follow("alice/audio", None).unwrap().available()) .await .expect("the undeafened path was never announced again") @@ -965,15 +972,14 @@ async fn a_rechecked_grant_resizes_live_sessions() { assert_open_until_closed(&script, [pub_session, sub_session], &format!("{scheme}: a resize")).await; relay.abort(); - })) - .await; + } } /// A re-check that moves the root still closes the session: nothing it holds is /// named the same way under the new one. #[tokio::test] async fn a_moved_root_closes_live_sessions() { - kio::clock::host_time(); + let clock = crate::manual(); let script = Script::new(grant(Duration::from_secs(3600))); let (port, relay) = spawn_relay(build_auth(script.spawn().await)).await; let (pub_session, sub_session) = connect_and_round_trip(&room_url("tcp", port)).await; @@ -982,8 +988,9 @@ async fn a_moved_root_closes_live_sessions() { moved.root = Some("elsewhere".into()); script.on_revalidate(Answer::Grant(moved)); - assert_closed(pub_session, Duration::from_secs(5), "publisher").await; - assert_closed(sub_session, Duration::from_secs(5), "subscriber").await; + clock.advance(CADENCE); + assert_closed(pub_session, "publisher").await; + assert_closed(sub_session, "subscriber").await; relay.abort(); } @@ -991,22 +998,22 @@ async fn a_moved_root_closes_live_sessions() { /// 403, 401, and an empty grant are all a no. #[tokio::test] async fn a_refusal_closes_live_sessions() { - kio::clock::host_time(); - // Each answer waits out a re-check cadence, so run them side by side. + let clock = crate::manual(); let answers = [ ("403", Answer::Status(403)), ("401", Answer::Status(401)), ("empty grant", Answer::Grant(Grant::default())), ]; - futures::future::join_all(answers.map(|(label, answer)| async move { + for (label, answer) in answers { let script = Script::new(grant(Duration::from_secs(3600))); let (port, relay) = spawn_relay(build_auth(script.spawn().await)).await; let (pub_session, sub_session) = connect_and_round_trip(&room_url("tcp", port)).await; script.on_revalidate(answer); - assert_closed(pub_session, Duration::from_secs(5), &format!("{label} publisher")).await; - assert_closed(sub_session, Duration::from_secs(5), &format!("{label} subscriber")).await; + clock.advance(CADENCE); + assert_closed(pub_session, &format!("{label} publisher")).await; + assert_closed(sub_session, &format!("{label} subscriber")).await; let ends = script.ends_at_least(2, &format!("{label}: an end per session")).await; for end in &ends { @@ -1017,8 +1024,7 @@ async fn a_refusal_closes_live_sessions() { } relay.abort(); - })) - .await; + } } /// The one-shot HTTP routes are sessions of their own: `/announced` is admitted @@ -1028,6 +1034,7 @@ async fn a_refusal_closes_live_sessions() { #[cfg(feature = "websocket")] #[tokio::test] async fn http_routes_hold_a_lease() { + let clock = crate::manual(); let script = Script::new(grant(Duration::from_secs(3600))); let (port, relay) = spawn_ws_relay(build_auth(script.spawn().await)).await; @@ -1137,7 +1144,8 @@ async fn http_routes_hold_a_lease() { // A refusal on the fetch's re-check cuts the transfer; the publisher is untouched. script.on_revalidate_http(Answer::Status(403)); - let cut = kio::clock::timeout(Duration::from_secs(5), async { + clock.advance(CADENCE); + let cut = kio::clock::timeout(TIMEOUT, async { loop { match fetch.chunk().await { Ok(Some(_)) => continue, @@ -1198,7 +1206,7 @@ async fn the_end_carries_duration_and_bytes() { /// reporting its end, and holds both registrations until those reports finish. #[tokio::test] async fn the_drain_waits_for_a_closed_session_to_report_its_end() { - let clock = manual(); + let clock = crate::manual(); let script = Script::new(grant(Duration::from_secs(3600))); let gate = script.hold_ends(); let auth = build_auth(script.spawn().await); @@ -1539,7 +1547,7 @@ async fn an_embedded_decider_admits_and_revokes() { // The publisher was admitted first; revoking its lease closes it alone. let publisher = decider.producers.lock().unwrap().remove(0); publisher.revoke(moq_auth::lease::Reason::Refused); - assert_closed(pub_session, Duration::from_secs(5), "publisher").await; + assert_closed(pub_session, "publisher").await; // The subscriber's lease reports the client's own close, so the revocation had // not ended it. let subscriber = decider.producers.lock().unwrap().remove(0); @@ -1575,7 +1583,7 @@ async fn an_embedded_decider_admits_and_revokes() { /// `supervise`, as `a_rechecked_grant_resizes_live_sessions` proves. #[tokio::test(start_paused = true)] async fn a_fixed_lease_still_expires() { - let clock = manual(); + let clock = crate::manual(); closes_at_expires(&clock, moq_auth::lease::Consumer::fixed(expiring())).await; } @@ -1627,7 +1635,7 @@ async fn memory_sessions() -> (moq_net::Session, moq_net::Session, kio::Producer /// server's outage keeps the grant is `moq_auth::Client`'s to test. #[tokio::test(start_paused = true)] async fn an_outage_keeps_the_session_until_expires() { - let clock = manual(); + let clock = crate::manual(); let (producer, consumer) = moq_auth::lease::Producer::new(expiring()); closes_at_expires(&clock, consumer).await; let (reason, _) = producer.closed().await; @@ -1637,12 +1645,6 @@ async fn an_outage_keeps_the_session_until_expires() { /// How long after the manual clock's start [`expiring`] grants run out. const EXPIRES: Duration = Duration::from_secs(3); -/// Make the test runtime's clock one that only the test moves, wall time included, -/// so a grant's life is measured exactly. Call it before anything reads the clock. -fn manual() -> kio::clock::Installed { - kio::clock::Manual::new().install(&tokio::runtime::Handle::current()) -} - /// Everything under the dialed path, until [`EXPIRES`] from now on the ambient clock. fn expiring() -> Grant { let mut grant = Grant::new(all(), all()); @@ -1771,24 +1773,22 @@ async fn connect_through_embedded_admission( /// comes closes the connection at the deadline, not after a fixed cap of its own. #[tokio::test] async fn an_unanswered_admission_closes_at_the_handshake_deadline() { - kio::clock::host_time(); - let (relay, mut admissions, connecting) = connect_through_embedded_admission(Duration::from_secs(1)).await; + let clock = crate::manual(); + let deadline = Duration::from_secs(1); + let (relay, mut admissions, connecting) = connect_through_embedded_admission(deadline).await; let trigger = relay.shutdown_trigger().clone(); let running = tokio::spawn(relay.run()); let admission = admissions.next().await.expect("the admission is asked"); - // Only timers decide what follows: the deadline passes with nobody answering. - tokio::time::pause(); - tokio::time::advance(Duration::from_secs(2)).await; - tokio::time::resume(); - // The close reaches the client over loopback within moments; the old fixed cap - // would have held the connection for another eight seconds. + // Only timers decide what follows: the deadline passes with nobody answering, then + // the clock stands still, so the close owes nothing to the old fixed cap. + clock.advance(deadline); let connecting = kio::clock::timeout(TIMEOUT, connecting) .await .expect("the client never settled") .expect("client task"); if let Ok(connection) = connecting { - kio::clock::timeout(Duration::from_secs(3), connection.closed()) + kio::clock::timeout(TIMEOUT, connection.closed()) .await .expect("the connection outlived the handshake deadline") .expect_err("closed as refused, not cleanly"); diff --git a/rs/moq-relay/tests/integration/main.rs b/rs/moq-relay/tests/integration/main.rs index 42c1bfa9b06..49c21c23910 100644 --- a/rs/moq-relay/tests/integration/main.rs +++ b/rs/moq-relay/tests/integration/main.rs @@ -53,6 +53,13 @@ fn certificates() -> moq_tokio::tls::Certificates { moq_tokio::tls::Certificates::from_pem(cert.pem().as_bytes()).expect("certificates") } +/// Make the test runtime's clock one that only the test moves, wall time included, so +/// the relay's timers fire only on `advance` while its sockets stay real. Call it before +/// anything reads the clock. +fn manual() -> kio::clock::Installed { + kio::clock::Manual::new().install(&tokio::runtime::Handle::current()) +} + /// Build a cluster and spawn its driver, the way an embedding edge runs one. A /// failed driver panics its task. fn spawn_cluster(options: moq_relay::cluster::Options) -> moq_relay::cluster::Cluster { diff --git a/rs/moq-relay/tests/integration/session_revalidate.rs b/rs/moq-relay/tests/integration/session_revalidate.rs index 555dbe5cfcd..6f31e5bb745 100644 --- a/rs/moq-relay/tests/integration/session_revalidate.rs +++ b/rs/moq-relay/tests/integration/session_revalidate.rs @@ -17,6 +17,8 @@ use moq_auth::{Event, Grant, Pattern, Patterns, Request}; use moq_relay::session::{Filter, List, Nudged}; use moq_relay::{Connection, auth, cluster, internal, web}; +/// Bounds a wait on the host clock. Under `crate::manual()` it never fires, and nextest's +/// `terminate-after` bounds the test instead. const TIMEOUT: Duration = Duration::from_secs(10); #[derive(Clone)] @@ -289,32 +291,28 @@ async fn connect(url: url::Url, bind: &str) -> moq_tokio::Connection { .expect("connect failed") } -async fn wait_listed(fixture: &Fixture, n: usize) -> List { - let deadline = kio::clock::now() + Duration::from_secs(5); - loop { - let (status, list) = fixture.list("").await; - assert_eq!(status, reqwest::StatusCode::OK); - if list.sessions.len() >= n { - return list; - } - assert!( - kio::clock::now() < deadline, - "expected {n} sessions, got {}", - list.sessions.len() - ); - kio::clock::sleep(Duration::from_millis(25)).await; +/// Every session on the node, once all `n` are. A session registers when admitted, +/// which its client's handshake need not wait for, and the registry signals nothing, +/// so yield until it holds them; no timer runs. +async fn listed(fixture: &Fixture, n: usize) -> List { + while fixture.sessions.list(&Filter::default()).len() < n { + tokio::task::yield_now().await; } + let (status, list) = fixture.list("").await; + assert_eq!(status, reqwest::StatusCode::OK); + assert_eq!(list.sessions.len(), n, "{list:?}"); + list } -/// A push by id closes a refused session well inside the cadence, and `end` says -/// `refused`. +/// A push by id closes a refused session with the clock standing still, so no +/// cadence did, and `end` says `refused`. #[tokio::test] async fn push_by_id_refuses_inside_the_cadence() { - kio::clock::host_time(); + let _clock = crate::manual(); let script = Script::new(grant()); let fixture = Fixture::tcp(build_auth(script.spawn().await)).await; let session = connect(fixture.path("/room"), "127.0.0.1:0").await; - let list = wait_listed(&fixture, 1).await; + let list = listed(&fixture, 1).await; assert_eq!(list.sessions.len(), 1); assert!( serde_json::to_value(&list.sessions[0]).unwrap().get("query").is_none(), @@ -327,10 +325,7 @@ async fn push_by_id_refuses_inside_the_cadence() { assert_eq!(status, reqwest::StatusCode::ACCEPTED); assert_eq!(nudged.ids.as_slice(), std::slice::from_ref(&id)); - kio::clock::timeout(Duration::from_secs(3), session.closed()) - .await - .expect("a push must close the session well inside the cadence") - .expect_err("refused as Unauthorized"); + session.closed().await.expect_err("refused as Unauthorized"); assert_eq!(script.end_of(&id).await, moq_auth::lease::Reason::Refused); assert_eq!(script.ends().len(), 1); @@ -345,9 +340,9 @@ async fn path_and_cidr_select_a_subset() { let a = connect(fixture.path("/demo/one"), "127.0.0.1:0").await; let b = connect(fixture.path("/demo/two"), "127.0.0.1:0").await; let c = connect(fixture.path("/other"), "127.0.0.1:0").await; - let listed = wait_listed(&fixture, 3).await; + let all = listed(&fixture, 3).await; let id_of = |path: &str| { - let session = listed.sessions.iter().find(|session| session.path == path); + let session = all.sessions.iter().find(|session| session.path == path); session.unwrap_or_else(|| panic!("{path} is listed")).id.clone() }; @@ -405,7 +400,7 @@ async fn websocket_is_listed_and_closes_on_a_push() { let script = Script::new(grant()); let fixture = Fixture::ws(build_auth(script.spawn().await)).await; let session = connect(fixture.path("/room"), "127.0.0.1:0").await; - let list = wait_listed(&fixture, 1).await; + let list = listed(&fixture, 1).await; assert_eq!(list.sessions[0].transport, moq_auth::Transport::WebSocket); script.on_revalidate(Answer::Status(403)); @@ -424,7 +419,7 @@ async fn empty_filter_reaches_all() { let fixture = Fixture::tcp(build_auth(script.spawn().await)).await; let _a = connect(fixture.path("/a"), "127.0.0.1:0").await; let _b = connect(fixture.path("/b"), "127.0.0.1:0").await; - wait_listed(&fixture, 2).await; + listed(&fixture, 2).await; let (status, nudged) = fixture.revalidate("").await; assert_eq!(status, reqwest::StatusCode::ACCEPTED); @@ -440,11 +435,11 @@ async fn empty_filter_reaches_all() { /// A retier arrives on the next reply after a push, and the session stays. #[tokio::test] async fn a_pushed_retier_arrives_on_the_next_reply() { - kio::clock::host_time(); + let _clock = crate::manual(); let script = Script::new(grant()); let fixture = Fixture::tcp(build_auth(script.spawn().await)).await; let session = connect(fixture.path("/room"), "127.0.0.1:0").await; - let id = wait_listed(&fixture, 1).await.sessions[0].id.clone(); + let id = listed(&fixture, 1).await.sessions[0].id.clone(); let mut moved = grant(); moved.tier = Some("gold".into()); @@ -452,9 +447,9 @@ async fn a_pushed_retier_arrives_on_the_next_reply() { let (status, _) = fixture.revalidate("").await; assert_eq!(status, reqwest::StatusCode::ACCEPTED); - // A re-check applies a new tier only once it has decided to keep the session. + // A re-check applies a new tier only once it has decided to keep the session. The + // stats signal no change, so yield until the reply lands; the clock stands still. let gold = moq_net::stats::Tier::new(moq_net::Path::from_static("gold")); - let deadline = kio::clock::now() + TIMEOUT; while !fixture .stats .snapshot() @@ -462,8 +457,7 @@ async fn a_pushed_retier_arrives_on_the_next_reply() { .iter() .any(|(tier, presence)| *tier == gold && presence.active() == 1) { - assert!(kio::clock::now() < deadline, "the pushed retier never applied"); - kio::clock::sleep(Duration::from_millis(20)).await; + tokio::task::yield_now().await; } session.abort(moq_net::Error::Cancel); assert_eq!( diff --git a/rs/moq-relay/tests/integration/socket_auth.rs b/rs/moq-relay/tests/integration/socket_auth.rs index d96b2f984d4..d624551e42e 100644 --- a/rs/moq-relay/tests/integration/socket_auth.rs +++ b/rs/moq-relay/tests/integration/socket_auth.rs @@ -341,7 +341,6 @@ async fn a_client_grant_neither_serves_nor_reads_clients() { /// track, so it cannot learn a session's name to read or forge it. #[tokio::test] async fn a_reader_of_everything_lists_no_session() { - kio::clock::host_time(); let fixture = Fixture::new().await; let (alice, _alice_origin, _alice_session) = fixture.endpoint(Role::Client, "alice", LITE).await; let (bob, _bob_origin, _bob_session) = fixture.endpoint(Role::Server, "bob", LITE).await;