From 8ab9ba7fad21e329b7b6a070766d1dfe52070fd0 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:48:54 -0700 Subject: [PATCH 1/3] chore: claim shared connection lifecycle From d077b5850d902e5cb54ba658364fe330e518a291 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:57:50 -0700 Subject: [PATCH 2/3] fix(relay): share connection lifecycle across runtimes Co-Authored-By: GPT-6 --- quest/m2/README.md | 1 - quest/m2/uring-connection-log.md | 57 ------ rs/moq-relay/src/connection.rs | 280 +++++++++++++++++++--------- rs/moq-relay/src/uring.rs | 129 ++++--------- rs/moq-relay/tests/runtime_uring.rs | 53 ++++++ rs/moq-relay/tests/smoke.rs | 44 +---- rs/moq-relay/tests/support/logs.rs | 44 +++++ 7 files changed, 332 insertions(+), 276 deletions(-) delete mode 100644 quest/m2/uring-connection-log.md create mode 100644 rs/moq-relay/tests/support/logs.rs diff --git a/quest/m2/README.md b/quest/m2/README.md index b41215ace05..76f47927f3f 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -118,7 +118,6 @@ ranked just below deterministic TS legs, which it requires. - [Archive recovery listing](/quest/m2/archive-recovery-listing.md) - a resumed DVR lists what changed since its checkpoint, not every stored group - [IETF on the ring](/quest/m2/uring-ietf.md) - the io_uring workers serve moq-transport sessions too, so a uring relay drops no client protocol - [Dropped uring session closes](/quest/m2/uring-drop-close.md) - a moq-uring session dropped without close() closes its connection -- [One connection lifecycle](/quest/m2/uring-connection-log.md) - every accept path, io_uring included, logs a connection's span and close from one owner, so Interop's accounting works on the ring - [Relay io_uring packages](/quest/m2/relay-io-uring-package.md) - Linux relay packages ship io_uring once the ring is on par with tokio - [Opus implementation](/quest/m2/audio-opus-backend.md) - compare Opus codec quality, CPU, build cost, and the loss recovery each backend offers - [Latency ledger](/quest/m2/latency-ledger.md) - a viewer reports its share of playback delay stage by stage: jitter buffer, decode, render, and device diff --git a/quest/m2/uring-connection-log.md b/quest/m2/uring-connection-log.md deleted file mode 100644 index 775ff2106cf..00000000000 --- a/quest/m2/uring-connection-log.md +++ /dev/null @@ -1,57 +0,0 @@ -# [S] One owner logs every relay connection's lifecycle - -## Goal - -Every relay accept path, the io_uring workers included, logs a connection's -`negotiated` and `connection closed` lines under the `moq_relay::connection` -target inside its `conn{id=N}` span, from one place. A connection that is -refused, fails its handshake, or panics still logs its close. The Interop -harness's accounting (`test/interop/relay-log.sh`), which filters to that -target and fails a round whose opened connection never closes, then works -unchanged on a relay started with `--runtime-io-uring`. - -## Plan - -From #5602, which moved the close log into `moq_relay::Connection::run` -(`rs/moq-relay/src/connection.rs`) so QUIC and WebSocket sessions pair their -span with their close under one filter. Checked on main at `de6c27302`: the -io_uring accept path (`serve_connection` in `rs/moq-relay/src/uring.rs`) -does not go through `Connection::run`. It logs `negotiated` and -`connection closed` under `moq_relay::uring`, with the id as a field and no -`conn` span. A panic in its supervise task is isolated and dropped without a -close line. Interop does not run the ring today, so nothing fails yet. - -Root problem: the ring path re-implements the connection lifecycle that -`Connection` owns: admission (the LAN credential branch, auth on the shared -runtime, refusal recording), limits, `lease.preparing`, supervision, and the -logs around them. The logs are only the visible drift. The copy exists -because the handshake and session driver must stay on the `!Send` worker -while admission and supervision run on the shared runtime. - -Recommendations: - -- Make one owner of the lifecycle that both accept paths call, so the span, - the id, the `negotiated` line, and the close (including a panic) cannot - drift again. Prefer sharing admission and supervision with `Connection` - over a logging helper that only hides the duplication; if part of the - admission cannot be shared across the worker boundary, record why. -- The ring connection spans two runtimes, so the `conn` span has to follow - both its worker future and its shared-runtime task. -- Regressions, logged with only the `moq_relay::connection` target as the - harness filters: - - An io_uring session shows its `conn` span and its close, like - `relay_websocket_logs_its_close_under_a_cluster_id` in - `rs/moq-relay/tests/smoke.rs`. It belongs in `tests/runtime_uring.rs`, - which skips loudly below the kernel floor. - - A panic inside the lifecycle owner, forced on the ring's - shared-runtime supervision, still logs that connection's close. This is - the concrete missing close today: `serve_connection` isolates the - supervise task's panic and drops the result. It must fail without the fix. - -Public API: none expected. Wire: none. - -## Related - -- [IETF on the ring](/quest/m2/uring-ietf.md) - also touches the ring's accept path -- [Relay io_uring packages](/quest/m2/relay-io-uring-package.md) - ships the ring, whose operators read these logs -- [Session outcomes](/quest/m1/session-outcomes.md) - counts refusals and ends on both paths, from the same duplicated admission diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index 6c8425e773e..22eb2c28fa4 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -2,15 +2,16 @@ use crate::{auth, cluster, refusals::Refusal}; use axum::http; use moq_tokio::server::Request; +use tracing::Instrument; /// An error carrying the HTTP status to send when closing the request. /// /// Used only on the pre-accept auth path so the caller can close once with /// the right code instead of sprinkling close/return at each failure site. -struct StatusError { - status: http::StatusCode, +pub(crate) struct StatusError { + pub(crate) status: http::StatusCode, refusal: Refusal, - source: anyhow::Error, + pub(crate) source: anyhow::Error, } impl From for StatusError { @@ -29,18 +30,18 @@ impl From for StatusError { /// [`run`](Self::run) to admit the request through its lease, wire up /// publish/subscribe origins, and serve the session until it closes. pub struct Connection { - /// A numeric identifier for logging. - id: u64, - /// The transport request to accept or reject. request: Request, - /// The cluster state used to resolve origins. + lifecycle: Lifecycle, +} + +/// Shared owner of admission, supervision and the connection's log span. +#[derive(Clone)] +pub(crate) struct Lifecycle { + id: u64, + span: tracing::Span, cluster: cluster::Cluster, - /// Where the session's grant comes from. auth: auth::Auth, - /// Relay-wide shutdown broadcast: when it fires, the session is drained with - /// a GOAWAY instead of being cut off. shutdown: crate::shutdown::Observer, - /// Live sessions on this node, so an operator can list and nudge this one. sessions: Option, } @@ -49,72 +50,62 @@ impl Connection { /// its grant through `auth`. pub fn new(request: Request, cluster: cluster::Cluster, auth: auth::Auth) -> Self { Self { - id: 0, + lifecycle: Lifecycle::new(cluster, auth, 0, request.remote_addr()), request, - cluster, - auth, - shutdown: crate::shutdown::Observer::disabled(), - sessions: None, } } /// Set the identifier this connection logs under. Defaults to 0. pub fn with_id(mut self, id: u64) -> Self { - self.id = id; + self.lifecycle.id = id; + self.lifecycle.span = tracing::info_span!( + "conn", + id, + remote = self.request.remote_addr().map(tracing::field::display), + session = tracing::field::Empty + ); self } /// Attach the relay-wide shutdown broadcast so the session drains with a /// GOAWAY when it fires. Without it the session is cut off on process exit. pub fn with_shutdown(mut self, shutdown: crate::shutdown::Observer) -> Self { - self.shutdown = shutdown; + self.lifecycle = self.lifecycle.with_shutdown(shutdown); self } /// Register this session in the node's live table after it is admitted. /// Without it the session is served but cannot be listed or nudged. pub fn with_sessions(mut self, sessions: crate::session::Registry) -> Self { - self.sessions = Some(sessions); + self.lifecycle = self.lifecycle.with_sessions(sessions); self } /// Admits and serves this connection until it closes, then logs the close. A panic /// ends only this connection, as an error. - #[tracing::instrument("conn", skip_all, fields(id = self.id, remote = self.request.remote_addr().map(tracing::field::display), session = tracing::field::Empty))] pub async fn run(self) { - let id = self.id; - // Boxed: debug builds copy the session's large future through each wrapper's - // stack frame, which overflows a test thread. - let result = moq_tokio::panic::isolate(Box::pin(self.serve())) - .await - .unwrap_or_else(|| Err(anyhow::anyhow!("connection panicked"))); - // Logged here, under the same target as the span's other lines, so every - // transport's `conn{id=...}` pairs with its close under one log filter. - match result { - Ok(()) => tracing::info!(id, "connection closed"), - Err(err) => tracing::warn!(id, %err, "connection closed"), - } + let Self { request, lifecycle } = self; + lifecycle.run(Self::serve(request, &lifecycle)).await; } - async fn serve(self) -> anyhow::Result<()> { + async fn serve(request: Request, lifecycle: &Lifecycle) -> anyhow::Result<()> { // Abandoned at the drain deadline, so the deadline bounds this task and its // socket like it bounds a session. - let admitted = self.shutdown.admit(self.admit()).await; + let admitted = lifecycle.admit(auth::request_for(&lifecycle.auth, &request)).await; let (admitted, registration) = match admitted { Ok(admitted) => admitted, Err(err) => { - self.cluster.refusals.record(err.refusal); let reject = match err.status { http::StatusCode::UNAUTHORIZED => moq_tokio::server::Reject::Unauthorized, http::StatusCode::FORBIDDEN => moq_tokio::server::Reject::Forbidden, status => moq_tokio::server::Reject::App(status.as_u16()), }; - let _ = self.request.reject(reject).await; + let _ = request.reject(reject).await; return Err(err.source); } }; - let transport = self.request.transport(); + let transport = request.transport(); let cluster::Admitted { lease, publisher, @@ -129,7 +120,7 @@ impl Connection { // // moq-net defaults the unset side to a fresh no-op origin, which is fine for a // publish-only or subscribe-only session. - let mut request = self.request.with_stats(stats).with_limits(self.cluster.limits(&lease)); + let mut request = request.with_stats(stats).with_limits(lifecycle.cluster.limits(&lease)); if let Some(subscriber) = subscriber { request = request.with_publisher(subscriber); } @@ -138,36 +129,127 @@ impl Connection { } let lease = lease.preparing(request.auth(), request.role())?; // Bounded too: a peer that stalls the SETUP reply must not outlive the relay. - let (session, driver) = self + let (session, driver) = lifecycle .shutdown .until_deadline(request.ok()) .await .ok_or_else(|| anyhow::anyhow!("the SETUP reply was still pending at the drain deadline"))??; - tracing::info!(version = %session.version(), %transport, "negotiated"); - - let result = supervise(session, lease, self.shutdown.clone(), registration).await; + let result = lifecycle.supervise(session, lease, registration, transport).await; driver.await??; result } +} - /// Admit this connection. Any failure is returned as a [`StatusError`] so - /// [`run`] can close the request with the mapped HTTP status exactly once. - /// - /// Every transport goes through the same lease; the request the server sees - /// carries what the transport knows. A LAN mesh dial is the one exception: its - /// credential is a secret the relay minted for itself, checked locally. - async fn admit(&self) -> Result<(cluster::Admitted, Option), StatusError> { - // Checked first so a `/.cluster` request is never routed through the public - // grant, and a relay without LAN discovery refuses it instead of treating the - // path as a broadcast root. - if cluster::Cluster::is_lan_path(self.request.path()) { - return self.admit_lan(); +impl Lifecycle { + pub(crate) fn new( + cluster: cluster::Cluster, + auth: auth::Auth, + id: u64, + remote: Option, + ) -> Self { + Self { + id, + span: tracing::info_span!( + "conn", + id, + remote = remote.map(tracing::field::display), + session = tracing::field::Empty + ), + cluster, + auth, + shutdown: crate::shutdown::Observer::disabled(), + sessions: None, } + } + + pub(crate) fn with_shutdown(mut self, shutdown: crate::shutdown::Observer) -> Self { + self.shutdown = shutdown; + self + } - let request = auth::request_for(&self.auth, &self.request); - tracing::Span::current().record("session", &request.id); - if self.request.peer_identity().is_some() { + pub(crate) fn with_sessions(mut self, sessions: crate::session::Registry) -> Self { + self.sessions = Some(sessions); + self + } + + pub(crate) async fn run(&self, task: impl std::future::Future>) { + async { + // Boxing keeps large protocol futures off nested debug-build stack frames. + let result = moq_tokio::panic::isolate(Box::pin(task)) + .await + .unwrap_or_else(|| Err(anyhow::anyhow!("connection panicked"))); + match result { + Ok(()) => tracing::info!(id = self.id, "connection closed"), + Err(err) => tracing::warn!(id = self.id, %err, "connection closed"), + } + } + .instrument(self.span.clone()) + .await + } + + /// Keep the connection span across a worker-to-shared-runtime handoff and return panics to its owner. + #[cfg(all(target_os = "linux", feature = "_uring"))] + pub(crate) async fn shared( + &self, + runtime: &tokio::runtime::Handle, + task: impl std::future::Future + Send + 'static, + ) -> anyhow::Result { + use anyhow::Context as _; + runtime + .spawn(moq_tokio::panic::isolate(task).instrument(self.span.clone())) + .await + .context("connection task failed")? + .ok_or_else(|| anyhow::anyhow!("connection task panicked")) + } + + pub(crate) async fn admit( + &self, + request: moq_auth::Request, + ) -> Result<(cluster::Admitted, Option), StatusError> { + self.span.record("session", &request.id); + let admitted = self.shutdown.admit(self.authorize(request)).await; + if let Err(err) = &admitted { + self.cluster.refusals.record(err.refusal); + } + admitted + } + + async fn authorize( + &self, + request: moq_auth::Request, + ) -> Result<(cluster::Admitted, Option), StatusError> { + if cluster::Cluster::is_lan_path(&request.path) { + let Some(presented) = cluster::Cluster::lan_credential(&request.path) else { + return Err(StatusError { + status: http::StatusCode::FORBIDDEN, + refusal: Refusal::Lan, + source: anyhow::anyhow!("LAN peer did not present a membership proof"), + }); + }; + match self.cluster.verify_lan_credential(presented) { + Some(true) => { + tracing::info!("accepted LAN peer"); + let lease = self.auth.admit_fixed("/", self.cluster.lan_peer_grant()); + return Ok((self.cluster.scope(lease, &request)?, None)); + } + Some(false) => { + return Err(StatusError { + status: http::StatusCode::FORBIDDEN, + refusal: Refusal::Lan, + source: anyhow::anyhow!("LAN peer did not present this listener's membership proof"), + }); + } + None => { + return Err(StatusError { + status: http::StatusCode::FORBIDDEN, + refusal: Refusal::Lan, + source: anyhow::anyhow!("/.cluster request refused: LAN discovery is not enabled"), + }); + } + } + } + if request.tls.is_some() { tracing::debug!("client certificate verified; reported to the auth server"); } let admitted = self.cluster.admit_session(&self.auth, request.clone()).await?; @@ -175,33 +257,15 @@ impl Connection { Ok((admitted, registration)) } - /// Authorize a `/.cluster/` dial against the live LAN advertisement. - fn admit_lan(&self) -> Result<(cluster::Admitted, Option), StatusError> { - let Some(presented) = cluster::Cluster::lan_credential(self.request.path()) else { - return Err(StatusError { - status: http::StatusCode::FORBIDDEN, - refusal: Refusal::Lan, - source: anyhow::anyhow!("LAN peer did not present a membership proof"), - }); - }; - match self.cluster.verify_lan_credential(presented) { - Some(true) => { - tracing::info!("accepted LAN peer"); - let lease = self.auth.admit_fixed("/", self.cluster.lan_peer_grant()); - let request = auth::request_for(&self.auth, &self.request); - Ok((self.cluster.scope(lease, &request)?, None)) - } - Some(false) => Err(StatusError { - status: http::StatusCode::FORBIDDEN, - refusal: Refusal::Lan, - source: anyhow::anyhow!("LAN peer did not present this listener's membership proof"), - }), - None => Err(StatusError { - status: http::StatusCode::FORBIDDEN, - refusal: Refusal::Lan, - source: anyhow::anyhow!("/.cluster request refused: LAN discovery is not enabled"), - }), - } + pub(crate) async fn supervise( + &self, + session: moq_net::Session, + lease: auth::Lease, + registration: Option, + transport: impl std::fmt::Display, + ) -> anyhow::Result<()> { + tracing::info!(version = %session.version(), %transport, "negotiated"); + supervise(session, lease, self.shutdown.clone(), registration).await } } @@ -275,3 +339,51 @@ pub(crate) fn session_bytes(session: &moq_net::Session) -> moq_auth::Bytes { received: stats.bytes_received.unwrap_or_default(), } } + +#[cfg(all(test, target_os = "linux", feature = "_uring"))] +#[path = "../tests/support/logs.rs"] +mod logs; + +#[cfg(all(test, target_os = "linux", feature = "_uring"))] +mod tests { + use super::*; + use tracing_subscriber::layer::SubscriberExt; + + use super::logs; + + /// A panic in the same shared-runtime handoff used for ring supervision reaches the close owner. + #[tokio::test(start_paused = true)] + async fn shared_supervision_panic_logs_the_connection_close() { + let logs = logs::Logs::default(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_writer(logs.clone()) + .finish() + .with( + tracing_subscriber::filter::Targets::new().with_target("moq_relay::connection", tracing::Level::INFO), + ); + let _logging = tracing::subscriber::set_default(subscriber); + let cluster = cluster::Cluster::new(cluster::Options::default()).unwrap(); + let (auth, _admissions) = auth::Auth::embedded("test"); + let lifecycle = Lifecycle::new(cluster, auth, 42, None); + lifecycle + .run(async { + lifecycle + .shared(&tokio::runtime::Handle::current(), async { + panic!("injected supervision panic"); + #[allow(unreachable_code)] + Ok::<_, anyhow::Error>(()) + }) + .await? + }) + .await; + logs.wait_for("connection closed").await; + let text = logs.text(); + assert!(text.contains("conn{id=42}"), "{text}"); + assert!( + text.contains("connection closed id=42 err=connection task panicked"), + "{text}" + ); + assert_eq!(text.matches("connection closed").count(), 1, "{text}"); + } +} diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 5df9462027d..4aa5dbdbdf1 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -21,8 +21,9 @@ use std::task::Poll; use anyhow::Context as _; use futures::FutureExt as _; +use tracing::Instrument; -use crate::{auth, cluster, refusals::Refusal, shutdown}; +use crate::{auth, cluster, connection::Lifecycle, shutdown}; /// A stop signal a worker parks on, wakeable from the shared runtime. #[derive(Default)] @@ -569,12 +570,15 @@ fn serve_worker(spawn: Spawn) -> bool { let id = serve.cluster.next_connection_id(); let conn_handle = accept_handle.clone(); let serve = serve.clone(); - let connection = async move { - if let Err(err) = serve_connection(conn_handle, conn, id, serve).await { - tracing::warn!(id, %err, "connection closed"); - } - }; - accept_handle.spawn(moq_tokio::panic::isolate(connection).map(drop)); + let lifecycle = + Lifecycle::new(serve.cluster.clone(), serve.auth.clone(), id, Some(conn.remote_addr())) + .with_shutdown(serve.shutdown.clone()) + .with_sessions(serve.sessions.clone()); + accept_handle.spawn(async move { + lifecycle + .run(serve_connection(conn_handle, conn, serve, &lifecycle)) + .await; + }); } Err(err) => { let _ = accept_failures @@ -636,8 +640,8 @@ impl Deadline { async fn serve_connection( handle: moq_uring::Handle, conn: moq_uring::quic::Connection, - id: u64, serve: Serve, + lifecycle: &Lifecycle, ) -> anyhow::Result<()> { use web_transport_trait::poll::Session as _; @@ -740,75 +744,20 @@ async fn serve_connection( auth_request.alpn = alpn.clone(); auth_request.role = request.role().and_then(crate::auth::role); auth_request.tls = identity.as_ref().and_then(crate::auth::peer); - let mut registration = None; - let admitted = if cluster::Cluster::is_lan_path(&path) { - let lease = match cluster::Cluster::lan_credential(&path) { - Some(presented) => match serve.cluster.verify_lan_credential(presented) { - Some(true) => serve.auth.admit_fixed("/", serve.cluster.lan_peer_grant()), - Some(false) => { - serve.cluster.refusals.record(Refusal::Lan); - request.close(moq_net::Error::Unauthorized); - anyhow::bail!("LAN peer did not present this listener's membership proof"); - } - None => { - serve.cluster.refusals.record(Refusal::Lan); - request.close(moq_net::Error::Unauthorized); - anyhow::bail!("/.cluster request refused: LAN discovery is not enabled"); + let admission = lifecycle.clone(); + let (admitted, registration) = match lifecycle + .shared(&serve.tokio, async move { admission.admit(auth_request).await }) + .await? + { + Ok(admitted) => admitted, + Err(err) => { + request.close(match err.status { + axum::http::StatusCode::UNAUTHORIZED | axum::http::StatusCode::FORBIDDEN => { + moq_net::Error::Unauthorized } - }, - None => { - serve.cluster.refusals.record(Refusal::Lan); - request.close(moq_net::Error::Unauthorized); - anyhow::bail!("LAN peer did not present a membership proof"); - } - }; - match serve.cluster.scope(lease, &auth_request) { - Ok(admitted) => admitted, - Err(err) => { - serve.cluster.refusals.record((&err).into()); - request.close(moq_net::Error::Unauthorized); - return Err(err.into()); - } - } - } else { - if identity.is_some() { - tracing::debug!(id, "client certificate verified; reported to the auth server"); - } - let auth = serve.auth.clone(); - let cluster = serve.cluster.clone(); - let sessions = serve.sessions.clone(); - let shutdown = serve.shutdown.clone(); - match serve - .tokio - .spawn(async move { - // Abandoned at the drain deadline: worker teardown only detaches this task. - let admitted = shutdown - .admit(cluster.admit_session(&auth, auth_request.clone())) - .await?; - Ok::<_, crate::auth::Error>((admitted, sessions.register(auth_request))) - }) - .await - .context("auth task failed")? - { - Ok((admitted, registered)) => { - registration = Some(registered); - admitted - } - Err(err) => { - serve.cluster.refusals.record((&err).into()); - // The status is what separates "your credential is bad" from "the - // auth server is down". Collapsing both into Unauthorized tells a - // client to stop reconnecting through an outage it could have - // waited out. - let status = axum::http::StatusCode::from(&err); - request.close(match status { - axum::http::StatusCode::UNAUTHORIZED | axum::http::StatusCode::FORBIDDEN => { - moq_net::Error::Unauthorized - } - other => moq_net::Error::App(other.as_u16()), - }); - return Err(anyhow::Error::new(err).context("authentication failed")); - } + other => moq_net::Error::App(other.as_u16()), + }); + return Err(err.source); } }; @@ -835,24 +784,16 @@ async fn serve_connection( err => tracing::debug!(%err, "session driver ended"), } }; - handle.spawn(moq_tokio::panic::isolate(driver).map(drop)); - - tracing::info!(id, version = %session.version(), %transport, "negotiated"); - - // The session handle is Send + Sync however its transport is driven, so - // its lifecycle (credential expiry, GOAWAY drain) lives with the timers - // and the shutdown broadcast on the shared runtime. - let shutdown = serve.shutdown.clone(); - serve.tokio.spawn(moq_tokio::panic::isolate(async move { - 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"), - } - })); - - Ok(()) + handle.spawn(moq_tokio::panic::isolate(driver).in_current_span().map(drop)); + + // The driver stays on this worker; admission and supervision share the same + // lifecycle owner on the runtime that owns auth's HTTP client and timers. + let supervisor = lifecycle.clone(); + lifecycle + .shared(&serve.tokio, async move { + supervisor.supervise(session, lease, registration, transport).await + }) + .await? } #[cfg(test)] diff --git a/rs/moq-relay/tests/runtime_uring.rs b/rs/moq-relay/tests/runtime_uring.rs index 5d66675801d..8033b99fb10 100644 --- a/rs/moq-relay/tests/runtime_uring.rs +++ b/rs/moq-relay/tests/runtime_uring.rs @@ -872,3 +872,56 @@ async fn stall_after_hello(server: SocketAddr) -> SocketAddr { }); addr } + +#[path = "support/logs.rs"] +mod logs; + +/// The harness keeps only the connection target, including its span on worker and shared tasks. +#[tokio::test] +async fn uring_sessions_log_their_close_under_a_connection_span() { + use tracing_subscriber::layer::SubscriberExt; + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + if !supported() { + return; + } + let logs = logs::Logs::default(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_writer(logs.clone()) + .finish() + .with(tracing_subscriber::filter::Targets::new().with_target("moq_relay::connection", tracing::Level::INFO)); + // Worker threads need the same dispatcher as the shared runtime. + tracing::subscriber::set_global_default(subscriber).expect("test owns the global subscriber"); + let dir = tempfile::tempdir().unwrap(); + let (cert, key) = certificate(dir.path()); + let relay = Relay::load(uring_config(&cert, &key)) + .await + .unwrap() + .with_signals(false); + let port = relay.addr().unwrap().port(); + let trigger = relay.shutdown_trigger().clone(); + let running = tokio::spawn(relay.run()); + for scheme in ["moql", "https"] { + let url = format!("{scheme}://127.0.0.1:{port}/logs").parse().unwrap(); + let connection = connect(client(), url).await; + tokio::time::timeout(TIMEOUT, logs.wait_for("negotiated")) + .await + .expect("negotiated log missing"); + let _ = connection.close().await; + } + trigger.start(); + tokio::time::timeout(TIMEOUT, running).await.unwrap().unwrap().unwrap(); + let text = logs.text(); + let negotiated: Vec<_> = text.lines().filter(|line| line.contains("negotiated")).collect(); + let closed: Vec<_> = text.lines().filter(|line| line.contains("connection closed")).collect(); + assert_eq!(negotiated.len(), 2, "{text}"); + assert_eq!(closed.len(), 2, "{text}"); + for id in 0..2 { + for lines in [&negotiated, &closed] { + assert!( + lines.iter().any(|line| line.contains(&format!("conn{{id={id} "))), + "missing conn {id}: {text}" + ); + } + } +} diff --git a/rs/moq-relay/tests/smoke.rs b/rs/moq-relay/tests/smoke.rs index 8844810d889..1ee971fd5b8 100644 --- a/rs/moq-relay/tests/smoke.rs +++ b/rs/moq-relay/tests/smoke.rs @@ -214,46 +214,9 @@ async fn relay_websocket_round_trip_uses_newest_version() { web_handle.abort(); } -/// Relay log output as text, waking every waiter on each write. -#[derive(Clone, Default)] -struct Logs { - text: std::sync::Arc>, - written: std::sync::Arc, -} - -impl Logs { - /// Wait until the output contains `needle`. - async fn wait_for(&self, needle: &str) { - loop { - // Created before the check, so a write in between still wakes it. - let written = self.written.notified(); - if self.text.lock().unwrap().contains(needle) { - return; - } - written.await; - } - } -} - -impl std::io::Write for Logs { - fn write(&mut self, data: &[u8]) -> std::io::Result { - self.text.lock().unwrap().push_str(&String::from_utf8_lossy(data)); - self.written.notify_waiters(); - Ok(data.len()) - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } -} - -impl tracing_subscriber::fmt::MakeWriter<'_> for Logs { - type Writer = Self; - - fn make_writer(&self) -> Self { - self.clone() - } -} +#[path = "support/logs.rs"] +mod logs; +use logs::Logs; /// A WebSocket session takes its id from the cluster-wide counter and logs its close, like /// a QUIC one, so the interop harness can pair its `conn{id=N}` span with that close. @@ -300,6 +263,7 @@ async fn relay_websocket_logs_its_close_under_a_cluster_id() { .await .expect("the WebSocket session's close was not logged"); + assert!(logs.text().contains("connection closed")); web_handle.abort(); } diff --git a/rs/moq-relay/tests/support/logs.rs b/rs/moq-relay/tests/support/logs.rs new file mode 100644 index 00000000000..c0e5dc7f07f --- /dev/null +++ b/rs/moq-relay/tests/support/logs.rs @@ -0,0 +1,44 @@ +/// Relay log output as text, waking every waiter on each write. +#[derive(Clone, Default)] +pub struct Logs { + text: std::sync::Arc>, + written: std::sync::Arc, +} + +impl Logs { + pub fn text(&self) -> String { + self.text.lock().unwrap().clone() + } + + /// Wait until the output contains `needle`. + pub async fn wait_for(&self, needle: &str) { + loop { + // Created before the check, so a write in between still wakes it. + let written = self.written.notified(); + if self.text.lock().unwrap().contains(needle) { + return; + } + written.await; + } + } +} + +impl std::io::Write for Logs { + fn write(&mut self, data: &[u8]) -> std::io::Result { + self.text.lock().unwrap().push_str(&String::from_utf8_lossy(data)); + self.written.notify_waiters(); + Ok(data.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl tracing_subscriber::fmt::MakeWriter<'_> for Logs { + type Writer = Self; + + fn make_writer(&self) -> Self { + self.clone() + } +} From a1bdd78dabbda0b9131fd9a91ea8256a1d5108b6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 22:55:19 -0700 Subject: [PATCH 3/3] fix(relay): log a ring session's close on the shared runtime A worker stopped at the end of a drain dropped the task that would log a still-draining session's close. Hand supervision and its close to the shared runtime once the worker's half starts the session, and cover a drain with sessions open in the ring log test. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/connection.rs | 67 +++++++++++++------ rs/moq-relay/src/uring.rs | 23 +++---- .../tests/integration/runtime_uring.rs | 21 ++++-- 3 files changed, 74 insertions(+), 37 deletions(-) diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index 45415cb43cc..f1396ce95ea 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -173,19 +173,47 @@ impl Lifecycle { self } + /// Run `task` in this connection's span, then log its close; a panic closes it as an error. pub(crate) async fn run(&self, task: impl std::future::Future>) { - async { - // Boxing keeps large protocol futures off nested debug-build stack frames. - let result = moq_tokio::panic::isolate(Box::pin(task)) - .await - .unwrap_or_else(|| Err(anyhow::anyhow!("connection panicked"))); - match result { - Ok(()) => tracing::info!(id = self.id, "connection closed"), - Err(err) => tracing::warn!(id = self.id, %err, "connection closed"), + let result = self.isolate(task).await; + self.closed(result); + } + + /// Run the ring worker's half, then supervise the started session on `runtime`, which logs + /// the close there: a worker stopped at shutdown drops its tasks, and the line with them. + #[cfg(all(target_os = "linux", feature = "_uring"))] + pub(crate) async fn start( + &self, + runtime: &tokio::runtime::Handle, + start: impl std::future::Future>, + ) where + S: std::future::Future> + Send + 'static, + { + match self.isolate(start).await { + Ok(supervise) => { + let lifecycle = self.clone(); + runtime.spawn(async move { lifecycle.run(supervise).await }); } + Err(err) => self.closed(Err(err)), + } + } + + async fn isolate(&self, task: impl std::future::Future>) -> anyhow::Result { + // Boxing keeps large protocol futures off nested debug-build stack frames. + moq_tokio::panic::isolate(Box::pin(task)) + .instrument(self.span.clone()) + .await + .unwrap_or_else(|| Err(anyhow::anyhow!("connection panicked"))) + } + + /// Logged here, under the span's own target, so every transport pairs its `conn{id=N}` + /// with its close under one log filter. + fn closed(&self, result: anyhow::Result<()>) { + let _span = self.span.enter(); + match result { + Ok(()) => tracing::info!(id = self.id, "connection closed"), + Err(err) => tracing::warn!(id = self.id, %err, "connection closed"), } - .instrument(self.span.clone()) - .await } /// Keep the connection span across a worker-to-shared-runtime handoff and return panics to its owner. @@ -351,7 +379,8 @@ mod tests { use super::logs; - /// A panic in the same shared-runtime handoff used for ring supervision reaches the close owner. + /// A panic in the ring's supervision, after the worker handed it to the shared runtime, still + /// logs the connection's close. #[tokio::test(start_paused = true)] async fn shared_supervision_panic_logs_the_connection_close() { let logs = logs::Logs::default(); @@ -367,21 +396,19 @@ mod tests { let (auth, _admissions) = auth::Auth::embedded("test"); let lifecycle = Lifecycle::new(cluster, auth, 42, None); lifecycle - .run(async { - lifecycle - .shared(&tokio::runtime::Handle::current(), async { - panic!("injected supervision panic"); - #[allow(unreachable_code)] - Ok::<_, anyhow::Error>(()) - }) - .await? + .start(&tokio::runtime::Handle::current(), async { + Ok::<_, anyhow::Error>(async { + panic!("injected supervision panic"); + #[allow(unreachable_code)] + Ok::<_, anyhow::Error>(()) + }) }) .await; logs.wait_for("connection closed").await; let text = logs.text(); assert!(text.contains("conn{id=42}"), "{text}"); assert!( - text.contains("connection closed id=42 err=connection task panicked"), + text.contains("connection closed id=42 err=connection panicked"), "{text}" ); assert_eq!(text.matches("connection closed").count(), 1, "{text}"); diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 050d594c395..1c28ad1c40a 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -562,8 +562,9 @@ fn serve_worker(spawn: Spawn) -> bool { .with_shutdown(serve.shutdown.clone()) .with_sessions(serve.sessions.clone()); accept_handle.spawn(async move { + let shared = serve.tokio.clone(); lifecycle - .run(serve_connection(conn_handle, conn, serve, &lifecycle)) + .start(&shared, serve_connection(conn_handle, conn, serve, &lifecycle)) .await; }); } @@ -622,14 +623,15 @@ impl Deadline { } } -/// Serve one accepted connection on its worker: handshake and session start -/// here; authentication and supervision on the shared runtime. +/// Start one accepted connection on its worker: handshake and session start +/// here, authentication on the shared runtime. Returns the session's +/// supervision, which runs on the shared runtime too. async fn serve_connection( handle: moq_uring::Handle, conn: moq_uring::quic::Connection, serve: Serve, lifecycle: &Lifecycle, -) -> anyhow::Result<()> { +) -> anyhow::Result> + Send + 'static> { use web_transport_trait::poll::Session as _; // The endpoint started this deadline at the connection's first packet and @@ -773,14 +775,11 @@ async fn serve_connection( }; handle.spawn(moq_tokio::panic::isolate(driver).in_current_span().map(drop)); - // The driver stays on this worker; admission and supervision share the same - // lifecycle owner on the runtime that owns auth's HTTP client and timers. - let supervisor = lifecycle.clone(); - lifecycle - .shared(&serve.tokio, async move { - supervisor.supervise(session, lease, registration, transport).await - }) - .await? + // The driver stays on this worker. The session handle is Send + Sync, so its + // supervision (credential expiry, GOAWAY drain) runs with the timers and the + // shutdown broadcast on the shared runtime. + let lifecycle = lifecycle.clone(); + Ok(async move { lifecycle.supervise(session, lease, registration, transport).await }) } #[cfg(test)] diff --git a/rs/moq-relay/tests/integration/runtime_uring.rs b/rs/moq-relay/tests/integration/runtime_uring.rs index a6676b8b8b4..b718be1ba10 100644 --- a/rs/moq-relay/tests/integration/runtime_uring.rs +++ b/rs/moq-relay/tests/integration/runtime_uring.rs @@ -886,7 +886,8 @@ async fn uring_sessions_log_their_close_under_a_connection_span() { .with_writer(logs.clone()) .finish() .with(tracing_subscriber::filter::Targets::new().with_target("moq_relay::connection", tracing::Level::INFO)); - // Worker threads need the same dispatcher as the shared runtime. + // Worker threads need the same dispatcher as the shared runtime. Global is + // safe because nextest runs each test in its own process. tracing::subscriber::set_global_default(subscriber).expect("test owns the global subscriber"); let dir = tempfile::tempdir().unwrap(); let (cert, key) = certificate(dir.path()); @@ -897,16 +898,26 @@ async fn uring_sessions_log_their_close_under_a_connection_span() { let port = relay.addr().unwrap().port(); let trigger = relay.shutdown_trigger().clone(); let running = tokio::spawn(relay.run()); - for scheme in ["moql", "https"] { + let mut connections = Vec::new(); + for (id, scheme) in ["moql", "https"].into_iter().enumerate() { let url = format!("{scheme}://127.0.0.1:{port}/logs").parse().unwrap(); - let connection = connect(client(), url).await; - tokio::time::timeout(TIMEOUT, logs.wait_for("negotiated")) + connections.push(connect(client(), url).await); + tokio::time::timeout(TIMEOUT, logs.wait_for(&format!("conn{{id={id} "))) .await .expect("negotiated log missing"); - let _ = connection.close().await; } + // Both sessions are still open when the drain starts, so the workers stop + // right after they leave: their closes must not depend on a stopped worker. trigger.start(); + for connection in connections { + let _ = connection.close().await; + } tokio::time::timeout(TIMEOUT, running).await.unwrap().unwrap().unwrap(); + for id in 0..2 { + tokio::time::timeout(TIMEOUT, logs.wait_for(&format!("connection closed id={id}"))) + .await + .unwrap_or_else(|_| panic!("conn {id} never logged its close: {}", logs.text())); + } let text = logs.text(); let negotiated: Vec<_> = text.lines().filter(|line| line.contains("negotiated")).collect(); let closed: Vec<_> = text.lines().filter(|line| line.contains("connection closed")).collect();