diff --git a/quest/m0/README.md b/quest/m0/README.md index d24c0c1129..11842d46a1 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -60,7 +60,6 @@ remain and no release waits on them. - [Self-hosted CI](/quest/m0/self-hosted-ci.md) - same-repo Check and Test run on a self-hosted NixOS runner with a main-written local cache, behind a `CI_RUNNER` kill switch - [CI host](/quest/m0/ci-host.md) - the maintainer brings up the spare desktop as the `moq-ci` and `moq-gpu` runner host - [Draft-22 media on 0.17](/quest/m0/release-22/README.md) - a 0.17.x with the LOCATION_FILTER and FIRST_OBJECT fixes and moq-noq 1.3.5, before Seattle -- [FFI publisher stall](/quest/m0/ffi-publisher-stall.md) - every Go, Python, and C++ publisher cell passes reliably with the serve budgets, and a cell fails when a connection idles out - [Paused spinner](/quest/m0/watch-paused-spinner.md) - the watch buffering spinner never covers the paused play button, so Interop's resume step can click it - [Check base](/quest/m0/check-base.md) - `just check` diffs against the PR's base whatever the local branch tracks, so scoped checks stay scoped - [web-transport releases the qmux fixes](/quest/m0/qmux-credit-upstream.md) - waiting on moq-dev/web-transport#412 and #413 to merge and ship, which qmux credit bumps to diff --git a/quest/m0/ffi-publisher-stall.md b/quest/m0/ffi-publisher-stall.md deleted file mode 100644 index 13ae5d6393..0000000000 --- a/quest/m0/ffi-publisher-stall.md +++ /dev/null @@ -1,60 +0,0 @@ -# [S] FFI publishers stay connected through interop - -## Goal - -Every cell of `just test interop --all` with a Go, Python, or C++ publisher passes -reliably on `main` and in nightly CI. Today a moq-ffi publisher goes silent, -not even sending QUIC keep-alives, until the relay's 10 s idle timeout closes -its connection and it reconnects. The browser cells (`go -> js`, -`python -> js`, `cpp -> js`) can't absorb that within their 30 s limit and fail at 32 to -33 s; `-> rust` and `-> gst` pass slowly at 10 to 11 s, and in some runs most -cells of one publisher fail at 11 s. The harness fails any cell whose -connection idles out, so a drop and reconnect can no longer pass as a slow -cell. - -## Plan - -The cause is known: a serve loop with work always ready never yields and -starves moq-ffi's single runtime thread until the relay times the publisher -out (found 2026-10-08 landing #4225). The publish serve loops now yield -through a `kio::coop::Budget` (#5088), which is meant to fix it: interop run -37889112106 passed every Go and Python publisher cell on an earlier design of -that PR. Confirm on `main` and nightly before closing. - -Decided 2026-10-08: - -- In m0, moved from m1 and widened from the browser cells: the stall masks - interop on every wire PR and may disconnect real FFI publishers. -- Re-check every Go, Python, and C++ publisher cell now that the budget landed. - The #4225 root-cause comment saw a 64-iteration yield still time the - publisher out, so if a cell still stalls, find the rest of the cause. - Never raise a timeout or add a retry. -- The harness fails a cell when any connection in it idles out, so this - class of regression fails every run instead of passing slowly. -- The harness shuts down every client it starts cleanly, so every idle-out - counts. Today `gst-launch` dies on SIGPIPE and `moq-cli` can be killed - under `timeout -k`, which may leave a QUIC connection to idle out. No - exception list for killed clients: it would hide a real idle-out behind a - harness kill. - -Harness facts (2026-10-08): - -- The one-byte subscribers exit on their first byte, so they pass even when - the publisher stalls after it; `-> rust` and `-> gst` exit only on their - next write after `head` closes, and the browser cells wait for rendered - video, then audio, pause, and resume. Judge a fix by those. -- The js cell (`test/interop/interop.sh` js subscriber, `driver.ts`) runs - each step within the cell's timeout, and `harness.ts` `waitForWatch` has - its own hard-coded 30 s. -- One relay serves the whole run, and its `connection closed err=... timed - out` warning (`rs/moq-relay/src/relay.rs`) names no connection and lands - about 10 s after the peer went quiet, often in a later cell. Tie each - idle-out to its connection first, by logging the connection there or - having the client report its own drop. - -Public API: none expected. Wire: none. - -## Related - -- [FFI runtime](/quest/m1/ffi-runtime.md) - FFI apps run on more than one thread, which also hides the stall -- [Audio group duration](/quest/m1/audio-group-duration.md) - the go client's 2.5 ms Opus frames stop minting a group each diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 87fb087201..3e9f68bf61 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -1248,8 +1248,9 @@ pub unsafe extern "C" fn moq_session_connect( /// Returns immediately: zero on success, or a negative code if the session is /// unknown or already closing. Does NOT free `user_data`. The /// [moq_session_connect] `on_status` callback still fires once more with a -/// terminal `0` (or a negative error), and that final callback is where -/// `user_data` should be released. Safe to call from any thread, including from +/// terminal `0` (or a negative error) once the close has reached the peer, after +/// at most a second of draining, and that final callback is where `user_data` +/// should be released. Safe to call from any thread, including from /// within `on_status`. #[unsafe(no_mangle)] pub extern "C" fn moq_session_close(session: u32) -> i32 { diff --git a/rs/moq-c/src/session.rs b/rs/moq-c/src/session.rs index ca44392717..50ee51e6da 100644 --- a/rs/moq-c/src/session.rs +++ b/rs/moq-c/src/session.rs @@ -120,10 +120,13 @@ impl Session { let _consume = consume; let res = tokio::select! { - // close() requested: a clean shutdown delivers a terminal 0. - _ = closed.1 => Ok(()), - res = Self::report(callback, reconnect) => res, + // close() requested: drain what the session queued, then close it. + _ = closed.1 => reconnect.clone().close().await.map_err(map_connect_error), + res = Self::report(callback, reconnect.clone()) => res, }; + // The terminal callback waits for the close to reach the peer, so a caller that exits + // on it doesn't leave the peer to time the connection out. + client.close().await; // Deliver one final terminal callback (0 = closed, < 0 = error), then // drop the entry. Pull it out from under the lock so the callback never diff --git a/rs/moq-gst/src/sink/session.rs b/rs/moq-gst/src/sink/session.rs index 5cfe0edaaa..d5ea8eda97 100644 --- a/rs/moq-gst/src/sink/session.rs +++ b/rs/moq-gst/src/sink/session.rs @@ -253,6 +253,9 @@ pub(crate) struct Session { join: tokio::task::JoinHandle<()>, /// The reconnect loop, held so [`stop`](Self::stop) can wait for it to end. connection: moq_tokio::Connection, + /// The client behind `connection`, closed by [`stop`](Self::stop) so the close reaches the + /// relay before the process exits rather than leaving it to time the connection out. + client: moq_tokio::Client, status: Arc, /// The live send-bitrate estimate, tracked across reconnects by the reconnect loop. Read directly /// by the `estimated-send-rate` getter. @@ -325,6 +328,7 @@ impl Session { Self { join, connection: reconnect, + client, status, send_bandwidth, recv_bandwidth, @@ -376,7 +380,11 @@ impl Session { // The status task goes first, so it never reports the loop's end as a failure. self.join.abort(); self.connection.abort(moq_net::Error::Cancel); - let _ = crate::block_on(self.connection.closed()); + crate::block_on(async { + let _ = self.connection.closed().await; + // A clone, since `Drop` keeps the fields in place; it closes the shared endpoint. + self.client.clone().close().await; + }); } } diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs index fff1b9c75f..aec167a879 100644 --- a/rs/moq-gst/src/source/imp.rs +++ b/rs/moq-gst/src/source/imp.rs @@ -97,27 +97,40 @@ struct SessionController { join: tokio::task::JoinHandle<()>, /// The one-shot connection, held so [`stop`](Self::stop) can end it and wait for it to close. connection: moq_tokio::Connection, + /// The client behind `connection`, closed by [`stop`](Self::stop) so the close reaches the + /// relay before `gst-launch` exits rather than leaving it to time the connection out. + client: moq_tokio::Client, } impl SessionController { fn start(settings: ResolvedSettings, element: glib::WeakRef) -> Result { - let (connection, origin) = connect(&settings)?; + let (client, connection, origin) = connect(&settings)?; let task_connection = connection.clone(); let task_element = element.clone(); - Ok(Self::spawn(connection, element, move |mut shutdown| async move { - run_session( - &task_connection, - origin, - settings.broadcast, - task_element, - &mut shutdown, - ) - .await - })) + Ok(Self::spawn( + client, + connection, + element, + move |mut shutdown| async move { + run_session( + &task_connection, + origin, + settings.broadcast, + task_element, + &mut shutdown, + ) + .await + }, + )) } /// Run `session` as this element's session task, reporting its error on the bus. - fn spawn(connection: moq_tokio::Connection, element: glib::WeakRef, session: F) -> Self + fn spawn( + client: moq_tokio::Client, + connection: moq_tokio::Connection, + element: glib::WeakRef, + session: F, + ) -> Self where F: FnOnce(watch::Receiver) -> Fut, Fut: Future> + Send + 'static, @@ -142,6 +155,7 @@ impl SessionController { shutdown: shutdown_tx, join, connection, + client, } } @@ -174,7 +188,12 @@ impl SessionController { ); } - let Self { join, connection, .. } = self; + let Self { + join, + connection, + client, + .. + } = self; crate::block_on(async move { // The connection outlives the session so a stop never reads as a dropped connection. if !own && let Err(err) = join.await { @@ -182,12 +201,15 @@ impl SessionController { } connection.abort(moq_net::Error::Cancel); let _ = connection.closed().await; + client.close().await; }); } } -/// Start the one-shot dial, returning it with the origin its announcements land in. -fn connect(settings: &ResolvedSettings) -> Result<(moq_tokio::Connection, moq_net::origin::Consumer)> { +/// Start the one-shot dial, returning it with its client and the origin its announcements land in. +fn connect( + settings: &ResolvedSettings, +) -> Result<(moq_tokio::Client, moq_tokio::Connection, moq_net::origin::Consumer)> { let mut config = moq_tokio::connect::Config::default(); config.tls.insecure = Some(settings.tls_disable_verify); @@ -197,12 +219,12 @@ fn connect(settings: &ResolvedSettings) -> Result<(moq_tokio::Connection, moq_ne let consumer = origin.consume(); // One-shot: the catalog subscription dies with the session anyway, so a background redial could // not resurrect this run. A drop surfaces as the catalog closing and the session winding down. - let connection = config + let client = config .init(Default::default())? .with_subscriber(origin) - .with_reconnect(false) - .connect(settings.url.clone()); - Ok((connection, consumer)) + .with_reconnect(false); + let connection = client.connect(settings.url.clone()); + Ok((client, connection, consumer)) } #[derive(Default)] @@ -1706,9 +1728,10 @@ mod session_tests { element: &super::super::MoqSrc, broadcast: moq_net::broadcast::Consumer, ) -> (SessionController, moq_tokio::Connection) { - let (connection, _) = super::connect(&unreachable()).unwrap(); + let (client, connection, _) = super::connect(&unreachable()).unwrap(); let weak = element.downgrade(); let session = SessionController::spawn( + client, connection.clone(), element.downgrade(), move |mut shutdown| async move { follow_catalog(broadcast, weak, &mut shutdown).await }, diff --git a/rs/moq-relay/src/relay.rs b/rs/moq-relay/src/relay.rs index 494466321e..7ae7335ba2 100644 --- a/rs/moq-relay/src/relay.rs +++ b/rs/moq-relay/src/relay.rs @@ -782,14 +782,18 @@ async fn serve_listening( sessions: crate::session::Registry, ) -> anyhow::Result<()> { while let Some(request) = listener.accept().await { + let id = cluster.next_connection_id(); let conn = Connection::new(request, cluster.clone(), auth.clone()) - .with_id(cluster.next_connection_id()) + .with_id(id) .with_shutdown(shutdown.clone()) .with_sessions(sessions.clone()); tokio::spawn(async move { - if let Err(err) = conn.run().await { - tracing::warn!(%err, "connection closed"); + match conn.run().await { + // A lease or shutdown ending the session closes it too: logged, so every + // accepted connection pairs with its end. + Ok(()) => tracing::info!(id, "connection closed"), + Err(err) => tracing::warn!(id, %err, "connection closed"), } }); } diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 7b0c969f26..66df3dbe53 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -786,8 +786,11 @@ async fn serve_connection( // and the shutdown broadcast on the shared runtime. let shutdown = serve.shutdown.clone(); serve.tokio.spawn(async move { - if let Err(err) = crate::connection::supervise(session, lease, shutdown, registration).await { - tracing::warn!(id, %err, "connection closed"); + match crate::connection::supervise(session, lease, shutdown, registration).await { + // A lease or shutdown ending the session closes it too: logged, so every + // accepted connection pairs with its end. + Ok(()) => tracing::info!(id, "connection closed"), + Err(err) => tracing::warn!(id, %err, "connection closed"), } }); diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index 6ccede73f0..2c217564de 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -572,18 +572,7 @@ impl Client { let moq = moq.clone(); let dial = Box::pin(async move { let quic = quic.await?; - Ok(Box::pin(async move { - let session = connect_session(&moq, quic).await?; - // Wait for the peer's SETUP before draining WebSocket. This crate's servers - // send SETUP after admission; other servers may still refuse afterward. - // A refusal, or a version with no SETUP to wait on, keeps WebSocket. - session.setup().await?; - // A session already told to leave would hand straight back out of QUIC. - if session.draining().peek().is_some() { - return Err(Error::ConnectFailed); - } - Ok(session) - }) as Handshake) + Ok(Box::pin(async move { Ok(connect_session(&moq, quic).await?) }) as Handshake) }); Upgrade::new(dial, transport) }); @@ -688,7 +677,7 @@ impl Dialed { } } -/// The MoQ handshake running on an upgrade's QUIC session. +/// The MoQ handshake starting on an upgrade's QUIC session, up to sending our SETUP. pub(crate) type Handshake = BoxFuture<'static, crate::Result>; /// A QUIC dial that lost the race to the WebSocket fallback but is still going. @@ -710,12 +699,20 @@ enum Stage { Dialing(BoxFuture<'static, crate::Result>), /// The QUIC transport is up and the MoQ handshake is running on it. Handshaking(Handshake), + /// Our SETUP is sent, and `setup` waits on the peer's. + Setup { + session: moq_net::Session, + setup: BoxFuture<'static, crate::Result<()>>, + }, } /// How far an [`Upgrade`] got on one poll. pub(crate) enum Step { /// The QUIC transport is up; the MoQ handshake has started on it. Handshaking, + /// The MoQ session exists and waits on the peer's SETUP. It is the caller's to close + /// if it stops before [`Step::Done`], since the peer already holds it open. + Session(moq_net::Session), /// The peer's SETUP arrived over QUIC, which is ready to take over on this transport. Done(moq_net::Session, crate::Transport), } @@ -737,8 +734,9 @@ impl Upgrade { } /// Drive the dial: `Ready(Ok(Step::Handshaking))` once when the QUIC transport - /// comes up, then `Ready(Ok(Step::Done))` once the peer's SETUP arrives. Not - /// polled again after `Done` or an error. + /// comes up, `Ready(Ok(Step::Session))` once our SETUP is sent, then + /// `Ready(Ok(Step::Done))` once the peer's SETUP arrives. Not polled again after + /// `Done` or an error. pub(crate) fn poll(&mut self, waiter: &moq_net::kio::Waiter) -> Poll> { if let Some((sleep, timeout)) = &mut self.deadline && waiter.poll_future(sleep.as_mut()).is_ready() @@ -754,7 +752,27 @@ impl Upgrade { } Stage::Handshaking(handshake) => { let session = ready!(waiter.poll_future(handshake.as_mut()))?; - Poll::Ready(Ok(Step::Done(session, self.transport))) + let peer = session.clone(); + let setup = Box::pin(async move { + // Wait for the peer's SETUP before draining WebSocket. This crate's servers + // send SETUP after admission; other servers may still refuse afterward. + // A refusal, or a version with no SETUP to wait on, keeps WebSocket. + peer.setup().await?; + // A session already told to leave would hand straight back out of QUIC. + if peer.draining().peek().is_some() { + return Err(Error::ConnectFailed); + } + Ok(()) + }); + self.stage = Stage::Setup { + session: session.clone(), + setup, + }; + Poll::Ready(Ok(Step::Session(session))) + } + Stage::Setup { session, setup } => { + ready!(waiter.poll_future(setup.as_mut()))?; + Poll::Ready(Ok(Step::Done(session.clone(), self.transport))) } } } diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index 5c5d8f555c..d9915fb07c 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -470,6 +470,10 @@ struct State { /// The loop's [`Draining`] predecessor and its handover deadline, so /// [`Connection::close`] drains it too without overstaying that window. predecessor: Option<(moq_net::Session, tokio::time::Instant)>, + /// A QUIC upgrade's session, from our SETUP until it takes over or is dropped, so + /// [`Connection::close`] and [`Connection::abort`] end it too. The peer already holds + /// it open, and the loop's teardown would only drop it after they return. + upgrade: Option, } /// The producer side of everything a [`Connection`] handle can observe. @@ -508,6 +512,21 @@ impl Shared { state.version = Some(session.version()); state.transport = Some(transport); state.session = Some(session.clone()); + // Whatever upgrade was coming up has taken over or is superseded. + state.upgrade = None; + } + } + + /// A QUIC upgrade's session exists, or (`None`) no longer needs ending on its own. + fn upgrading(&self, session: Option<&moq_net::Session>) { + // Held across the publish, as in `connected`. + let closed = self.closed.lock().unwrap(); + if let (Some(err), Some(session)) = (closed.as_ref(), session) { + session.abort(err.clone()); + return; + } + if let Ok(mut state) = self.state.write() { + state.upgrade = session.cloned(); } } @@ -543,6 +562,7 @@ impl Shared { state.version = None; state.transport = None; state.session = None; + state.upgrade = None; } let _ = self.send_bw.set(None); let _ = self.recv_bw.set(None); @@ -738,11 +758,15 @@ impl Connection { pub fn abort(&self, err: moq_net::Error) { // Record the close and take the session under one lock, so a redial landing // in the abort window is refused rather than parked: see [`CloseGuard`]. - let session = { + let (session, upgrade) = { let mut closed = self.task.closed.lock().unwrap(); *closed = Some(err.clone()); - self.state.read().session.clone() + let state = self.state.read(); + (state.session.clone(), state.upgrade.clone()) }; + if let Some(upgrade) = upgrade { + upgrade.abort(err.clone()); + } if let Some(session) = session { session.abort(err); } @@ -751,6 +775,7 @@ impl Connection { /// Stop the loop for every clone, closing the live session (and any predecessor /// still finishing after a GOAWAY) once the data it queued has been delivered. + /// A QUIC upgrade still in its handshake has served nothing, so it closes at once. /// /// See [`moq_net::Session::close`]: finished tracks deliver their last groups and /// FIN and announcements are withdrawn first, bounded by a one second deadline @@ -759,11 +784,11 @@ impl Connection { /// when nothing was live, and a session's error if it did not drain. pub async fn close(self) -> crate::Result<()> { // Refuse redials and take the sessions under one lock: see [`CloseGuard`]. - let (session, predecessor) = { + let (session, predecessor, upgrade) = { let mut closed = self.task.closed.lock().unwrap(); *closed = Some(moq_net::Error::Cancel); let state = self.state.read(); - (state.session.clone(), state.predecessor.clone()) + (state.session.clone(), state.predecessor.clone(), state.upgrade.clone()) }; self.task.handle.abort(); let session = async move { @@ -786,7 +811,14 @@ impl Connection { } } }; - let (session, predecessor) = tokio::join!(session, predecessor); + let upgrade = async move { + if let Some(upgrade) = upgrade { + upgrade.abort(moq_net::Error::Cancel); + // Within the live session's one second, so a stuck transport can't hold `close`. + let _ = tokio::time::timeout(Duration::from_secs(1), upgrade.closed()).await; + } + }; + let (session, predecessor, ()) = tokio::join!(session, predecessor, upgrade); Ok(session.and(predecessor)?) } @@ -856,6 +888,8 @@ impl Connection { } } }; + // Any upgrade still coming up is dropped with the session it was replacing. + shared.upgrading(None); // A session that stayed up past the initial backoff is healthy; one that // ended sooner counts as a failed attempt however it ended. @@ -1474,6 +1508,7 @@ fn poll_upgrade( while let Some(pending) = upgrade.as_mut() { match ready!(pending.poll(waiter)) { Ok(crate::client::Step::Handshaking) => shared.migrating(), + Ok(crate::client::Step::Session(session)) => shared.upgrading(Some(&session)), Ok(crate::client::Step::Done(session, transport)) => { *upgrade = None; return Poll::Ready((session, transport)); @@ -1483,6 +1518,7 @@ fn poll_upgrade( // either way: this session was serving throughout. tracing::debug!(%err, "QUIC upgrade failed; staying on WebSocket"); shared.stayed(); + shared.upgrading(None); *upgrade = None; } } @@ -2646,6 +2682,96 @@ mod tests { assert_eq!(next_group(&mut sub).await, 1); } + /// Aborting while a QUIC upgrade is still in its MoQ handshake ends the upgrade's + /// session too, with the caller's error, before `abort` returns. + /// + /// Left to the loop's teardown, the session would only be dropped once the aborted + /// task unwound, with a bare cancel. A process that exits in between, as an embedder + /// does right after `abort` or `close`, leaves the server to time the connection out. + /// The error the server reads is what tells the two paths apart. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test] + async fn abort_ends_an_upgrade_in_flight() { + let origin = crate::origin::spawn(); + let mut fallback = Fallback::start(&origin, Quic::Refused).await; + + let mut config = crate::connect::Config::default(); + config.tls.insecure = Some(true); + let client = config + .init(Default::default()) + .unwrap() + .with_subscriber(crate::origin::spawn()); + let mut connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) + .await + .expect("never connected") + .unwrap(); + assert_eq!(connection.transport(), Some(crate::Transport::WebSocket)); + let _websocket = fallback.accept().await; + + // The QUIC transport comes up and its handshake waits on the server's answer. + fallback.forwarder.open(); + let request = fallback.held().await; + let status = tokio::time::timeout(UPGRADE_WAIT, connection.status()).await.unwrap(); + assert_eq!(status.unwrap(), Status::Migrating); + + connection.abort(moq_net::Error::GoawayTimeout); + let closed = async { + match request.ok().await { + Ok(session) => session.closed().await, + Err(err) => panic!("the server could not answer the upgrade: {err}"), + } + }; + let err = tokio::time::timeout(UPGRADE_WAIT, closed) + .await + .expect("the upgrade's connection was left open"); + let code = |err: &moq_net::Error| moq_net::SessionError::from(err).to_code(); + assert_eq!( + code(&err), + code(&moq_net::Error::GoawayTimeout), + "the upgrade ended without the abort's error: {err}" + ); + } + + /// Closing while a QUIC upgrade is still in its MoQ handshake ends the upgrade too, and + /// returns promptly rather than waiting on it. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test] + async fn close_ends_an_upgrade_in_flight() { + let origin = crate::origin::spawn(); + let mut fallback = Fallback::start(&origin, Quic::Refused).await; + + let mut config = crate::connect::Config::default(); + config.tls.insecure = Some(true); + let client = config + .init(Default::default()) + .unwrap() + .with_subscriber(crate::origin::spawn()); + let mut connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) + .await + .expect("never connected") + .unwrap(); + let _websocket = fallback.accept().await; + + fallback.forwarder.open(); + let request = fallback.held().await; + let status = tokio::time::timeout(UPGRADE_WAIT, connection.status()).await.unwrap(); + assert_eq!(status.unwrap(), Status::Migrating); + + // Inside the one second the live session may take to drain, plus slack. + tokio::time::timeout(Duration::from_secs(2), connection.close()) + .await + .expect("close waited on the upgrade") + .unwrap(); + let closed = async { + if let Ok(session) = request.ok().await { + session.closed().await; + } + }; + tokio::time::timeout(UPGRADE_WAIT, closed) + .await + .expect("the upgrade's connection was left open"); + } + /// When WebSocket wins the race but its MoQ handshake fails, the attempt falls back /// to the QUIC dial still pending instead of failing. /// diff --git a/test/interop/README.md b/test/interop/README.md index eaab54b809..9e7ed9b0fa 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -19,6 +19,15 @@ subscriber checks end-to-end: the browser encodes fake microphone audio, and the Python, Go, and C++ clients encode a synthetic tone through `moq-ffi` at a 2.5 ms frame duration, so the matrix covers the FFI audio path with a non-default codec config. +Every client closes its session on the way out: publishers stop when their input +ends (the browser on SIGTERM), and subscribers close once they have their frame. +After each publisher's round, the harness waits for the relay to log the close of +every connection the round opened, and fails the round for any connection the relay +timed out instead. A client that stalls and reconnects therefore fails, rather than +passing as a slow cell, even when the idle-out lands after its cell finished. The +check covers runs built from this checkout; the wire-compat lanes, which swap in +released relays and clients, skip it. + Whenever the browser subscriber is in the run, the matrix ends with a close-code case: Chromium dials the relay with a token its public rules refuse, and `WebTransport.closed` must carry the relay's code and reason. Chromium treats a diff --git a/test/interop/clients/cpp/main.cpp b/test/interop/clients/cpp/main.cpp index 4ff00cf831..4af0aaee29 100644 --- a/test/interop/clients/cpp/main.cpp +++ b/test/interop/clients/cpp/main.cpp @@ -94,6 +94,12 @@ std::shared_ptr connect(const std::string &url, Clock::time_point return until(client->connect(url), deadline, "connect"); } +// Closes the session and waits for the close to reach the relay. `cancel` would return at +// once, and stopping the runtime right after it can leave the relay to time the connection out. +void close_session(const std::shared_ptr &session) { + until(session->shutdown(), Clock::now() + std::chrono::seconds(5), "session shutdown"); +} + // Feeds the encoder a real-time tone until `stop` is set. void publish_tone(const std::shared_ptr &audio, const std::atomic &stop) { const auto started = Clock::now(); @@ -161,7 +167,7 @@ int publish(const std::string &url, const std::string &path) { ok(audio->finish(), "audio finish"); ok(media->finish(), "video finish"); ok(broadcast->close(), "broadcast close"); - session->cancel(0); + close_session(session); moq::shutdown(); return 0; } @@ -206,7 +212,7 @@ int subscribe(const std::string &url, const std::string &path, double timeout) { fail("no frame data received"); } std::printf("received %zu bytes from \"%s\"\n", total, path.c_str()); - session->cancel(0); + close_session(session); moq::shutdown(); return 0; } diff --git a/test/interop/clients/js-native/subscribe.ts b/test/interop/clients/js-native/subscribe.ts index 767bca5ab8..bf7fe40a5e 100644 --- a/test/interop/clients/js-native/subscribe.ts +++ b/test/interop/clients/js-native/subscribe.ts @@ -123,7 +123,10 @@ async function run(): Promise { throw new Error("no frame data received"); } finally { requested?.close(); - connection.close(); // returns void, not a promise + // Wait for the transport to report the close too, so it reaches the relay before + // process.exit; otherwise the relay times the connection out. + await connection.close(); + await connection.closed; origin.close(); } } diff --git a/test/interop/clients/js/driver.ts b/test/interop/clients/js/driver.ts index 508cf30f47..85d80e095a 100644 --- a/test/interop/clients/js/driver.ts +++ b/test/interop/clients/js/driver.ts @@ -24,6 +24,7 @@ import { pageUrl, pause, readPlayerState, + readResources, SELECTORS, serve, sleep, @@ -32,6 +33,22 @@ import { waitForWatch, } from "./harness"; +/** Remove the page's players, which closes their sessions, and wait until every transport has. */ +async function closeSessions(page: Page): Promise { + await page + .evaluate(() => { + for (const el of document.querySelectorAll("moq-watch, moq-publish")) el.remove(); + }) + .catch(() => {}); + const deadline = Date.now() + 5000; + while (Date.now() < deadline) { + const live = await readResources(page).catch(() => undefined); + if (!live || live.transports === 0) return; + await sleep(POLL_INTERVAL_MS); + } + console.error("a session was still open 5 s after its player was removed"); +} + const { positionals, values } = parseArgs({ allowPositionals: true, options: { @@ -99,7 +116,9 @@ try { if (role === "publish") { console.error(`publishing ${broadcast} (fake camera + microphone) to ${url}`); - await new Promise(() => {}); // stream until the orchestrator kills us + // Stream until the orchestrator stops us. + await new Promise((resolve) => process.once("SIGTERM", resolve)); + code = 0; } else { const start = Date.now(); const startupDeadline = start + timeoutMs; @@ -170,6 +189,12 @@ try { } finally { // `code` is 0 only when the role's checks all passed, so it decides whether the trace is kept. await finishTraces(code !== 0); + // Close every session before the browser: closing it, or even the page, can end them without + // a close, leaving the relay to time them out. + for (const page of browser.contexts().flatMap((context) => context.pages())) { + await closeSessions(page); + await page.close().catch(() => {}); + } await browser.close().catch(() => {}); server.stop(); } diff --git a/test/interop/clients/js/harness.ts b/test/interop/clients/js/harness.ts index 9b79020537..23c0226895 100644 --- a/test/interop/clients/js/harness.ts +++ b/test/interop/clients/js/harness.ts @@ -111,7 +111,9 @@ export function pageUrl(origin: string, role: string, params: Record { - return chromium.launch({ channel: "chromium", headless: true, args }); + // Playwright closes the browser on SIGTERM by default, which ends its sessions without a + // close. The drivers close it themselves; the harness reaps whatever a signal leaves behind. + return chromium.launch({ channel: "chromium", headless: true, args, handleSIGTERM: false }); } /** Contexts tracing this process, saved by {@link finishTraces} when the run fails. */ diff --git a/test/interop/clients/js/src/setup.ts b/test/interop/clients/js/src/setup.ts index eecdd38221..d37fa0f8c8 100644 --- a/test/interop/clients/js/src/setup.ts +++ b/test/interop/clients/js/src/setup.ts @@ -29,6 +29,8 @@ if (role === "publish") { // camera and microphone input. Audio is encoded lazily when a player asks. el.setAttribute("source", "camera"); document.body.appendChild(el); + // So the driver can wait for the session to close before closing the browser. + watchResources(); // Playwright reads the shared DOM, so mirror the public source state onto the element. const sample = () => { type CaptureSource = { out: { source: { peek(): unknown }; error?: { peek(): Error | undefined } } }; diff --git a/test/interop/compat/client.ts b/test/interop/compat/client.ts index 4c6e65a42f..27ef8213ab 100644 --- a/test/interop/compat/client.ts +++ b/test/interop/compat/client.ts @@ -52,6 +52,8 @@ if (command === "sign") { publish: origin.consume(), websocket: { enabled: false }, }); + // Stopped by the harness with SIGTERM: close, so the relay sees a close rather than timing it out. + process.once("SIGTERM", () => void connection.close()); console.log("published"); // `demand()` replaced the producer's own `used` after @moq/net 0.4.2, and this one // client runs against both packages. diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 497a778107..27886383ab 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -30,6 +30,14 @@ RERUN="$(harness_env INTEROP_TIMEOUT INTEROP_FPS INTEROP_SIZE INTEROP_PORT INTER PUBLISHERS="rust" SUBSCRIBERS="rust" +# The idle-out check guards this checkout's relay and clients. A run that swaps in other +# binaries or JS clients (the wire-compat lanes run released ones) can't be held to it: a +# released client may not close cleanly, and a released relay names no connection when one +# closes. +IDLE_CHECK=1 +if [[ -n "${RELAY_BIN:-}${MOQ_BIN:-}${INTEROP_SUB_MOQ:-}${INTEROP_NATIVE_CLIENT:-}${INTEROP_JS_PUBLISH_CLIENT:-}" ]]; then + IDLE_CHECK=0 +fi TIMEOUT="${INTEROP_TIMEOUT:-20}" FPS="${INTEROP_FPS:-30}" SIZE="${INTEROP_SIZE:-320x240}" @@ -512,12 +520,14 @@ run_publisher() { ffmpeg_h264 | "$CPP_INTEROP" publish --url "$URL" --broadcast "$broadcast" ;; js-native) - bun "$INTEROP_JS_PUBLISH_CLIENT/client.ts" publish "$URL" "$broadcast" + # exec, so the client leads the group and `stop_publisher` waits for its own close. + exec bun "$INTEROP_JS_PUBLISH_CLIENT/client.ts" publish "$URL" "$broadcast" ;; js) # Headless Chromium encodes its own H.264 from a fake camera via - # WebCodecs (lazily, once a subscriber creates demand). - cd "$CLIENTS/js" && bun driver.ts publish \ + # WebCodecs (lazily, once a subscriber creates demand). exec, so the + # driver leads the group and `stop_publisher` waits for its own close. + cd "$CLIENTS/js" && exec bun driver.ts publish \ --url "$URL" --broadcast "$broadcast" ;; *) @@ -535,6 +545,84 @@ start_publisher() { PUB_PID="$HARNESS_PID" } +# Stop a publisher the way a real one ends, so it closes its session and the idle-out check +# below counts only real stalls. A stdin-fed client finishes once ffmpeg exits and its input +# ends; the browser driver and the native JS client close on SIGTERM. Whatever is still +# running after the wait is killed, and its connection then fails the round as an idle-out. +stop_publisher() { + local pid="$1" lang="$2" deadline=$((SECONDS + 10)) + case "$lang" in + # The leader alone: a SIGTERM to the group would reach Chromium too, which then exits + # without closing its sessions. + js | js-native) kill -TERM "$pid" 2>/dev/null || true ;; + *) pkill -TERM -g "$pid" -x ffmpeg 2>/dev/null || true ;; + esac + while ! harness_exited "$pid" && ((SECONDS < deadline)); do + sleep 0.1 + done + if ! harness_exited "$pid"; then + echo " WARN publisher '$lang' was still running 10 s after it was stopped; killing it" + fi + harness_reap "$pid" +} + +# The relay log line the next connection starts on. +relay_next_line() { + echo $(($(wc -l <"$HARNESS_RUN/relay.log") + 1)) +} + +# The relay log from line FROM on, without its ANSI colors. +relay_log_since() { + tail -n +"$1" "$HARNESS_RUN/relay.log" | sed 's/\x1b\[[0-9;]*m//g' +} + +# Print each relay connection that appears from log line FROM on, with how it closed: the +# relay's close error, "ended" for a close without one, or "open" while it hasn't closed yet. +# The patterns follow the relay's `conn{id=...}` span and its `connection closed` log +# (`rs/moq-relay/src/relay.rs`). Only connections opened in the window count, so a straggler +# from an earlier round closing here isn't blamed on this one. +relay_connections() { + relay_log_since "$1" | awk ' + match($0, /conn\{id=[0-9]+ /) { + id = substr($0, RSTART + 8, RLENGTH - 9) + if (!(id in state)) { state[id] = "open"; order[++n] = id } + } + match($0, /connection closed id=[0-9]+/) { + id = substr($0, RSTART + 21, RLENGTH - 21) + if (id in state) { + rest = substr($0, RSTART + RLENGTH) + state[id] = sub(/^ err=/, "", rest) ? rest : "ended" + } + } + END { for (i = 1; i <= n; i++) print order[i], state[order[i]] } + ' +} + +# Fail the round for every relay connection it opened that idled out rather than closed. +# +# The relay times a silent peer out after its 10 s QUIC idle timeout, which can land long +# after the cell that stalled finished. Every client closes its session on the way out, so +# each close shows up at once, and an idle-out is a real stall that a reconnect would +# otherwise hide behind a slow cell. A connection still open after the idle timeout is about +# to idle out, so it fails the same way. +check_idle_outs() { + local pub="$1" from="$2" deadline=$((SECONDS + 15)) id reason role failed=0 + # Plain grep, not -q: under pipefail an early exit fails the pipeline on SIGPIPE. + while relay_connections "$from" | grep ' open$' >/dev/null && ((SECONDS < deadline)); do + sleep 0.2 + done + while read -r id reason; do + [[ "$reason" == "open" || "$reason" == *"timed out"* ]] || continue + # The relay subscribes upstream only on the publisher's connection. + role=subscriber + relay_log_since "$from" | grep "conn{id=$id .*::subscriber: subscribe started" >/dev/null && role=publisher + echo " FAIL $pub round: relay connection $id ($role) idled out instead of closing ($reason)" + relay_log_since "$from" | grep -E "conn\{id=$id |connection closed id=$id " | sed 's/^/ /' + failed=1 + done < <(relay_connections "$from") + ((failed == 0)) || overall=1 +} + # Run a native-JS subscriber and judge it by the "received N bytes" marker it # prints, not its exit code. The @moq/web-transport NAPI addon can segfault # during the runtime's exit teardown *after* a frame has arrived (an upstream bug @@ -638,18 +726,22 @@ PYCODE # video_0 by name: a bare `moqsrc ! filesink` would take whichever pad # appears first, so a publisher with audio (every one but the Rust CLI) # could pass this cell on audio bytes without video ever flowing. We grab one byte, the - # same "bytes moved" bar as the rust subscriber (no decode). head closing - # the pipe SIGPIPEs gst-launch, so success returns at once; no data just - # runs out the timeout. Our plugin dir rides on top of the system path + # same "bytes moved" bar as the rust subscriber (no decode). SIGPIPE is ignored, so + # head closing the pipe fails filesink's next write and gst-launch stops the pipeline, + # closing its session, rather than dying with it open; no data just runs out the + # timeout. Our plugin dir rides on top of the system path # (which provides filesink); a private registry keeps the scan off the # user's cache. buffer-mode=2 makes filesink unbuffered so the first frame # reaches head immediately. local n - n=$(GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_REGISTRY_1_0="$HARNESS_RUN/gst-run-registry.bin" \ - timeout -k 3 "$TIMEOUT" gst-launch-1.0 -q \ - moqsrc name=s url="$URL" broadcast="$broadcast" \ - s.video_0 ! filesink location=/dev/stdout buffer-mode=2 \ - 2>/dev/null | head -c 1 | wc -c | tr -d ' ' || true) + n=$( + trap '' PIPE + GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_REGISTRY_1_0="$HARNESS_RUN/gst-run-registry.bin" \ + timeout -k 3 "$TIMEOUT" gst-launch-1.0 -q \ + moqsrc name=s url="$URL" broadcast="$broadcast" \ + s.video_0 ! filesink location=/dev/stdout buffer-mode=2 \ + 2>/dev/null | head -c 1 | wc -c | tr -d ' ' || true + ) [[ "${n:-0}" -ge 1 ]] ;; js) @@ -704,7 +796,7 @@ run_cell() { } run_round() { - local pub="$1" broadcast="$2" pub_pid="$3" + local pub="$1" broadcast="$2" pub_pid="$3" from="$4" local pids=() names=() i sub for sub in "${SUB_LIST[@]}"; do if is_broken "$sub"; then @@ -744,11 +836,13 @@ run_round() { echo " INFO publisher '$pub' log:" sed 's/^/ /' "$HARNESS_RUN/pub-$pub.log" 2>/dev/null || true fi - # `harness_reap` retires the entry, so teardown never signals this now-reaped - # (possibly recycled) PID again. + # `stop_publisher` reaps it, retiring the entry, so teardown never signals this + # now-reaped (possibly recycled) PID again. if [[ -n "$pub_pid" ]]; then - harness_reap "$pub_pid" + stop_publisher "$pub_pid" "$pub" fi + # The negative control's subscribers end by timing out, which is its point. + [[ "$NEGATIVE" -eq 1 || "$IDLE_CHECK" -eq 0 ]] || check_idle_outs "$pub" "$from" return 0 } @@ -792,7 +886,7 @@ elif [[ "$NEGATIVE" -eq 1 ]]; then # Negative control: no publisher. Every subscriber must FAIL (time out with # no data), proving the harness can actually report failure. echo "=== negative control: subscribers expect NO data ===" - run_round "none" "interop-missing-$$-$RANDOM.hang" "" + run_round "none" "interop-missing-$$-$RANDOM.hang" "" "$(relay_next_line)" else for pub in "${PUB_LIST[@]}"; do broadcast="interop-${pub}-$$-${RANDOM}.hang" @@ -804,8 +898,9 @@ else overall=1 continue fi + from=$(relay_next_line) start_publisher "$pub" "$broadcast" - run_round "$pub" "$broadcast" "$PUB_PID" + run_round "$pub" "$broadcast" "$PUB_PID" "$from" done # The relay refuses a token on its public rules, and Chromium has to read the close code and