test(interop): fail a round when any connection idles out

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

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 {

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

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<Status>,
 	/// 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;
+		});
 	}
 }
 

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<super::MoqSrc>) -> Result<Self> {
-		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<F, Fut>(connection: moq_tokio::Connection, element: glib::WeakRef<super::MoqSrc>, session: F) -> Self
+	fn spawn<F, Fut>(
+		client: moq_tokio::Client,
+		connection: moq_tokio::Connection,
+		element: glib::WeakRef<super::MoqSrc>,
+		session: F,
+	) -> Self
 	where
 		F: FnOnce(watch::Receiver<bool>) -> Fut,
 		Fut: Future<Output = Result<()>> + Send + 'static,
@@ -142,6 +155,7 @@ impl SessionController {
 			shutdown: shutdown_tx,
 			join,
 			connection,
+			client,
 		}
 	}
 
@@ -174,20 +188,28 @@ 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 {
 				gst::warning!(CAT, "session task ended with error: {err:?}");
 			}
 			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 },

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"),
 			}
 		});
 	}

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"),
 		}
 	});
 

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<moq_net::Session>>;
 
 /// 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<Handshake>>),
 	/// 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<crate::Result<Step>> {
 		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)))
 			}
 		}
 	}