From 7a833f4ba1af8e7d926a5505c03d3ee7c45ac897 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 22 Sep 2026 08:56:31 -0700 Subject: [PATCH] test(moq-gst): decide the cancelled pump race instead of sleeping on it `a_subscription_resolving_after_cancellation_creates_no_pad` cleared the catalog and answered the subscription immediately after, assuming the session had reconciled the removal by then. It had not on a slower machine: the pump won its select, went live, and took the pad the test forbids. CI failed it twice in a row while it passed 12/12 here. The interleaving is not the test's to choose, so it stops trying. The session test now waits for the cancelled pump to drop its subscription, an edge it can observe through `Request::poll_unused`, and asserts with no sleep at all. What that wait gives up, a subscription resolving *after* the teardown, is what `PumpState` decides, so the compare and swap that refuses the losing side gets a unit test of its own. Breaking `go_live` fails that test; the old one only caught it by luck. Co-Authored-By: Claude Opus 5 --- rs/moq-gst/src/source/imp.rs | 54 ++++++++++++++++++++++++++++++------ 1 file changed, 45 insertions(+), 9 deletions(-) diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs index eb364b2463..c7cd09b910 100644 --- a/rs/moq-gst/src/source/imp.rs +++ b/rs/moq-gst/src/source/imp.rs @@ -892,7 +892,7 @@ fn audio_caps(config: &hang::catalog::AudioConfig) -> Result { #[cfg(test)] mod tests { - use super::{plan_reconcile, relative_pts}; + use super::{PumpState, plan_reconcile, relative_pts}; use moq_net::Timestamp; use std::collections::HashMap; @@ -961,6 +961,29 @@ mod tests { gst::ClockTime::from_mseconds(500) ); } + + /// The pad claim and the teardown race for every pump, and only one may win: a subscription + /// resolving after a teardown must not publish a rendition the session has finished with and + /// then yank it without an EOS. The cancel watch alone cannot say which happened, so this is + /// the state that does. + #[test] + fn a_pump_either_goes_live_or_is_cancelled() { + let cancelled = PumpState::new(); + assert!(cancelled.cancel_before_live()); + assert!(!cancelled.go_live(), "a cancelled pump still claimed a pad"); + assert!( + !cancelled.cancel_before_live(), + "a second teardown claimed the same transition" + ); + + let live = PumpState::new(); + assert!(live.go_live()); + assert!( + !live.cancel_before_live(), + "a live pump was dropped without its cancel watch" + ); + assert!(!live.go_live(), "a live pump claimed a second pad"); + } } #[cfg(test)] @@ -1120,11 +1143,12 @@ mod session_tests { super::RUNTIME.block_on(session).unwrap().unwrap(); } - /// A subscription can resolve after its pump was already torn down. The pump has to stay - /// dead: exposing a pad at that point publishes a rendition the session has finished with, - /// and then yanks it without an EOS. + /// A rendition delisted while its pump is still subscribing ends that pump, and answering + /// the subscription afterwards must not resurrect it into a pad. Which of the two the pump + /// sees first is the runtime's to decide, so the state machine that refuses the losing side + /// is covered by `a_pump_either_goes_live_or_is_cancelled` instead. #[test] - fn a_subscription_resolving_after_cancellation_creates_no_pad() { + fn a_rendition_delisted_while_subscribing_takes_no_pad() { let _pad_ids = pad_ids(); let element = element(); @@ -1150,11 +1174,23 @@ mod session_tests { guard.video.renditions.clear(); } - // Only now answer it. The pump was torn down, so nothing may reach a pad. Proving a pad - // never appears has no edge to wait on, unlike `await_pad`, so this gives the runtime a - // window in which the un-cancelled version reliably creates one. + // A cancelled pump returns out of its subscribe, dropping the only consumer this request + // has: that edge says the session reconciled the removal, which a fixed beat can only + // guess at. Answering before it lands is a pump that legitimately goes live, so the + // wait is what the assertion below is about. + super::RUNTIME + .block_on(async { + tokio::time::timeout( + Duration::from_secs(10), + moq_net::kio::wait(|waiter| request.poll_unused(waiter)), + ) + .await + }) + .expect("the cancelled pump never dropped its subscription"); + + // Only now answer it. The pump is gone and its state is terminal, so no later scheduling + // can produce a pad. let _serving = request.accept(moq_net::track::Info::default()); - std::thread::sleep(Duration::from_millis(500)); assert!(pads(&element, "video_").is_empty(), "a cancelled pump still took a pad"); let _ = shutdown.send(true);