From 01d0b1ec45de3ade895aa481b8d8ea1a2458c54c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 19:23:23 -0700 Subject: [PATCH 1/7] quest: claim quest/m0/ffi-publisher-stall From ee04682957e665e1ee7011c67673df4af351c967 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 20:37:44 -0700 Subject: [PATCH 2/7] test(interop): fail a round when any connection idles out The FFI publisher stall the quest tracked is gone on main: #5088's serve budgets fix it. Disabling the budget and pinning the run to one CPU makes every Go, Python, and C++ publisher idle out mid-round; with it, none do. To keep that class of regression from passing as a slow cell, every client now closes its session on the way out, and each round waits for the relay to log every close, failing on any idle-out: - The relay logs the connection id with its close reason. - Publishers stop by ending their input (the browser on SIGTERM) instead of being killed. - The C++ client shuts its session down instead of cancelling it, the native JS client awaits its close, the browser closes its pages first, and gst-launch ignores SIGPIPE so it stops its pipeline. - moq-c's moq_session_close delivers its terminal callback once the close has reached the peer, so a C app that exits on it closes cleanly. Co-Authored-By: Claude Opus 5.5 --- quest/m0/README.md | 1 - quest/m0/ffi-publisher-stall.md | 60 ----------- rs/moq-c/src/api.rs | 5 +- rs/moq-c/src/session.rs | 7 +- rs/moq-relay/src/relay.rs | 5 +- test/interop/README.md | 7 ++ test/interop/clients/cpp/main.cpp | 10 +- test/interop/clients/js-native/subscribe.ts | 5 +- test/interop/clients/js/driver.ts | 9 +- test/interop/interop.sh | 108 +++++++++++++++++--- 10 files changed, 129 insertions(+), 88 deletions(-) delete mode 100644 quest/m0/ffi-publisher-stall.md diff --git a/quest/m0/README.md b/quest/m0/README.md index 09f37dba68..f48cf27a5e 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -62,7 +62,6 @@ remain and no release waits on them. ## Required - [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.4, 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 - [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 - [qmux credit](/quest/m0/qmux-credit.md) - qmux returns connection credit for dropped and stopped streams and delivers its close frame, on both lines 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..c28b016736 100644 --- a/rs/moq-c/src/session.rs +++ b/rs/moq-c/src/session.rs @@ -120,9 +120,10 @@ 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: the terminal 0 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. + _ = closed.1 => reconnect.clone().close().await.map_err(map_connect_error), + res = Self::report(callback, reconnect.clone()) => res, }; // Deliver one final terminal callback (0 = closed, < 0 = error), then diff --git a/rs/moq-relay/src/relay.rs b/rs/moq-relay/src/relay.rs index 494466321e..cd2ed80bb3 100644 --- a/rs/moq-relay/src/relay.rs +++ b/rs/moq-relay/src/relay.rs @@ -782,14 +782,15 @@ 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"); + tracing::warn!(id, %err, "connection closed"); } }); } diff --git a/test/interop/README.md b/test/interop/README.md index 8fc5d53e08..cca6312de7 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -19,6 +19,13 @@ 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. + 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..b5074b3919 100644 --- a/test/interop/clients/js/driver.ts +++ b/test/interop/clients/js/driver.ts @@ -99,7 +99,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 +172,11 @@ 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 the pages first, which closes their sessions. Closing the browser alone can end + // them without a close, leaving the relay to time them out. + for (const page of browser.contexts().flatMap((context) => context.pages())) { + await page.close().catch(() => {}); + } await browser.close().catch(() => {}); server.stop(); } diff --git a/test/interop/interop.sh b/test/interop/interop.sh index a6c0c93fde..c97ecf1e5a 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -516,8 +516,9 @@ run_publisher() { ;; 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 +536,74 @@ 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 closes its pages 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 + js) 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 + 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, or "open" while it hasn't closed yet. +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]+ err=/) { + id = substr($0, RSTART + 21, RLENGTH - 26) + if (!(id in state)) order[++n] = id + state[id] = substr($0, RSTART + RLENGTH) + } + 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 +707,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 +777,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 +817,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 ]] || check_idle_outs "$pub" "$from" return 0 } @@ -791,7 +866,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" @@ -803,8 +878,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 From 55fab927caf257a33043a1f35c90be31c846eb02 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 22:20:36 -0700 Subject: [PATCH 3/7] fix(moq-tokio): close and abort end a QUIC upgrade still in its handshake When the WebSocket fallback wins the race, the QUIC dial keeps going as an upgrade. Once our SETUP is out, the peer holds that session open, but only the loop task owned it: `close` and `abort` ended the live session and then aborted the task, so the upgrade was dropped later with a bare cancel. An embedder that exits right after them (moq-c's terminal callback, gst's stop) left the peer to time the connection out. The upgrade's session now sits in the connection state from our SETUP until it takes over or is dropped, and `close` and `abort` end it with the live one. Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/src/client.rs | 50 ++++++++++++------ rs/moq-tokio/src/connection.rs | 95 ++++++++++++++++++++++++++++++++-- 2 files changed, 124 insertions(+), 21 deletions(-) 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..f6278085ba 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,13 @@ impl Connection { } } }; - let (session, predecessor) = tokio::join!(session, predecessor); + let upgrade = async move { + if let Some(upgrade) = upgrade { + upgrade.abort(moq_net::Error::Cancel); + upgrade.closed().await; + } + }; + let (session, predecessor, ()) = tokio::join!(session, predecessor, upgrade); Ok(session.and(predecessor)?) } @@ -856,6 +887,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 +1507,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 +1517,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 +2681,56 @@ 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}" + ); + } + /// When WebSocket wins the race but its MoQ handshake fails, the attempt falls back /// to the QUIC dial still pending instead of failing. /// From c080a2f239eec8968120b29031467dcd9bf73aa7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 23:05:03 -0700 Subject: [PATCH 4/7] fix(moq-gst,moq-c): close the client before reporting a session stopped Client::close is what puts each connection's close on the wire. moq-c's terminal callback and gst's stop returned once the connection loop ended, so a process exiting right after them (gst-launch reaching NULL, a C app returning from main) could leave the relay to time the connection out. Interop caught it on CI as a gst subscriber idling out after its cell. Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/src/session.rs | 6 ++-- rs/moq-gst/src/sink/session.rs | 10 +++++- rs/moq-gst/src/source/imp.rs | 63 +++++++++++++++++++++++----------- 3 files changed, 56 insertions(+), 23 deletions(-) diff --git a/rs/moq-c/src/session.rs b/rs/moq-c/src/session.rs index c28b016736..50ee51e6da 100644 --- a/rs/moq-c/src/session.rs +++ b/rs/moq-c/src/session.rs @@ -120,11 +120,13 @@ impl Session { let _consume = consume; let res = tokio::select! { - // close() requested: the terminal 0 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. + // 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 }, From 0b7b31aed6e736019ef4364364e3a4e6533d1b1f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 05:54:44 -0700 Subject: [PATCH 5/7] test(interop): address review on the idle-out check - moq-tokio: bound `close`'s wait on an in-flight upgrade to one second, and cover `close` mid-upgrade with a test. - The native JS publisher (the wire-compat js-publish lanes) closes on SIGTERM and leads its group, like the browser driver, instead of being killed. - The idle-out check covers runs built from this checkout. The wire-compat lanes swap in released relays and clients, which this tree can't hold to it. - The browser driver removes its players and waits for their transports to close before closing the browser. - stop_publisher warns when a publisher outlives its stop, and the log patterns name the relay code they follow. Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/src/connection.rs | 43 +++++++++++++++++++++++++++- test/interop/README.md | 4 ++- test/interop/clients/js/driver.ts | 22 ++++++++++++-- test/interop/clients/js/src/setup.ts | 2 ++ test/interop/compat/client.ts | 2 ++ test/interop/interop.sh | 25 ++++++++++++---- 6 files changed, 88 insertions(+), 10 deletions(-) diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index f6278085ba..d9915fb07c 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -814,7 +814,8 @@ impl Connection { let upgrade = async move { if let Some(upgrade) = upgrade { upgrade.abort(moq_net::Error::Cancel); - upgrade.closed().await; + // 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); @@ -2731,6 +2732,46 @@ mod tests { ); } + /// 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 b6b152e09b..9e7ed9b0fa 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -24,7 +24,9 @@ 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. +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 diff --git a/test/interop/clients/js/driver.ts b/test/interop/clients/js/driver.ts index b5074b3919..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: { @@ -172,9 +189,10 @@ 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 the pages first, which closes their sessions. Closing the browser alone can end - // them without a close, leaving the relay to time them out. + // 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(() => {}); 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 6ae1927863..ba7a1ee807 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,7 +520,8 @@ 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 @@ -538,17 +547,20 @@ start_publisher() { # 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 closes its pages on SIGTERM. Whatever is still running after the -# wait is killed, and its connection then fails the round as an idle-out. +# 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 - js) kill -TERM -- -"$pid" 2>/dev/null || true ;; + 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" } @@ -563,7 +575,8 @@ relay_log_since() { } # Print each relay connection that appears from log line FROM on, with how it closed: the -# relay's close error, or "open" while it hasn't closed yet. +# relay's close error, or "open" while it hasn't closed yet. The patterns follow the relay's +# `conn{id=...}` span and its `connection closed` warning (`rs/moq-relay/src/relay.rs`). relay_connections() { relay_log_since "$1" | awk ' match($0, /conn\{id=[0-9]+ /) { @@ -823,7 +836,7 @@ run_round() { stop_publisher "$pub_pid" "$pub" fi # The negative control's subscribers end by timing out, which is its point. - [[ "$NEGATIVE" -eq 1 ]] || check_idle_outs "$pub" "$from" + [[ "$NEGATIVE" -eq 1 || "$IDLE_CHECK" -eq 0 ]] || check_idle_outs "$pub" "$from" return 0 } From 7d024132c2e27f1653ccd20c890a42d1ea1dc15e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:17:53 -0700 Subject: [PATCH 6/7] test(interop): stop the browser publisher without killing Chromium first Playwright closes the browser on SIGTERM by default, and the harness sent the signal to the whole group, Chromium included. Either way the browser ended before the driver removed its players, so its session went without a close and the relay timed it out (2 of 12 browser-publisher rounds locally). The harness now signals only the driver, and the launch leaves SIGTERM to it. Co-Authored-By: Claude Opus 5.5 --- test/interop/clients/js/harness.ts | 4 +++- test/interop/interop.sh | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) 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/interop.sh b/test/interop/interop.sh index ba7a1ee807..9f3a99a9c1 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -552,7 +552,9 @@ start_publisher() { stop_publisher() { local pid="$1" lang="$2" deadline=$((SECONDS + 10)) case "$lang" in - js | js-native) kill -TERM -- -"$pid" 2>/dev/null || true ;; + # 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 From 04ea2014a5e62ccdf7cbfff0cea3941afff802ee Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 06:58:13 -0700 Subject: [PATCH 7/7] test(interop): pair every relay connection with its end - The relay logs `connection closed` when a lease or shutdown ends a session too, not only on an error, so the idle-out check never waits out a clean close it can't see. - The check counts only connections opened in its round, so a straggler from an earlier round closing late isn't blamed on this one. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/relay.rs | 7 +++++-- rs/moq-relay/src/uring.rs | 7 +++++-- test/interop/interop.sh | 16 ++++++++++------ 3 files changed, 20 insertions(+), 10 deletions(-) diff --git a/rs/moq-relay/src/relay.rs b/rs/moq-relay/src/relay.rs index cd2ed80bb3..7ae7335ba2 100644 --- a/rs/moq-relay/src/relay.rs +++ b/rs/moq-relay/src/relay.rs @@ -789,8 +789,11 @@ async fn serve_listening( .with_sessions(sessions.clone()); tokio::spawn(async move { - if let Err(err) = conn.run().await { - tracing::warn!(id, %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/test/interop/interop.sh b/test/interop/interop.sh index 9f3a99a9c1..27886383ab 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -577,18 +577,22 @@ relay_log_since() { } # Print each relay connection that appears from log line FROM on, with how it closed: the -# relay's close error, or "open" while it hasn't closed yet. The patterns follow the relay's -# `conn{id=...}` span and its `connection closed` warning (`rs/moq-relay/src/relay.rs`). +# 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]+ err=/) { - id = substr($0, RSTART + 21, RLENGTH - 26) - if (!(id in state)) order[++n] = id - state[id] = substr($0, RSTART + RLENGTH) + 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]] } '