From 0ab059cc5a20d063eb4da24bafd193e2b60efc7f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 18:14:39 -0700 Subject: [PATCH 01/16] chore: claim session outcomes Co-Authored-By: GPT-6 From 37ebd83308140726c3d3882e788754e12cf57737 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 18:54:06 -0700 Subject: [PATCH 02/16] feat(stats): count refused and failed sessions Share fixed session outcome counters across stats tracks, admission and metrics, preserving typed close causes and logical reconnect lifetimes. Co-Authored-By: GPT-6 --- doc/bin/relay/config.md | 6 + doc/concept/stats.md | 16 ++ quest/m1/README.md | 1 - quest/m1/drain-cleanup.md | 1 - quest/m1/refusal-reason.md | 9 +- quest/m1/session-outcomes.md | 98 ------------ quest/m2/grant-label.md | 4 +- quest/m2/uring-connection-log.md | 1 - quest/m2/wire-stats/README.md | 9 +- rs/moq-ffi/src/error.rs | 3 + rs/moq-net/src/coding/codes.rs | 2 +- rs/moq-net/src/error.rs | 16 +- rs/moq-net/src/ietf/auth.rs | 9 +- rs/moq-net/src/ietf/session.rs | 9 +- rs/moq-net/src/lite/publisher.rs | 11 +- rs/moq-net/src/lite/session.rs | 5 +- rs/moq-net/src/model/resume.rs | 1 + rs/moq-net/src/stats.rs | 114 ++++++++++++++ rs/moq-net/src/stats/outcomes.rs | 205 ++++++++++++++++++++++++++ rs/moq-net/src/transport.rs | 17 +++ rs/moq-relay/src/auth.rs | 41 +++++- rs/moq-relay/src/cluster.rs | 92 +++++++++++- rs/moq-relay/src/connection.rs | 76 ++++++++-- rs/moq-relay/src/internal.rs | 20 ++- rs/moq-relay/src/refusals.rs | 117 ++++----------- rs/moq-relay/src/uring.rs | 8 +- rs/moq-relay/tests/auth_embedder.rs | 28 ++++ rs/moq-relay/tests/refusals.rs | 155 +++++++++++++++++++ rs/moq-stats/src/aggregate.rs | 63 +++++++- rs/moq-stats/src/lib.rs | 2 +- rs/moq-tokio/src/transport/owned.rs | 64 ++++++++ rs/moq-uring/src/quic/mod.rs | 3 + rs/moq-uring/src/quic/noq/mod.rs | 1 + rs/moq-uring/src/transport/adapter.rs | 31 ++++ rs/moq-uring/tests/transport.rs | 3 + 35 files changed, 996 insertions(+), 245 deletions(-) delete mode 100644 quest/m1/session-outcomes.md create mode 100644 rs/moq-net/src/stats/outcomes.rs diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 041991962b..7266bfc1a1 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -272,6 +272,12 @@ per-session limits, past which one is closed with `TOO_MANY_REQUESTS`. [Stats](/concept/stats) describes the paths, tracks, and encodings; read them with the [`moq-stats`](https://docs.rs/moq-stats) crate. +Session outcome counters in the stats feed distinguish admission refusals from +abnormal closes; see [session stats](/concept/stats#sessions). `/metrics` exposes +`moq_relay_sessions_refused_total` by reason and +`moq_relay_sessions_failed_total` by tier and kind. Refusals remain counted when +stats publishing is disabled. + ## \[iroh] Iroh requires building the relay from source with `cargo build --release -p moq-relay --features iroh`. Published binaries, Docker images, and the Nix package leave it out. diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 58e3cbef7f..421b37a218 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -181,6 +181,22 @@ group on the tier for `totals.json`. `sessions` and `sessions_closed` are their legacy spellings. A session moved to a new tier ends on the old one and starts on the new. +`sessions_refused` counts refusals by `refused`, `unavailable`, `request`, +`forbidden`, and `lan`. Relay admission refusals have no resolved grant, so they +belong to the empty root on the default tier. An embedder that refuses a session +outside relay admission can record it on its stats handle under the root and tier +it resolved. Refused sessions never increase the live session count. + +`sessions_failed` counts abnormal ends by `idle_timeout`, `transport`, +`application`, and `expired`. A clean peer close, local close, drain, or shutdown +is not a failure. These are subsets of the sessions that end, leaving the live +count unchanged. A reconnecting cluster dial retains one logical session across +physical reconnects, so intermediate retry statuses add no failure; its typed +terminal session error counts once when that logical session ends. Initial +setup errors and configuration failures do not become admitted-session failures. +Both objects default to zero when reading an older publisher; +their counters survive pruning and follow the node epoch and reset rules. + `requests` counts admitted request/response exchanges under the token's root and tier, including HTTP `/fetch` and `/announced`. They never start or end a session. A fetch's delivered group bytes count against the broadcast path on that tier; diff --git a/quest/m1/README.md b/quest/m1/README.md index cc1cccde9c..c48d11fa22 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -128,7 +128,6 @@ above them. - [Rendition preference](/quest/m1/rendition-preference.md) - automatic selection by ``, `Video::ranked`, and WHEP keeps the highest `preference` that decodes, so a compatibility transcode is only picked when nothing preferred decodes - [Rust PROBE on demand](/quest/m1/probe-on-demand-rs.md) - a Rust client opens PROBE only while a `recv_bandwidth()` consumer is held, and `Stats` drops `estimated_recv_rate` - [QoS](/quest/m1/qos/README.md) - broadcast health: relay starvation and timeliness histograms -- [Session outcomes](/quest/m1/session-outcomes.md) - the relay's session stats count refusals by reason and ends by kind, per root and tier, including refusals an embedder records on its stats handle - [Typed refusal reason](/quest/m1/refusal-reason.md) - an auth server's 403 names the reason (including `expired`) and the root and tier, so the relay's session outcomes count and attribute refusals - [JS track handover](/quest/m1/js-group-handover.md) - a JS track subscription resumes across a route swap from the first frame it lacks, so `test/drain` passes at zero latency budget - [Scope track priority](/quest/m1/track-priority-scope.md) - priority orders one owner's streams, and a shared cluster session is fair across tenants diff --git a/quest/m1/drain-cleanup.md b/quest/m1/drain-cleanup.md index cabb9b574e..1629b37e21 100644 --- a/quest/m1/drain-cleanup.md +++ b/quest/m1/drain-cleanup.md @@ -89,4 +89,3 @@ Wire: none. ## Related - [Handshake deadline covers admission](/quest/m1/handshake-admission-deadline.md) - bounds pending admissions, which the drain deadline must keep cutting -- [Session outcomes](/quest/m1/session-outcomes.md) - counts how sessions end, which needs this end recorded diff --git a/quest/m1/refusal-reason.md b/quest/m1/refusal-reason.md index 88463f65be..4aa2030456 100644 --- a/quest/m1/refusal-reason.md +++ b/quest/m1/refusal-reason.md @@ -15,7 +15,7 @@ refusal to that root and tier. Decided 2026-10-08: moved out of the auth line to m1, since it extends session outcomes and the HTTP auth contract, not in-band AUTH. -Decided 2026-10-08 (split from [Session outcomes](/quest/m1/session-outcomes.md)): +Decided 2026-10-08: - **Why.** `TokenExpired` is raised only in moq-auth's key check (`rs/moq-auth/src/key.rs`), on the auth server. `serve.rs` sends every @@ -40,8 +40,7 @@ Decided 2026-10-08 (split from [Session outcomes](/quest/m1/session-outcomes.md) path and `Admission::refuse` one refusal shape (reason, optional root and tier), under the same bound, and update `From`, which sends anything but `Refused` to `Unavailable` today. A refusal an embedder - records on its stats handle ([Session outcomes](/quest/m1/session-outcomes.md)) - takes `expired` too. + records through `stats::Handle::refuse` takes `expired` too. - **Relay.** Map the typed error to a `Refusal` reason (adding `expired`) in `/metrics`, and count it in the root's `Presence` on that tier, or the default tier when the body names none. A refusal without a root stays @@ -56,10 +55,6 @@ Decided 2026-10-08 (split from [Session outcomes](/quest/m1/session-outcomes.md) Public API: moq-auth's refusal error and the HTTP refusal body. Wire: the auth HTTP contract only, not MoQ. -## Required - -- [Session outcomes](/quest/m1/session-outcomes.md) - the refusal counters - this attributes and extends with `expired` ## Related diff --git a/quest/m1/session-outcomes.md b/quest/m1/session-outcomes.md deleted file mode 100644 index 0889e03493..0000000000 --- a/quest/m1/session-outcomes.md +++ /dev/null @@ -1,98 +0,0 @@ -# [M] Session outcomes - -## Goal - -The relay's session stats carry, per tier in the `totals.json` track and per -root in the requested `/presence.json` track -([Stats](/doc/concept/stats.md#tracks)), how many sessions were -refused, by reason, and how many ended abnormally, by kind: idle timeout, -transport error, application error, or token expired. A dashboard computes an -error rate from the stats feed alone, with no log scraping. Refusals count -unattributed, on the default tier, until -[Typed refusal reason](/quest/m1/refusal-reason.md) adds `expired` and -attributes a refusal to the root and tier the auth server resolved. - -An embedding application records a refusal on its stats handle, with the -same reasons and its own root and tier, for a connection it refuses outside -the relay's admission (a gateway protocol it terminates itself). A refusal -through `Cluster::admit` counts like any other. - -## Plan - -Decided 2026-10-07, for per-project error counts downstream (browser -playback needs an error rate it cannot see client-side): - -- **Refusals reuse moq#4825's reasons** (`refusals.rs`: refused, unavailable, - request, forbidden, lan), so `/metrics` and the stats track share one - vocabulary. -- **The relay's refusals stay unattributed here** (decided 2026-10-08). The - relay verifies no token itself (its deciders are the auth server, public rules, or - refuse-all in `rs/moq-relay/src/auth.rs`), and a refused session has no - grant, so no root or tier. Count a refusal under the empty root on the - default tier. `group_key` sends the empty root to the root group - (`/node/`) at any `--stats-depth`, so it lands there, created - on first sight and following the stats linger and epoch; a project's - group never sees it. Sessions admitted under the empty root (public rules) - share that group. `expired` and per-root attribution need the decider to - say so, which [Typed refusal reason](/quest/m1/refusal-reason.md) adds. -- **Embedder refusals** (decided 2026-10-10, maintainer priority). An - embedder that terminates a gateway protocol itself refuses some - connections before or without relay admission, and today cannot count - them: `Refusals` (`rs/moq-relay/src/refusals.rs`) is `pub(crate)`, and - `Registry::tier(..).session(root)` has no refusal API. Give the stats - handle one (shaped like `tier(..).refuse(root, reason)`), with the reason - type public beside `Presence` so both share one taxonomy. The embedder - names the root and tier; it is trusted to name one it resolved, never a - dialed path, the same bound as - [Typed refusal reason](/quest/m1/refusal-reason.md). `/metrics` sees these - refusals too, ideally by rendering refusals from the registry the way - `moq_relay_sessions_opened_total` already is, so there is one counter. -- **Gateway admission counts** (folded from the m2 refusal-reasons quest, - 2026-10-10). `Cluster::admit` returns a refusal (from `Auth::admit` or - `scope`) to the embedder uncounted. Native QUIC (`Connection::run` in - `connection.rs`) and the non-LAN io_uring branch (`uring.rs`) already record - whatever `cluster.admit` returns, and WebSocket records around its own - `scope`. Make `Cluster::admit` the single owner of its refusals, unattributed - like the relay's own, and drop the QUIC and io_uring records of its errors; - those callers keep recording only what they refuse themselves (LAN). The - embedder records through the handle only what the relay never saw. -- **Ends are classified from a typed close kind, not a string.** A peer's - application close is already typed (`Error::from_transport` decodes it into - `Error::Session`), but `moq_net::Error::Transport(String)` still flattens - idle timeout and connection error, and `supervise` in - `rs/moq-relay/src/connection.rs` reports `err.to_string()`. Add the typed - kind where the QUIC backend reports the close. Clean covers `Error::Cancel`, - a peer's zero-code close (`Error::Session(SessionError::Cancel)`), a local - close (`Error::Closed`), drain, and shutdown; `supervise` matches only - `Error::Cancel` today, so a classifier built on it would count every normal - browser close as failed. A lease ending `expired` is the expired kind. -- **Wire shape:** `Presence` gains `sessions_refused` and `sessions_failed`, - fixed named counters per reason and per kind, so `Presence` stays `Copy` - (what downstream aggregators plan against). `sessions_ended` stays - the total, so the live count (`sessions_started - sessions_ended`) is - unchanged. Readers take them from the totals and presence tracks (quest - audit, 2026-10-08); `sessions.json` stays for existing readers (#5084, kept - backwards compatible 2026-10-12) and carries the same `Presence` entries. -- **Outcomes follow the stats counter rules.** A pruned root folds its - counts into the group totals, and the aggregator merges with a baseline per - upstream (node, epoch), so a node leaving or returning never reads as new - errors or a drop. -- Tests: a forbidden path and an auth server refusal each count under their - reason; an embedder refusal on its handle counts under its root, tier, and - reason, in the stats track and `/metrics`; a `Cluster::admit` refusal - counts once, through an embedder, native QUIC, and io_uring alike; a refusal with `--stats-depth 1` and no admitted session - appears in the root group's totals; a clean close, a peer's zero-code - close, an idle timeout, and a peer application error each count as their - kind; the aggregator sums the counters across nodes, and a node departing - and returning while another stays leaves the merged counts unchanged. - -Public API: a refusal reason type and a refusal method on `stats::Handle`; -`Presence` gains the counters. Wire: the stats track's `Presence` fields. - -## Related - -- [Typed refusal reason](/quest/m1/refusal-reason.md) - adds `expired` - and per-root, per-tier refusal attribution on top of these counters, and - the `/metrics` split of an expired token from an invalid one -- [Own the QUIC stack](/quest/m1/quic/README.md) - where a typed close kind - is natural once `moq-quic` owns the connection diff --git a/quest/m2/grant-label.md b/quest/m2/grant-label.md index b920f9fde0..f6c33b6ed4 100644 --- a/quest/m2/grant-label.md +++ b/quest/m2/grant-label.md @@ -78,9 +78,9 @@ Public API: a new optional `Grant::label` in `moq-auth` and `@moq/auth`, and `stats::Session::set_tier` becomes a `(tier, label)` retag. Wire: the auth HTTP grant body and the stats track's per-label byte fields, not MoQ. +The per-root session outcome counters follow the same label. + ## Related -- [Session outcomes](/quest/m1/session-outcomes.md) - its per-root counters - follow the same label - [Typed refusal reason](/quest/m1/refusal-reason.md) - a refusal body names a root too; decide there whether it may name a label diff --git a/quest/m2/uring-connection-log.md b/quest/m2/uring-connection-log.md index 775ff2106c..753a675af6 100644 --- a/quest/m2/uring-connection-log.md +++ b/quest/m2/uring-connection-log.md @@ -54,4 +54,3 @@ Public API: none expected. Wire: none. - [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/quest/m2/wire-stats/README.md b/quest/m2/wire-stats/README.md index 4ca9b5939b..549342e0cd 100644 --- a/quest/m2/wire-stats/README.md +++ b/quest/m2/wire-stats/README.md @@ -52,8 +52,8 @@ see cost per tenant. which loses a crashed session and lags a long one. - **One source registry for sessions and embedders.** An embedder registers a gateway connection's wire source against a (tier, root) on its stats - handle, the same path the relay's own sessions use, like - [session outcomes](/quest/m1/session-outcomes.md)' embedder refusals. + handle, the same path the relay's own sessions use, like the existing + `stats::Handle::refuse` API. ## Required @@ -66,8 +66,3 @@ see cost per tenant. reports its wire bytes - [moq-rtc wire source](/quest/m2/wire-stats/rtc.md) - a WebRTC peer reports its wire bytes - -## Related - -- [Session outcomes](/quest/m1/session-outcomes.md) - the other per-root - `Presence` counters and the embedder stats-handle pattern diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index f2d6ab0029..747bdca237 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -295,6 +295,9 @@ impl From for MoqError { fn from(err: moq_net::Error) -> Self { match err { moq_net::Error::Transport(message) => Self::Transport { reason: message }, + moq_net::Error::IdleTimeout => Self::Transport { + reason: "transport idle timeout".into(), + }, moq_net::Error::NotFound | moq_net::Error::NotFetchable => Self::NotFound, moq_net::Error::Closed | moq_net::Error::GoingAway | moq_net::Error::SessionClosed => Self::Closed, moq_net::Error::Cancel => Self::Cancelled, diff --git a/rs/moq-net/src/coding/codes.rs b/rs/moq-net/src/coding/codes.rs index c12ec1abe5..fad36ce85d 100644 --- a/rs/moq-net/src/coding/codes.rs +++ b/rs/moq-net/src/coding/codes.rs @@ -39,7 +39,7 @@ pub trait StreamCodes { return self.decode_stream_code(code).into(); } - Error::Transport(err.to_string()) + Error::from_transport(err) } } diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index a913cf80f5..e7737ff569 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -520,6 +520,9 @@ pub enum Error { /// Session and stream unknowns are [`Self::Session`] / [`Self::Stream`], not this. #[error("remote error: code={0}")] Remote(u32), + /// The underlying connection reached its transport idle deadline. + #[error("transport idle timeout")] + IdleTimeout, } impl Error { @@ -556,7 +559,11 @@ impl Error { return StreamError::from_code(code).into(); } - Self::Transport(err.to_string()) + match err.close_kind() { + crate::transport::CloseKind::IdleTimeout => Self::IdleTimeout, + crate::transport::CloseKind::Closed => Self::Closed, + _ => Self::Transport(err.to_string()), + } } } @@ -668,6 +675,13 @@ impl From<&Error> for StreamError { } impl crate::transport::Error for Error { + fn close_kind(&self) -> crate::transport::CloseKind { + match self { + Self::IdleTimeout => crate::transport::CloseKind::IdleTimeout, + Self::Closed => crate::transport::CloseKind::Closed, + _ => crate::transport::CloseKind::Transport, + } + } fn session_error(&self) -> Option<(u32, String)> { None } diff --git a/rs/moq-net/src/ietf/auth.rs b/rs/moq-net/src/ietf/auth.rs index 53ae67d2c5..05369c078a 100644 --- a/rs/moq-net/src/ietf/auth.rs +++ b/rs/moq-net/src/ietf/auth.rs @@ -286,7 +286,7 @@ where } match &err { - Error::Cancel | Error::Unsupported | Error::Transport(_) | Error::Session(_) => { + Error::Cancel | Error::Unsupported | Error::Transport(_) | Error::IdleTimeout | Error::Session(_) => { tracing::debug!(%err, "auth token ended") } err => tracing::warn!(%err, "auth token ended"), @@ -465,7 +465,12 @@ impl Serve { Ok(()) => Error::Cancel, Err(err) => { match &err { - Error::Cancel | Error::Unsupported | Error::Stream(_) | Error::Session(_) | Error::Transport(_) => { + Error::Cancel + | Error::Unsupported + | Error::Stream(_) + | Error::Session(_) + | Error::Transport(_) + | Error::IdleTimeout => { tracing::debug!(%err, "auth request ended") } err => tracing::warn!(%err, "auth request error"), diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 997fdeeb27..022eb6dc2b 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -600,7 +600,7 @@ where }); match &res { - Err(err @ Error::Transport(_)) => { + Err(err @ (Error::Transport(_) | Error::IdleTimeout)) => { tracing::info!(%err, "session terminated"); session.close(super::error::to_session_code(&SessionError::Internal), ""); } @@ -1095,7 +1095,12 @@ impl Uni { fn died_before_header(err: &Error) -> bool { matches!( err, - Error::Cancel | Error::Stream(_) | Error::Remote(_) | Error::Transport(_) | Error::Decode(DecodeError::Short) + Error::Cancel + | Error::Stream(_) + | Error::Remote(_) + | Error::Transport(_) + | Error::IdleTimeout + | Error::Decode(DecodeError::Short) ) } diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 5a3b1070b6..deabf3c509 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -435,7 +435,12 @@ impl AuthServe { Ok(()) => Error::Cancel, Err(err) => { match &err { - Error::Cancel | Error::Unsupported | Error::Stream(_) | Error::Session(_) | Error::Transport(_) => { + Error::Cancel + | Error::Unsupported + | Error::Stream(_) + | Error::Session(_) + | Error::Transport(_) + | Error::IdleTimeout => { tracing::debug!(%err, "auth stream ended") } err => tracing::warn!(%err, "auth stream error"), @@ -895,7 +900,8 @@ impl AnnounceServe { Error::Cancel | Error::Stream(crate::StreamError::Cancel) | Error::Session(crate::SessionError::Cancel) - | Error::Transport(_) => { + | Error::Transport(_) + | Error::IdleTimeout => { tracing::debug!(prefix = %origin.absolute(""), "announcing cancelled"); } err => { @@ -1388,6 +1394,7 @@ fn is_cancel(err: &Error) -> bool { | Error::Stream(crate::StreamError::Cancel) | Error::Session(crate::SessionError::Cancel) | Error::Transport(_) + | Error::IdleTimeout ) } diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 8b97b1d84e..ceed15b2ac 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -87,6 +87,7 @@ impl SetupAccept { | Error::Stream(_) | Error::Remote(_) | Error::Transport(_) + | Error::IdleTimeout | Error::Decode(crate::DecodeError::Short), )) => { pending.swap_remove(index); @@ -394,7 +395,7 @@ where self.subscriber.abort(err); } match &res { - Err(Error::Transport(_)) => { + Err(Error::Transport(_) | Error::IdleTimeout) => { tracing::info!("session terminated"); self.session.close(SessionError::Internal.to_code(), ""); } @@ -549,7 +550,7 @@ impl kio::Task for PresentToken { err => crate::auth::protocol_error(err), }; match &err { - Error::Cancel | Error::Unsupported | Error::Transport(_) | Error::Session(_) => { + Error::Cancel | Error::Unsupported | Error::Transport(_) | Error::IdleTimeout | Error::Session(_) => { tracing::debug!(%err, "auth token ended") } err => tracing::warn!(%err, "auth token ended"), diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 51e5f021e3..a165f151b9 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -216,6 +216,7 @@ fn route_failed(err: &Error) -> bool { matches!( err, Error::Transport(_) + | Error::IdleTimeout | Error::Session(_) | Error::SessionClosed | Error::GoingAway diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index f2b1635c61..3b773c3c05 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -128,6 +128,10 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer}; use crate::{AsPath, PathOwned, Pattern, Patterns}; +mod outcomes; +pub use outcomes::{Failure, Failures, Refusal, Refusals}; +use outcomes::{FailureCounters, RefusalCounters}; + /// Cumulative atomic counters for a single `(tier, role)` on a broadcast. /// /// Started counters bump when a model handle records activity; their `_ended` @@ -269,6 +273,8 @@ impl Content { /// bumps on connect, `sessions_ended` on disconnect. #[derive(Default, Debug)] struct SessionCounters { + sessions_refused: RefusalCounters, + sessions_failed: FailureCounters, requests: AtomicU64, sessions_started: AtomicU64, sessions_ended: AtomicU64, @@ -288,9 +294,12 @@ impl SessionCounters { /// before started with `Relaxed`, the same pairing as [`Counters::snapshot`], /// so the readout never shows `ended > started`. fn snapshot(&self) -> Presence { + let sessions_failed = self.sessions_failed.snapshot(); let sessions_ended = self.sessions_ended.load(Ordering::Acquire); let sessions_started = self.sessions_started.load(Ordering::Relaxed); Presence { + sessions_refused: self.sessions_refused.snapshot(), + sessions_failed, requests: self.requests.load(Ordering::Relaxed), sessions_started, sessions_ended, @@ -537,6 +546,10 @@ impl Traffic { #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] #[non_exhaustive] pub struct Presence { + /// Cumulative sessions refused before admission, by reason. + pub sessions_refused: Refusals, + /// Cumulative admitted sessions ending abnormally, by kind. + pub sessions_failed: Failures, /// Cumulative request/response exchanges admitted. pub requests: u64, /// Cumulative sessions connected. @@ -552,6 +565,8 @@ pub struct Presence { #[derive(Serialize)] struct PresenceSer { + sessions_refused: Refusals, + sessions_failed: Failures, requests: u64, sessions_started: u64, sessions: u64, @@ -564,6 +579,8 @@ struct PresenceSer { impl From for PresenceSer { fn from(p: Presence) -> Self { Self { + sessions_refused: p.sessions_refused, + sessions_failed: p.sessions_failed, requests: p.requests, sessions_started: p.sessions_started, sessions: p.sessions_started, @@ -578,6 +595,8 @@ impl From for PresenceSer { #[derive(Default, Deserialize)] #[serde(default)] struct PresenceDe { + sessions_refused: Refusals, + sessions_failed: Failures, requests: u64, sessions_started: Edge, sessions: Edge, @@ -590,6 +609,8 @@ struct PresenceDe { impl From for Presence { fn from(d: PresenceDe) -> Self { Self { + sessions_refused: d.sessions_refused, + sessions_failed: d.sessions_failed, requests: d.requests, sessions_started: counter_edge(d.sessions_started, d.sessions), sessions_ended: counter_edge(d.sessions_ended, d.sessions_closed), @@ -614,6 +635,8 @@ impl<'de> Deserialize<'de> for Presence { impl Presence { /// Fold another readout into this one. The peaks are per session, so they take the larger. pub fn add(&mut self, other: Presence) { + self.sessions_refused.add(other.sessions_refused); + self.sessions_failed.add(other.sessions_failed); self.requests += other.requests; self.sessions_started += other.sessions_started; self.sessions_ended += other.sessions_ended; @@ -912,6 +935,11 @@ impl Registry { } } + /// Whether this registry collects counters. + pub fn is_enabled(&self) -> bool { + self.shared.is_some() + } + /// The excluded patterns. See [`Config::exclude`]. pub fn exclude(&self) -> &Patterns { &self.exclude @@ -1100,6 +1128,16 @@ pub struct Handle { } impl Handle { + /// Count a session refused outside relay admission under its resolved root and tier. + pub fn refuse(&self, root: impl AsPath, reason: Refusal) { + if let Some(presence) = self.stats.session_counters(&self.tier, root) { + presence + .sessions_refused + .counter(reason) + .fetch_add(1, Ordering::Relaxed); + } + } + /// The registry this handle is tied to. pub fn parent(&self) -> &Registry { &self.stats @@ -1141,6 +1179,7 @@ impl Handle { root, connected, current: Mutex::new(Current { + failure: None, tier: self.tier.clone(), presence, }), @@ -1236,6 +1275,7 @@ struct SessionInner { /// The tier a [`Session`] records under right now. struct Current { + failure: Option, tier: Tier, /// The presence gauge for `(tier, root)`, or `None` for a disabled registry. presence: Option>, @@ -1249,6 +1289,18 @@ struct Viewer { } impl Session { + /// Record the first abnormal end on this connected session, shared across clones. + pub fn fail(&self, kind: Failure) { + let Some(inner) = &self.inner else { return }; + if !inner.connected { + return; + } + let mut current = inner.current.lock().unwrap_or_else(PoisonError::into_inner); + if current.failure.is_none() { + current.failure = Some(kind); + } + } + /// Record later traffic and session presence under `tier`; a request stays counted where admitted. pub fn set_tier(&self, tier: Tier) { let Some(inner) = &self.inner else { return }; @@ -1387,6 +1439,10 @@ impl Drop for SessionInner { // Release pairs with the readout's Acquire load of `sessions_ended` // (see the module-level "Snapshot atomicity" note). presence.sessions_ended.fetch_add(1, Ordering::Release); + if let Some(kind) = current.failure { + // Read failures before ended; this release also publishes its matching end. + presence.sessions_failed.counter(kind).fetch_add(1, Ordering::Release); + } } } } @@ -1635,6 +1691,64 @@ mod tests { use super::*; + #[test] + fn outcomes_survive_clones_tier_changes_and_pruning() { + let stats = test_stats(); + let handle = stats.tier(Tier::new("gateway")); + handle.refuse("root", Refusal::Forbidden); + let session = handle.session("root"); + let clone = session.clone(); + session.set_tier(Tier::new("next")); + clone.fail(Failure::IdleTimeout); + session.fail(Failure::Application); + drop(clone); + drop(session); + let before = stats.snapshot(); + let gateway = before + .sessions() + .into_iter() + .find(|(tier, _)| tier.as_str() == "gateway") + .unwrap() + .1; + assert_eq!(gateway.sessions_refused.forbidden, 1); + assert_eq!(gateway.sessions_failed, Failures::default()); + let next = before + .sessions() + .into_iter() + .find(|(tier, _)| tier.as_str() == "next") + .unwrap() + .1; + assert_eq!(next.sessions_failed.idle_timeout, 1); + assert_eq!(next.sessions_failed.application, 0); + assert_eq!(next.active(), 0); + drain(&stats); + let after = stats.snapshot(); + assert_eq!(before.sessions(), after.sessions(), "retirement keeps outcomes once"); + let disabled = Registry::disabled(); + disabled.tier(Tier::default()).refuse("root", Refusal::Refused); + disabled.tier(Tier::default()).session("root").fail(Failure::Expired); + assert!(disabled.snapshot().sessions().is_empty()); + } + + #[test] + fn outcome_wire_fields_default_and_roundtrip_without_changing_live_count() { + let old: Presence = serde_json::from_str(r#"{"sessions":4,"sessions_closed":2}"#).unwrap(); + assert_eq!(old.sessions_refused, Refusals::default()); + assert_eq!(old.sessions_failed, Failures::default()); + let mut current = old; + current.sessions_refused.request = 3; + current.sessions_failed.expired = 1; + assert_eq!(current.active(), 2); + assert_eq!( + serde_json::from_str::(&serde_json::to_string(¤t).unwrap()).unwrap(), + current + ); + let mut sum = current; + sum.add(current); + assert_eq!(sum.sessions_refused.request, 6); + assert_eq!(sum.sessions_failed.expired, 2); + } + #[test] fn default_tier_has_empty_label() { let tier = Tier::default(); diff --git a/rs/moq-net/src/stats/outcomes.rs b/rs/moq-net/src/stats/outcomes.rs new file mode 100644 index 0000000000..a4042439c6 --- /dev/null +++ b/rs/moq-net/src/stats/outcomes.rs @@ -0,0 +1,205 @@ +//! Fixed session-outcome counters shared by embedders and relay metrics. + +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicU64, Ordering}; + +/// The reason a session was refused. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub enum Refusal { + /// The admission decider refused the session. + Refused, + /// The admission decider could not answer. + Unavailable, + /// The admission request cannot be decided. + Request, + /// The grant covers none of the requested operations. + Forbidden, + /// The LAN membership proof was missing or invalid. + Lan, +} +impl Refusal { + /// Every currently defined kind, in display order. + pub const ALL: &'static [Self] = &[ + Self::Refused, + Self::Unavailable, + Self::Request, + Self::Forbidden, + Self::Lan, + ]; + /// The stable stats and metrics name. + pub const fn as_str(self) -> &'static str { + match self { + Self::Refused => "refused", + Self::Unavailable => "unavailable", + Self::Request => "request", + Self::Forbidden => "forbidden", + Self::Lan => "lan", + } + } +} + +/// Sessions refused before admission, with one cumulative count per kind. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +#[non_exhaustive] +pub struct Refusals { + /// The admission decider refused the session. + pub refused: u64, + /// The admission decider could not answer. + pub unavailable: u64, + /// The admission request cannot be decided. + pub request: u64, + /// The grant covers none of the requested operations. + pub forbidden: u64, + /// The LAN membership proof was missing or invalid. + pub lan: u64, +} +impl Refusals { + /// The cumulative count for this kind. + pub fn count(&self, kind: Refusal) -> u64 { + match kind { + Refusal::Refused => self.refused, + Refusal::Unavailable => self.unavailable, + Refusal::Request => self.request, + Refusal::Forbidden => self.forbidden, + Refusal::Lan => self.lan, + } + } + /// Add every kind from another readout. + pub fn add(&mut self, other: Self) { + self.refused += other.refused; + self.unavailable += other.unavailable; + self.request += other.request; + self.forbidden += other.forbidden; + self.lan += other.lan; + } + /// Whether any cumulative count went backwards. + pub fn regressed(&self, previous: &Self) -> bool { + self.refused < previous.refused + || self.unavailable < previous.unavailable + || self.request < previous.request + || self.forbidden < previous.forbidden + || self.lan < previous.lan + } +} +#[derive(Debug, Default)] +pub(super) struct RefusalCounters { + refused: AtomicU64, + unavailable: AtomicU64, + request: AtomicU64, + forbidden: AtomicU64, + lan: AtomicU64, +} +impl RefusalCounters { + pub(super) fn counter(&self, kind: Refusal) -> &AtomicU64 { + match kind { + Refusal::Refused => &self.refused, + Refusal::Unavailable => &self.unavailable, + Refusal::Request => &self.request, + Refusal::Forbidden => &self.forbidden, + Refusal::Lan => &self.lan, + } + } + pub(super) fn snapshot(&self) -> Refusals { + Refusals { + refused: self.refused.load(Ordering::Relaxed), + unavailable: self.unavailable.load(Ordering::Relaxed), + request: self.request.load(Ordering::Relaxed), + forbidden: self.forbidden.load(Ordering::Relaxed), + lan: self.lan.load(Ordering::Relaxed), + } + } +} + +/// The reason a session ended abnormally. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub enum Failure { + /// The transport reached its idle deadline. + IdleTimeout, + /// The transport failed. + Transport, + /// The application or protocol failed. + Application, + /// The session credential expired. + Expired, +} +impl Failure { + /// Every currently defined kind, in display order. + pub const ALL: &'static [Self] = &[Self::IdleTimeout, Self::Transport, Self::Application, Self::Expired]; + /// The stable stats and metrics name. + pub const fn as_str(self) -> &'static str { + match self { + Self::IdleTimeout => "idle_timeout", + Self::Transport => "transport", + Self::Application => "application", + Self::Expired => "expired", + } + } +} + +/// Admitted sessions ending abnormally, with one cumulative count per kind. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +#[non_exhaustive] +pub struct Failures { + /// The transport reached its idle deadline. + pub idle_timeout: u64, + /// The transport failed. + pub transport: u64, + /// The application or protocol failed. + pub application: u64, + /// The session credential expired. + pub expired: u64, +} +impl Failures { + /// The cumulative count for this kind. + pub fn count(&self, kind: Failure) -> u64 { + match kind { + Failure::IdleTimeout => self.idle_timeout, + Failure::Transport => self.transport, + Failure::Application => self.application, + Failure::Expired => self.expired, + } + } + /// Add every kind from another readout. + pub fn add(&mut self, other: Self) { + self.idle_timeout += other.idle_timeout; + self.transport += other.transport; + self.application += other.application; + self.expired += other.expired; + } + /// Whether any cumulative count went backwards. + pub fn regressed(&self, previous: &Self) -> bool { + self.idle_timeout < previous.idle_timeout + || self.transport < previous.transport + || self.application < previous.application + || self.expired < previous.expired + } +} +#[derive(Debug, Default)] +pub(super) struct FailureCounters { + idle_timeout: AtomicU64, + transport: AtomicU64, + application: AtomicU64, + expired: AtomicU64, +} +impl FailureCounters { + pub(super) fn counter(&self, kind: Failure) -> &AtomicU64 { + match kind { + Failure::IdleTimeout => &self.idle_timeout, + Failure::Transport => &self.transport, + Failure::Application => &self.application, + Failure::Expired => &self.expired, + } + } + pub(super) fn snapshot(&self) -> Failures { + Failures { + idle_timeout: self.idle_timeout.load(Ordering::Acquire), + transport: self.transport.load(Ordering::Acquire), + application: self.application.load(Ordering::Acquire), + expired: self.expired.load(Ordering::Acquire), + } + } +} diff --git a/rs/moq-net/src/transport.rs b/rs/moq-net/src/transport.rs index 409ed432e7..f3146973ca 100644 --- a/rs/moq-net/src/transport.rs +++ b/rs/moq-net/src/transport.rs @@ -97,10 +97,27 @@ pub trait Stats { pub struct StatsUnavailable; impl Stats for StatsUnavailable {} +/// The typed transport condition when no application code was received. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[non_exhaustive] +pub enum CloseKind { + /// The transport failed without a more specific classification. + #[default] + Transport, + /// The connection reached its idle deadline. + IdleTimeout, + /// This endpoint closed the connection locally. + Closed, +} + /// A transport failure with optional session and stream application codes. /// /// Session codes and stream codes belong to separate registries. pub trait Error: std::error::Error + MaybeSend + MaybeSync + 'static { + /// The typed transport condition, independent of application close codes. + fn close_kind(&self) -> CloseKind { + CloseKind::Transport + } /// Returns the error code and reason if this was an application error. /// /// Close reasons are exposed as UTF-8 strings at the transport boundary. diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index fd8ad9c9ef..691d55b265 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -530,6 +530,21 @@ impl Lease { self } + pub(crate) fn close_session(self, error: &moq_net::Error, bytes: Bytes) { + let stats = self.stats.clone(); + let reason = if crate::connection::failure(error).is_none() { + "closed".to_string() + } else { + error.to_string() + }; + let reason = self.close(reason, bytes); + if reason != lease::Reason::Expired + && let Some(kind) = crate::connection::failure(error) + { + stats.fail(kind); + } + } + /// Attach the session's stats context, so a re-checked tier retags it live. pub fn with_stats(mut self, stats: moq_net::stats::Session) -> Self { self.stats = stats; @@ -640,7 +655,11 @@ impl Lease { { reason = ended; } - self.consumer.close(reason, bytes) + let reason = self.consumer.close(reason, bytes); + if reason == lease::Reason::Expired { + self.stats.fail(moq_net::stats::Failure::Expired); + } + reason } } @@ -1312,3 +1331,23 @@ mod tests { assert_eq!(token.recheck(&everything()).root, Path::new("vanity/room").to_owned()); } } + +#[cfg(test)] +mod outcome_tests { + use super::*; + + #[test] + fn typed_expiry_wins_a_racing_transport_close_without_string_classification() { + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session("tenant"); + let (producer, consumer) = lease::Producer::new(Grant::default()); + let lease = Lease::new("/tenant", consumer).with_stats(stats.clone()); + producer.revoke(lease::Reason::Expired); + lease.close_session(&moq_net::Error::Transport("closed".into()), Bytes::default()); + drop(stats); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_failed.expired, 1); + assert_eq!(presence.sessions_failed.transport, 0); + assert_eq!(presence.sessions_ended, 1); + } +} diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index eb41f63d18..b9d2edb04e 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -1075,11 +1075,15 @@ impl Cluster { /// Admit a gateway session and scope both origin directions from its grant. pub async fn admit(&self, auth: &auth::Auth, request: moq_auth::Request) -> Result { - if !matches!(request.event, moq_auth::Event::Connect) { - return Err(auth::Error::Request("admission requires a connect event".into())); + let result = async { + if !matches!(request.event, moq_auth::Event::Connect) { + return Err(auth::Error::Request("admission requires a connect event".into())); + } + let lease = auth.admit(request.clone()).await?; + self.scope(lease, &request) } - let lease = auth.admit(request.clone()).await?; - self.scope(lease, &request) + .await; + result.inspect_err(|err| self.refusals.record(err.into())) } /// Network sessions enforce changing grants in AUTH, within a fixed root and SETUP role. @@ -1088,8 +1092,12 @@ impl Cluster { auth: &auth::Auth, request: moq_auth::Request, ) -> Result { - let lease = auth.admit(request.clone()).await?; - self.scope_session(lease, &request) + let result = async { + let lease = auth.admit(request.clone()).await?; + self.scope_session(lease, &request) + } + .await; + result.inspect_err(|err| self.refusals.record(err.into())) } pub(crate) fn scope_session( @@ -1225,6 +1233,7 @@ impl Cluster { /// its own handle: publishing lasts exactly as long as the cluster does. pub fn with_stats(mut self, stats: moq_stats::Producer) -> Self { self.stats = stats.registry().clone(); + self.refusals = crate::refusals::Refusals::new(&self.stats); self._stats_publisher = Some(stats); self } @@ -1802,10 +1811,11 @@ impl Cluster { true => self.origin.clone().upstream(), false => self.origin.clone().peer(), }; + let stats = self.stats.tier(self.cluster_tier()).session(""); let mut client = client .with_publisher(origin.consume().with_hidden(true)) .with_subscriber(origin) - .with_stats(self.stats.tier(self.cluster_tier()).session("")) + .with_stats(stats.clone()) .with_limits(self.peer_limits); if let Some(cost) = target.cost { client = client.with_cost(cost); @@ -1824,8 +1834,9 @@ impl Cluster { // old session finally closed instead of migrating to the replacement. let mut reconnect = client.with_reconnect(true).connect(addrs.clone()); let mut connection = None; + let mut established = false; loop { - match reconnect.status().await? { + match remote_status(&stats, &mut established, reconnect.status().await)? { moq_tokio::Status::Connected if connection.is_none() => { connection = Some(self.nodes.connect_outbound(id, redacted.to_string())); } @@ -2040,6 +2051,24 @@ pub(crate) fn canonicalize_peer_key(peer: &str) -> String { } } +/// A reconnecting peer keeps one logical presence until its loop actually ends. +fn remote_status( + stats: &moq_net::stats::Session, + established: &mut bool, + status: moq_tokio::Result, +) -> moq_tokio::Result { + match &status { + Ok(moq_tokio::Status::Connected) => *established = true, + Err(moq_tokio::Error::MoqNet(error)) if *established => { + if let Some(kind) = crate::connection::failure(error) { + stats.fail(kind); + } + } + _ => {} + } + status +} + #[cfg(test)] mod tests { use super::*; @@ -2196,6 +2225,53 @@ mod tests { /// `?cost=` is read and stripped (it rides SETUP, not the URL), other /// query params survive, and a garbage value is an error rather than a /// silent default. + #[test] + fn remote_logical_outcome_ignores_retries_and_counts_one_terminal_close() { + use moq_net::stats::{Failure, Registry}; + let registry = Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session(""); + let mut established = false; + let _ = remote_status( + &stats, + &mut established, + Err(moq_tokio::Error::MoqNet(moq_net::Error::ProtocolViolation)), + ); + for status in [ + moq_tokio::Status::Connected, + moq_tokio::Status::Disconnected, + moq_tokio::Status::Connected, + moq_tokio::Status::Migrating, + ] { + remote_status(&stats, &mut established, Ok(status)).unwrap(); + } + let _ = remote_status(&stats, &mut established, Err(moq_tokio::Error::ConnectFailed)); + assert_eq!( + registry.snapshot().sessions()[0] + .1 + .sessions_failed + .count(Failure::Transport), + 0 + ); + let _ = remote_status( + &stats, + &mut established, + Err(moq_tokio::Error::MoqNet(moq_net::Error::IdleTimeout)), + ); + let _ = remote_status( + &stats, + &mut established, + Err(moq_tokio::Error::MoqNet(moq_net::Error::Session( + moq_net::SessionError::App(42), + ))), + ); + drop(stats); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_started, 1); + assert_eq!(presence.sessions_ended, 1); + assert_eq!(presence.sessions_failed.idle_timeout, 1); + assert_eq!(presence.sessions_failed.application, 0); + } + #[test] fn cost_param_is_consumed() { let mut url = Url::parse("https://peer.example/?jwt=abc&cost=0").unwrap(); diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index f944205f2c..2265e48ac0 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -9,7 +9,7 @@ use moq_tokio::server::Request; /// the right code instead of sprinkling close/return at each failure site. struct StatusError { status: http::StatusCode, - refusal: Refusal, + refusal: Option, source: anyhow::Error, } @@ -17,7 +17,7 @@ impl From for StatusError { fn from(err: auth::Error) -> Self { Self { status: (&err).into(), - refusal: (&err).into(), + refusal: Some((&err).into()), source: err.into(), } } @@ -103,7 +103,9 @@ impl Connection { let (admitted, registration) = match admitted { Ok(admitted) => admitted, Err(err) => { - self.cluster.refusals.record(err.refusal); + if let Some(refusal) = err.refusal { + self.cluster.refusals.record(refusal); + } let reject = match err.status { http::StatusCode::UNAUTHORIZED => moq_tokio::server::Reject::Unauthorized, http::StatusCode::FORBIDDEN => moq_tokio::server::Reject::Forbidden, @@ -168,7 +170,16 @@ impl Connection { if self.request.peer_identity().is_some() { tracing::debug!("client certificate verified; reported to the auth server"); } - let admitted = self.cluster.admit_session(&self.auth, request.clone()).await?; + let admitted = self + .cluster + .admit_session(&self.auth, request.clone()) + .await + .map_err(|err| { + // Cluster admission owns this refusal; only local/drain refusals remain here. + let mut error = StatusError::from(err); + error.refusal = None; + error + })?; let registration = self.sessions.as_ref().map(|sessions| sessions.register(request)); Ok((admitted, registration)) } @@ -178,7 +189,7 @@ impl Connection { let Some(presented) = cluster::Cluster::lan_credential(self.request.path()) else { return Err(StatusError { status: http::StatusCode::FORBIDDEN, - refusal: Refusal::Lan, + refusal: Some(Refusal::Lan), source: anyhow::anyhow!("LAN peer did not present a membership proof"), }); }; @@ -191,12 +202,12 @@ impl Connection { } Some(false) => Err(StatusError { status: http::StatusCode::FORBIDDEN, - refusal: Refusal::Lan, + refusal: Some(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, + refusal: Some(Refusal::Lan), source: anyhow::anyhow!("/.cluster request refused: LAN discovery is not enabled"), }), } @@ -237,11 +248,7 @@ pub async fn supervise( }; tokio::select! { err = session.closed() => { - let reason = match &err { - moq_net::Error::Cancel => "closed".to_string(), - other => other.to_string(), - }; - lease.close(reason, session_bytes(&session)); + lease.close_session(&err, session_bytes(&session)); return Err(err.into()); } why = lease.ended() => { @@ -273,3 +280,48 @@ pub(crate) fn session_bytes(session: &moq_net::Session) -> moq_auth::Bytes { received: stats.bytes_received.unwrap_or_default(), } } + +/// Close classification comes only from typed protocol and transport causes. +pub(crate) fn failure(error: &moq_net::Error) -> Option { + use moq_net::stats::Failure; + use moq_net::{Error, SessionError}; + match error { + Error::Cancel + | Error::Closed + | Error::GoingAway + | Error::SessionClosed + | Error::Session(SessionError::Cancel) => None, + Error::IdleTimeout => Some(Failure::IdleTimeout), + Error::Transport(_) => Some(Failure::Transport), + _ => Some(Failure::Application), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use moq_net::{Error, SessionError, stats::Failure}; + + #[test] + fn clean_closes_are_not_failures_and_typed_causes_keep_their_kind() { + for error in [ + Error::Cancel, + Error::Closed, + Error::GoingAway, + Error::SessionClosed, + Error::Session(SessionError::Cancel), + ] { + assert_eq!(failure(&error), None); + } + assert_eq!(failure(&Error::IdleTimeout), Some(Failure::IdleTimeout)); + assert_eq!( + failure(&Error::Transport("idle timeout".into())), + Some(Failure::Transport), + "messages never choose the kind" + ); + assert_eq!( + failure(&Error::Session(SessionError::App(42))), + Some(Failure::Application) + ); + } +} diff --git a/rs/moq-relay/src/internal.rs b/rs/moq-relay/src/internal.rs index b690f9e328..e1df6d0ba6 100644 --- a/rs/moq-relay/src/internal.rs +++ b/rs/moq-relay/src/internal.rs @@ -527,6 +527,23 @@ fn render_metrics( ); } + let _ = writeln!( + out, + "# HELP moq_relay_sessions_failed_total Admitted sessions ending abnormally, by kind." + ); + let _ = writeln!(out, "# TYPE moq_relay_sessions_failed_total counter"); + for (tier, presence) in snap.sessions() { + for &kind in moq_net::stats::Failure::ALL { + let _ = writeln!( + out, + "moq_relay_sessions_failed_total{{tier=\"{}\",kind=\"{}\"}} {}", + tier.as_str(), + kind.as_str(), + presence.sessions_failed.count(kind) + ); + } + } + render_accepts(&mut out, listeners); render_uring(&mut out, uring); @@ -560,12 +577,13 @@ fn render_refusals(out: &mut String, refusals: &crate::refusals::Refusals) { "# HELP moq_relay_sessions_refused_total Sessions admission refused, by reason." ); let _ = writeln!(out, "# TYPE moq_relay_sessions_refused_total counter"); + let counts = refusals.snapshot(); for &refusal in Refusal::ALL { let _ = writeln!( out, "moq_relay_sessions_refused_total{{reason=\"{}\"}} {}", refusal.as_str(), - refusals.count(refusal) + counts.count(refusal) ); } } diff --git a/rs/moq-relay/src/refusals.rs b/rs/moq-relay/src/refusals.rs index 7b2a5203ab..1d498c3c63 100644 --- a/rs/moq-relay/src/refusals.rs +++ b/rs/moq-relay/src/refusals.rs @@ -1,55 +1,8 @@ -//! Sessions admission turned away, counted by reason. -//! -//! A refused session never gets a stats context, so without these counts the -//! only trace of one is a log line. `/metrics` renders them. - -use std::sync::{ - Arc, - atomic::{AtomicU64, Ordering}, -}; +//! Relay refusal collection, shared with published stats when enabled. use crate::auth; - -/// Why admission turned a session away. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum Refusal { - /// The decider said no: the auth server, the public rules, or an embedder. - Refused, - /// The decider could not answer: the auth server was unreachable or answered - /// with neither a grant nor a refusal, a decider sent a grant the relay cannot - /// use, or an embedder did not answer (in time, or before the drain deadline). - Unavailable, - /// An embedder refused the request as one it cannot decide. The relay's own - /// deciders never answer this way. - Request, - /// The grant allows nothing the session asked for, or an embedder refused the - /// session as forbidden. - Forbidden, - /// A LAN peer's membership proof was missing or wrong, or LAN discovery is off. - Lan, -} - -impl Refusal { - /// Every reason, so `/metrics` can list each one from zero. - pub(crate) const ALL: &'static [Refusal] = &[ - Refusal::Refused, - Refusal::Unavailable, - Refusal::Request, - Refusal::Forbidden, - Refusal::Lan, - ]; - - /// The `reason` label value. - pub(crate) const fn as_str(self) -> &'static str { - match self { - Refusal::Refused => "refused", - Refusal::Unavailable => "unavailable", - Refusal::Request => "request", - Refusal::Forbidden => "forbidden", - Refusal::Lan => "lan", - } - } -} +pub(crate) use moq_net::stats::Refusal; +use moq_net::stats::{Registry, Tier}; impl From<&auth::Error> for Refusal { // Exhaustive on purpose: a new `auth::Error` variant must pick its reason here. @@ -63,54 +16,36 @@ impl From<&auth::Error> for Refusal { } } -/// Refused-session counts by [`Refusal`]. Clones share the counts. -#[derive(Clone, Default)] -pub(crate) struct Refusals(Arc); +/// Refusal-only collection when publishing is disabled, otherwise the shared registry. +#[derive(Clone)] +pub(crate) struct Refusals(Registry); -#[derive(Default)] -struct Counts { - refused: AtomicU64, - unavailable: AtomicU64, - request: AtomicU64, - forbidden: AtomicU64, - lan: AtomicU64, +impl Default for Refusals { + fn default() -> Self { + Self(Registry::new(Default::default())) + } } impl Refusals { - /// Count one refused session. - pub(crate) fn record(&self, refusal: Refusal) { - self.counter(refusal).fetch_add(1, Ordering::Relaxed); + pub(crate) fn new(registry: &Registry) -> Self { + if registry.is_enabled() { + Self(registry.clone()) + } else { + Self::default() + } } - - /// Sessions refused for `refusal` so far. - pub(crate) fn count(&self, refusal: Refusal) -> u64 { - self.counter(refusal).load(Ordering::Relaxed) + pub(crate) fn record(&self, reason: Refusal) { + self.0.tier(Tier::default()).refuse("", reason); } - - fn counter(&self, refusal: Refusal) -> &AtomicU64 { - match refusal { - Refusal::Refused => &self.0.refused, - Refusal::Unavailable => &self.0.unavailable, - Refusal::Request => &self.0.request, - Refusal::Forbidden => &self.0.forbidden, - Refusal::Lan => &self.0.lan, + pub(crate) fn snapshot(&self) -> moq_net::stats::Refusals { + let mut counts = moq_net::stats::Refusals::default(); + for (_, presence) in self.0.snapshot().sessions() { + counts.add(presence.sessions_refused); } + counts } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn clones_share_counts() { - let refusals = Refusals::default(); - let clone = refusals.clone(); - clone.record(Refusal::Refused); - clone.record(Refusal::Refused); - refusals.record(Refusal::Forbidden); - assert_eq!(refusals.count(Refusal::Refused), 2); - assert_eq!(clone.count(Refusal::Forbidden), 1); - assert_eq!(refusals.count(Refusal::Unavailable), 0); + #[cfg(test)] + pub(crate) fn count(&self, reason: Refusal) -> u64 { + self.snapshot().count(reason) } } diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 5df9462027..623b23ecce 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -783,8 +783,11 @@ async fn serve_connection( .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?; + .admit(async { + Ok::<_, crate::auth::Error>(cluster.admit_session(&auth, auth_request.clone()).await) + }) + .await + .inspect_err(|err| cluster.refusals.record(err.into()))??; Ok::<_, crate::auth::Error>((admitted, sessions.register(auth_request))) }) .await @@ -795,7 +798,6 @@ async fn serve_connection( 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 diff --git a/rs/moq-relay/tests/auth_embedder.rs b/rs/moq-relay/tests/auth_embedder.rs index 6c295340a9..61be067868 100644 --- a/rs/moq-relay/tests/auth_embedder.rs +++ b/rs/moq-relay/tests/auth_embedder.rs @@ -147,3 +147,31 @@ async fn a_moved_gateway_tier_retags_its_stats() { assert_eq!(active("gateway"), 0); work.abort(); } + +#[tokio::test(start_paused = true)] +async fn gateway_admission_refuses_once_in_the_shared_stats_registry() { + let cluster = cluster::Cluster::new(cluster::Options::default()).unwrap(); + let publisher = moq_stats::Producer::new(moq_stats::produce::Config::new().with_origin(cluster.origin.clone())); + let registry = publisher.registry().clone(); + let cluster = cluster.with_stats(publisher); + let (auth, mut admissions) = auth::Auth::embedded("edge"); + let (result, ()) = tokio::join!( + cluster.admit(&auth, auth.request(Transport::Srt, "/unattributed")), + async { + admissions.next().await.unwrap().refuse(auth::Error::Refused); + } + ); + assert!(matches!(result, Err(auth::Error::Refused))); + let values = registry.snapshot().sessions(); + assert_eq!(values.len(), 1); + assert_eq!(values[0].0, moq_net::stats::Tier::default()); + assert_eq!(values[0].1.sessions_refused.refused, 1); + assert_eq!(values[0].1.sessions_started, 0); + let mut report = moq_net::stats::Report::default(); + registry.report(&mut report); + assert_eq!( + report.sessions[0].root.as_str(), + "", + "refused admission cannot claim the dialed root" + ); +} diff --git a/rs/moq-relay/tests/refusals.rs b/rs/moq-relay/tests/refusals.rs index bd9e54f18d..61a03c09f6 100644 --- a/rs/moq-relay/tests/refusals.rs +++ b/rs/moq-relay/tests/refusals.rs @@ -39,6 +39,8 @@ fn config(dir: &std::path::Path, ws: bool) -> Config { } struct Running { + stats: moq_net::stats::Registry, + origin: moq_net::origin::Consumer, quic: SocketAddr, #[cfg_attr(not(feature = "websocket"), allow(dead_code))] http: SocketAddr, @@ -50,6 +52,8 @@ struct Running { async fn start(config: Config) -> Running { let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); let relay = Relay::load(config).await.expect("load relay"); + let stats = relay.stats().registry().clone(); + let origin = relay.cluster().origin.consume(); let quic = relay.quic_addr().expect("quic listener bound"); let http = relay.web_addrs().http.expect("http listener bound"); let internal = relay.internal().addr().expect("internal listener bound"); @@ -58,6 +62,8 @@ async fn start(config: Config) -> Running { let running = tokio::spawn(relay.run()); ready.wait().await.expect("relay ready"); Running { + stats, + origin, quic, http, metrics: format!("http://{internal}/metrics"), @@ -200,3 +206,152 @@ async fn unavailable_auth_server_is_counted() { assert_eq!(relay.refused("refused").await, 0); relay.stop().await; } + +/// A refusal creates the root stats group even when no session was ever admitted. +#[tokio::test] +async fn refused_only_stats_depth_one_has_root_totals_and_shared_embedder_metrics() { + use moq_net::stats::{Refusal, Tier}; + let dir = tempfile::tempdir().unwrap(); + let mut config = config(dir.path(), false); + config.auth.public = vec!["anon/**".parse().unwrap()]; + config.stats.enabled = true; + config.stats.depth = 1; + config.stats.node = Some("test".into()); + let relay = start(config).await; + assert_refused(format!("https://{}/rooms", relay.quic).parse().unwrap(), false).await; + assert_eq!( + relay.refused("refused").await, + 1, + "native admission counts exactly once" + ); + tokio::time::timeout(TIMEOUT, relay.origin.routed(".stats/node/test")) + .await + .unwrap() + .unwrap(); + let broadcast = tokio::time::timeout(TIMEOUT, relay.origin.request_broadcast(".stats/node/test", None)) + .await + .unwrap() + .unwrap(); + let consumer = moq_stats::Consumer::new(broadcast, Default::default()); + let mut totals = consumer.totals().await.unwrap(); + tokio::time::timeout(TIMEOUT, async { + loop { + let frame = totals.next().await.unwrap().unwrap(); + if frame + .get("") + .is_some_and(|total| total.sessions.sessions_refused.refused == 1) + { + break; + } + } + }) + .await + .expect("refusal-only root totals"); + relay + .stats + .tier(Tier::new("gateway")) + .refuse("tenant", Refusal::Forbidden); + assert_eq!( + relay.refused("forbidden").await, + 1, + "embedder stats and metrics share the counter" + ); + tokio::time::timeout(TIMEOUT, relay.origin.routed(".stats/tenant/node/test")) + .await + .unwrap() + .unwrap(); + let broadcast = tokio::time::timeout(TIMEOUT, relay.origin.request_broadcast(".stats/tenant/node/test", None)) + .await + .unwrap() + .unwrap(); + let consumer = moq_stats::Consumer::new(broadcast, Default::default()); + let mut totals = consumer.totals().await.unwrap(); + tokio::time::timeout(TIMEOUT, async { + loop { + let frame = totals.next().await.unwrap().unwrap(); + if frame + .get("gateway") + .is_some_and(|total| total.sessions.sessions_refused.forbidden == 1) + { + break; + } + } + }) + .await + .expect("embedder outcome in its own tier and root group"); + relay.stop().await; +} + +/// Typed peer closes flow through the actual QUIC listener into failure stats. +#[tokio::test] +async fn peer_zero_close_is_clean_and_application_close_counts_once() { + let dir = tempfile::tempdir().unwrap(); + let mut config = config(dir.path(), false); + config.auth.public = vec![moq_auth::Pattern::all()]; + config.stats.enabled = true; + config.stats.node = Some("test".into()); + let relay = start(config).await; + tokio::time::timeout(TIMEOUT, relay.origin.routed(".stats/node/test")) + .await + .unwrap() + .unwrap(); + let broadcast = relay.origin.request_broadcast(".stats/node/test", None).await.unwrap(); + let consumer = moq_stats::Consumer::new(broadcast, Default::default()); + let mut totals = consumer.totals().await.unwrap(); + for (index, error) in [moq_net::Error::Cancel, moq_net::Error::App(42)] + .into_iter() + .enumerate() + { + let connection = client() + .with_reconnect(false) + .connect(format!("https://{}/", relay.quic).parse::().unwrap()); + let session = connection.established().await.unwrap(); + session.abort(error); + tokio::time::timeout(TIMEOUT, async { + loop { + let frame = totals.next().await.unwrap().unwrap(); + if let Some(total) = frame.get("") + && total.sessions.sessions_ended > index as u64 + { + assert_eq!(total.sessions.sessions_failed.application, index as u64); + assert_eq!(total.sessions.sessions_failed.idle_timeout, 0); + assert_eq!(total.sessions.sessions_failed.transport, 0); + break; + } + } + }) + .await + .expect("peer close reaches the session stats"); + } + relay.stop().await; +} + +#[tokio::test] +async fn explicit_auth_server_refusal_counts_once() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/", listener.local_addr().unwrap()); + let app = axum::Router::new().route( + "/", + axum::routing::post(|| async { axum::http::StatusCode::UNAUTHORIZED }), + ); + let (stop, stopped) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = stopped.await; + }) + .await + .unwrap(); + }); + let dir = tempfile::tempdir().unwrap(); + let mut config = config(dir.path(), false); + config.auth.url = Some(url.parse().unwrap()); + config.stats.enabled = true; + let relay = start(config).await; + assert_refused(format!("https://{}/room", relay.quic).parse().unwrap(), false).await; + assert_eq!(relay.refused("refused").await, 1); + assert_eq!(relay.stats.snapshot().sessions()[0].1.sessions_refused.refused, 1); + relay.stop().await; + stop.send(()).unwrap(); + server.await.unwrap(); +} diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index 01585b950c..c0eae2c602 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -361,7 +361,9 @@ impl Mergeable for Presence { } fn regressed(&self, prev: &Self) -> bool { - self.requests < prev.requests + self.sessions_refused.regressed(&prev.sessions_refused) + || self.sessions_failed.regressed(&prev.sessions_failed) + || self.requests < prev.requests || self.sessions_started < prev.sessions_started || self.sessions_ended < prev.sessions_ended || self.announces_peak < prev.announces_peak @@ -1825,6 +1827,65 @@ mod tests { (merged, (path, None)) } + #[test] + fn outcomes_survive_node_departure_resume_and_counter_only_reset() { + let (mut merged, key) = hand_merged::(); + let sample = |refused, failed| { + let mut value = Presence::default(); + value.sessions_refused.forbidden = refused; + value.sessions_failed.transport = failed; + value + }; + let path = PathOwned::from(".stats/acme/node/b"); + merged.apply_announce( + moq_net::announce::Announce { + prefix: path.clone(), + captures: None, + route: origin::Route::default(), + }, + true, + ); + let second = (path, None); + merged + .nodes + .get_mut(&second) + .unwrap() + .observe(BTreeMap::from([(String::new(), sample(2, 1))])); + + merged + .nodes + .get_mut(&key) + .unwrap() + .observe(BTreeMap::from([(String::new(), sample(5, 3))])); + merged.nodes.get_mut(&key).unwrap().depart(); + assert_eq!(merged.merged()[""], sample(7, 4)); + merged + .nodes + .get_mut(&key) + .unwrap() + .observe(BTreeMap::from([(String::new(), sample(5, 3))])); + assert_eq!( + merged.merged()[""], + sample(7, 4), + "same epoch does not count old errors again" + ); + merged + .nodes + .get_mut(&key) + .unwrap() + .observe(BTreeMap::from([(String::new(), sample(6, 4))])); + merged + .nodes + .get_mut(&key) + .unwrap() + .observe(BTreeMap::from([(String::new(), sample(2, 1))])); + assert_eq!( + merged.merged()[""], + sample(10, 6), + "outcome-only reset starts a fresh segment" + ); + } + #[test] fn a_reset_closes_the_old_open_pairs() { let (mut merged, key) = hand_merged::(); diff --git a/rs/moq-stats/src/lib.rs b/rs/moq-stats/src/lib.rs index ab7dacdce5..379dbd1150 100644 --- a/rs/moq-stats/src/lib.rs +++ b/rs/moq-stats/src/lib.rs @@ -42,7 +42,7 @@ use std::collections::BTreeMap; /// Counter collection, re-exported from [`moq_net::stats`] so stats consumers /// can depend on this crate alone. -pub use moq_net::stats::{Handle, Presence, Registry, Role, Tier, Traffic}; +pub use moq_net::stats::{Failure, Failures, Handle, Presence, Refusal, Refusals, Registry, Role, Tier, Traffic}; use moq_net::{AsPath, Path, PathOwned}; diff --git a/rs/moq-tokio/src/transport/owned.rs b/rs/moq-tokio/src/transport/owned.rs index 44d98ef0d7..de83c945e5 100644 --- a/rs/moq-tokio/src/transport/owned.rs +++ b/rs/moq-tokio/src/transport/owned.rs @@ -9,6 +9,9 @@ use web_transport_trait::{self as wt, poll as backend}; pub struct Error(E); impl moq::Error for Error { + fn close_kind(&self) -> moq::CloseKind { + close_kind(&self.0) + } fn session_error(&self) -> Option<(u32, String)> { self.0.session_error() } @@ -128,3 +131,64 @@ impl poll::RecvStream for super::RecvStream { backend::RecvStream::poll_closed(self, cx).map_err(Error) } } + +fn close_kind(error: &(dyn std::error::Error + 'static)) -> moq::CloseKind { + #[cfg(feature = "noq")] + if let Some(error) = error.downcast_ref::() { + use web_transport_moq::{SessionError, noq::ConnectionError}; + return match error { + SessionError::ConnectionError(ConnectionError::TimedOut) => moq::CloseKind::IdleTimeout, + SessionError::ConnectionError(ConnectionError::LocallyClosed) => moq::CloseKind::Closed, + _ => moq::CloseKind::Transport, + }; + } + #[cfg(feature = "iroh")] + if let Some(error) = error.downcast_ref::() { + use web_transport_iroh::iroh::endpoint::ConnectionError; + return match error { + ConnectionError::TimedOut => moq::CloseKind::IdleTimeout, + ConnectionError::LocallyClosed => moq::CloseKind::Closed, + _ => moq::CloseKind::Transport, + }; + } + error.source().map(close_kind).unwrap_or_default() +} + +#[cfg(all(test, feature = "noq"))] +mod outcome_tests { + use super::*; + use moq_net::transport::Error as _; + use web_transport_moq::{SessionError, noq::ConnectionError}; + + #[test] + fn typed_idle_and_local_closes_survive_session_and_stream_wrappers() { + let idle = Error(SessionError::ConnectionError(ConnectionError::TimedOut)); + assert!(matches!( + moq_net::Error::from_transport(idle), + moq_net::Error::IdleTimeout + )); + let idle = Error(web_transport_moq::ReadError::SessionError( + SessionError::ConnectionError(ConnectionError::TimedOut), + )); + assert_eq!(idle.close_kind(), moq::CloseKind::IdleTimeout); + let local = Error(SessionError::ConnectionError(ConnectionError::LocallyClosed)); + assert!(matches!(moq_net::Error::from_transport(local), moq_net::Error::Closed)); + } +} + +#[cfg(all(test, feature = "iroh"))] +mod iroh_outcome_tests { + use super::*; + use web_transport_iroh::{SessionError, iroh::endpoint::ConnectionError}; + + #[test] + fn typed_iroh_close_kinds_survive_the_error_source_chain() { + let idle = Error(SessionError::ConnectionError(ConnectionError::TimedOut)); + assert!(matches!( + moq_net::Error::from_transport(idle), + moq_net::Error::IdleTimeout + )); + let local = Error(SessionError::ConnectionError(ConnectionError::LocallyClosed)); + assert!(matches!(moq_net::Error::from_transport(local), moq_net::Error::Closed)); + } +} diff --git a/rs/moq-uring/src/quic/mod.rs b/rs/moq-uring/src/quic/mod.rs index 9ed2b81116..ea6c2e4e1d 100644 --- a/rs/moq-uring/src/quic/mod.rs +++ b/rs/moq-uring/src/quic/mod.rs @@ -244,6 +244,9 @@ pub enum Error { /// missing or unwritable, or the writer thread could not be started. #[error("qlog error: {0}")] Qlog(String), + /// This endpoint closed the connection locally. + #[error("locally closed")] + Closed, } impl web_transport_trait::Error for Error { diff --git a/rs/moq-uring/src/quic/noq/mod.rs b/rs/moq-uring/src/quic/noq/mod.rs index 00d457bcdb..31a4728954 100644 --- a/rs/moq-uring/src/quic/noq/mod.rs +++ b/rs/moq-uring/src/quic/noq/mod.rs @@ -70,6 +70,7 @@ impl From for Error { reason: err.reason.clone(), }, ConnectionError::TimedOut => Self::TimedOut, + ConnectionError::LocallyClosed => Self::Closed, // The rest (a stateless reset, an unsupported version, exhausted // ids) have no code to report, and a local close is published // where it happens rather than waited for as an event. diff --git a/rs/moq-uring/src/transport/adapter.rs b/rs/moq-uring/src/transport/adapter.rs index 0a751d7671..99de76699a 100644 --- a/rs/moq-uring/src/transport/adapter.rs +++ b/rs/moq-uring/src/transport/adapter.rs @@ -9,6 +9,9 @@ use web_transport_trait::{self as wt, poll as backend}; pub struct Error(E); impl moq::Error for Error { + fn close_kind(&self) -> moq::CloseKind { + close_kind(&self.0) + } fn session_error(&self) -> Option<(u32, String)> { self.0.session_error() } @@ -150,6 +153,18 @@ impl poll::RecvStream for RecvStream { } } +fn close_kind(error: &(dyn std::error::Error + 'static)) -> moq::CloseKind { + #[cfg(all(target_os = "linux", feature = "noq"))] + if let Some(error) = error.downcast_ref::() { + return match error { + crate::quic::Error::TimedOut => moq::CloseKind::IdleTimeout, + crate::quic::Error::Closed => moq::CloseKind::Closed, + _ => moq::CloseKind::Transport, + }; + } + error.source().map(close_kind).unwrap_or_default() +} + #[cfg(test)] mod tests { use super::*; @@ -311,3 +326,19 @@ mod tests { assert_eq!(state.get(), 21); } } + +#[cfg(all(test, target_os = "linux", feature = "noq"))] +mod outcome_tests { + use super::*; + + #[test] + fn typed_ring_close_kinds_survive_the_adapter() { + let idle = Error(crate::quic::Error::TimedOut); + assert!(matches!( + moq_net::Error::from_transport(idle), + moq_net::Error::IdleTimeout + )); + let local = Error(crate::quic::Error::Closed); + assert!(matches!(moq_net::Error::from_transport(local), moq_net::Error::Closed)); + } +} diff --git a/rs/moq-uring/tests/transport.rs b/rs/moq-uring/tests/transport.rs index 84fe4df6e5..ed93d67e99 100644 --- a/rs/moq-uring/tests/transport.rs +++ b/rs/moq-uring/tests/transport.rs @@ -1,3 +1,6 @@ +#[cfg(all(target_os = "linux", feature = "noq"))] +use moq_uring::quic; + // Exercise the transport adapter on native hosts without a browser or ring. #[path = "../src/transport/adapter.rs"] mod adapter; From d1d7fd149005570e4218e80c1a6a239ccbd39660 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 19:24:33 -0700 Subject: [PATCH 03/16] fix(net): keep virtual stream closure out of connection metadata Co-Authored-By: GPT-6 --- rs/moq-net/src/coding/codes.rs | 12 ++++++++++++ rs/moq-net/src/error.rs | 2 +- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/coding/codes.rs b/rs/moq-net/src/coding/codes.rs index fad36ce85d..860419f14c 100644 --- a/rs/moq-net/src/coding/codes.rs +++ b/rs/moq-net/src/coding/codes.rs @@ -174,6 +174,18 @@ mod tests { } } + /// `Closed` also describes virtual streams, so it cannot claim that the transport's + /// connection closed. Native transport adapters supply that metadata themselves. + #[test] + fn a_virtual_stream_close_does_not_claim_the_connection_closed() { + let version = ietf::Version::Draft16; + assert!(matches!(version.transport_error(Error::Closed), Error::Transport(_))); + assert!(matches!( + version.transport_error(Error::IdleTimeout), + Error::IdleTimeout + )); + } + /// A session close is not stream-scoped, so it keeps the session registry whichever /// wire the stream is on. #[test] diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index b9d6870d41..1139baefc9 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -678,7 +678,7 @@ impl crate::transport::Error for Error { fn close_kind(&self) -> crate::transport::CloseKind { match self { Self::IdleTimeout => crate::transport::CloseKind::IdleTimeout, - Self::Closed => crate::transport::CloseKind::Closed, + // Closed also ends model handles and virtual streams, not just connections. _ => crate::transport::CloseKind::Transport, } } From c93e963a418ddb887e9311b69aa465d1b6d69f13 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 19:33:11 -0700 Subject: [PATCH 04/16] fix(net): preserve the session boundary on local transport close Co-Authored-By: GPT-6 --- rs/moq-net/src/error.rs | 3 ++- rs/moq-net/src/model/resume.rs | 4 ++-- rs/moq-tokio/src/transport/owned.rs | 10 ++++++++-- rs/moq-uring/src/transport/adapter.rs | 5 ++++- 4 files changed, 16 insertions(+), 6 deletions(-) diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 1139baefc9..08d36af525 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -561,7 +561,8 @@ impl Error { match err.close_kind() { crate::transport::CloseKind::IdleTimeout => Self::IdleTimeout, - crate::transport::CloseKind::Closed => Self::Closed, + // A connection ending must not become the model's clean end-of-content. + crate::transport::CloseKind::Closed => SessionError::Cancel.into(), _ => Self::Transport(err.to_string()), } } diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index a165f151b9..4a3df60bd1 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -652,8 +652,8 @@ impl Reader { None | Some(Some(_)) => return Some(Poll::Ready(Ok(None))), Some(None) => {} }, - // No front is left to replace the copy: the track ends as it does. A copy whose - // session closed locally ends it cleanly. + // No front is left to replace the copy: the track ends as it does. A locally + // closed copy ends it cleanly; a connection ending retains its session error. Some(None) => match serving.map(&done) { None => return Some(Poll::Ready(Err(Error::Dropped))), Some(Some(Ok(()) | Err(Error::Closed))) => return Some(Poll::Ready(Ok(None))), diff --git a/rs/moq-tokio/src/transport/owned.rs b/rs/moq-tokio/src/transport/owned.rs index de83c945e5..895dbefe90 100644 --- a/rs/moq-tokio/src/transport/owned.rs +++ b/rs/moq-tokio/src/transport/owned.rs @@ -172,7 +172,10 @@ mod outcome_tests { )); assert_eq!(idle.close_kind(), moq::CloseKind::IdleTimeout); let local = Error(SessionError::ConnectionError(ConnectionError::LocallyClosed)); - assert!(matches!(moq_net::Error::from_transport(local), moq_net::Error::Closed)); + assert!(matches!( + moq_net::Error::from_transport(local), + moq_net::Error::Session(moq_net::SessionError::Cancel) + )); } } @@ -189,6 +192,9 @@ mod iroh_outcome_tests { moq_net::Error::IdleTimeout )); let local = Error(SessionError::ConnectionError(ConnectionError::LocallyClosed)); - assert!(matches!(moq_net::Error::from_transport(local), moq_net::Error::Closed)); + assert!(matches!( + moq_net::Error::from_transport(local), + moq_net::Error::Session(moq_net::SessionError::Cancel) + )); } } diff --git a/rs/moq-uring/src/transport/adapter.rs b/rs/moq-uring/src/transport/adapter.rs index 99de76699a..2d8e89baaa 100644 --- a/rs/moq-uring/src/transport/adapter.rs +++ b/rs/moq-uring/src/transport/adapter.rs @@ -339,6 +339,9 @@ mod outcome_tests { moq_net::Error::IdleTimeout )); let local = Error(crate::quic::Error::Closed); - assert!(matches!(moq_net::Error::from_transport(local), moq_net::Error::Closed)); + assert!(matches!( + moq_net::Error::from_transport(local), + moq_net::Error::Session(moq_net::SessionError::Cancel) + )); } } From 74942bc9dabe7e5fb47501e0cddb854757389b15 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 00:00:34 -0700 Subject: [PATCH 05/16] docs(stats): show outcome counters in the example and trim to reader-facing rules Add sessions_refused and sessions_failed to the sessions JSON example, cut cluster-dial and setup-error internals from stats.md, and move the sessions_failed /metrics note into the existing /metrics alert list. Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/config.md | 6 ------ doc/bin/relay/http.md | 4 ++++ doc/concept/stats.md | 35 +++++++++++++++++++---------------- 3 files changed, 23 insertions(+), 22 deletions(-) diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index cc3707ed9f..f4a5099962 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -276,12 +276,6 @@ per-session limits, past which one is closed with `TOO_MANY_REQUESTS`. [Stats](/concept/stats) describes the paths, tracks, and encodings; read them with the [`moq-stats`](https://docs.rs/moq-stats) crate. -Session outcome counters in the stats feed distinguish admission refusals from -abnormal closes; see [session stats](/concept/stats#sessions). `/metrics` exposes -`moq_relay_sessions_refused_total` by reason and -`moq_relay_sessions_failed_total` by tier and kind. Refusals remain counted when -stats publishing is disabled. - ## \[iroh] Iroh requires building the relay from source with `cargo build --release -p moq-relay --features iroh`. Published binaries, Docker images, and the Nix package leave it out. diff --git a/doc/bin/relay/http.md b/doc/bin/relay/http.md index 6f9af1776e..d0f5fa6f8c 100644 --- a/doc/bin/relay/http.md +++ b/doc/bin/relay/http.md @@ -54,6 +54,10 @@ A few worth alerting on: - `moq_relay_sessions_refused_total`, by `reason`. A rise in `unavailable` points at the auth server; a rise in `refused` at the credentials clients present. Each attempt counts, so a client that retries is counted again. +- `moq_relay_sessions_failed_total`, by `tier` and `kind` (`idle_timeout`, + `transport`, `application`, `expired`): admitted sessions that ended + abnormally. A clean close is not a failure. Reported only while + [stats](/bin/relay/config#stats) are enabled. - `moq_relay_stale_bytes_total` and friends: content dropped for drifting past a subscriber's latency budget. - `moq_relay_draining_sessions`: sessions sent a GOAWAY during a diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 421b37a218..43c1983a16 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -171,7 +171,15 @@ should prefer the canonical name and fall back to the legacy one. ### Sessions ```json -{ "acme": { "requests": 3, "sessions_started": 12, "sessions_ended": 10, "sessions": 12, "sessions_closed": 10, "announces_peak": 40, "subscriptions_peak": 900 } } +{ + "acme": { + "requests": 3, "sessions_started": 12, "sessions_ended": 10, + "sessions": 12, "sessions_closed": 10, + "sessions_refused": { "refused": 0, "unavailable": 0, "request": 0, "forbidden": 0, "lan": 0 }, + "sessions_failed": { "idle_timeout": 1, "transport": 0, "application": 0, "expired": 0 }, + "announces_peak": 40, "subscriptions_peak": 900 + } +} ``` `sessions_started` and `sessions_ended` count connects and disconnects, whether @@ -181,21 +189,16 @@ group on the tier for `totals.json`. `sessions` and `sessions_closed` are their legacy spellings. A session moved to a new tier ends on the old one and starts on the new. -`sessions_refused` counts refusals by `refused`, `unavailable`, `request`, -`forbidden`, and `lan`. Relay admission refusals have no resolved grant, so they -belong to the empty root on the default tier. An embedder that refuses a session -outside relay admission can record it on its stats handle under the root and tier -it resolved. Refused sessions never increase the live session count. - -`sessions_failed` counts abnormal ends by `idle_timeout`, `transport`, -`application`, and `expired`. A clean peer close, local close, drain, or shutdown -is not a failure. These are subsets of the sessions that end, leaving the live -count unchanged. A reconnecting cluster dial retains one logical session across -physical reconnects, so intermediate retry statuses add no failure; its typed -terminal session error counts once when that logical session ends. Initial -setup errors and configuration failures do not become admitted-session failures. -Both objects default to zero when reading an older publisher; -their counters survive pruning and follow the node epoch and reset rules. +`sessions_refused` counts sessions turned away before admission, by reason. +They never start a session. The relay's own refusals have no resolved root, so +they count under the empty root on the default tier; an embedder can count one +it refuses itself with `stats::Handle::refuse`, under the root and tier it +resolved. + +`sessions_failed` counts admitted sessions that ended abnormally, by kind: a +subset of `sessions_ended`, so the live count is unchanged. A clean close by +either side, a drain, or a shutdown is not a failure. A reader treats a missing +object as zeros. `requests` counts admitted request/response exchanges under the token's root and tier, including HTTP `/fetch` and `/announced`. They never start or end a session. From d3c10b1de3c0f754815afc80c58eaf3b2eb2a471 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 11:57:03 -0700 Subject: [PATCH 06/16] fix(net): classify moq-quic close kinds after the backend move Main replaced the noq backend with moq-quic, so the noq-gated close classification no longer compiled. Gate it on the quic feature and match moq-tokio's own WebTransport session error. Also tidy Lease::close_session, restore a misplaced test doc comment, and drop a stale driver comment. Co-Authored-By: Claude Opus 5.5 --- quest/m1/refusal-reason.md | 1 - rs/moq-net/src/stats/outcomes.rs | 8 ++++++++ rs/moq-relay/src/auth.rs | 14 +++++++------- rs/moq-relay/src/cluster.rs | 6 +++--- rs/moq-tokio/src/transport/owned.rs | 12 ++++++------ rs/moq-uring/src/quic/driver/mod.rs | 3 +-- rs/moq-uring/src/transport/adapter.rs | 4 ++-- rs/moq-uring/tests/transport.rs | 2 +- 8 files changed, 28 insertions(+), 22 deletions(-) diff --git a/quest/m1/refusal-reason.md b/quest/m1/refusal-reason.md index 4aa2030456..e8ed0e34ac 100644 --- a/quest/m1/refusal-reason.md +++ b/quest/m1/refusal-reason.md @@ -55,7 +55,6 @@ Decided 2026-10-08: Public API: moq-auth's refusal error and the HTTP refusal body. Wire: the auth HTTP contract only, not MoQ. - ## Related - [Expired token error](/quest/m1/auth/expired-error.md) - the MoQ wire diff --git a/rs/moq-net/src/stats/outcomes.rs b/rs/moq-net/src/stats/outcomes.rs index a4042439c6..86cb3567ef 100644 --- a/rs/moq-net/src/stats/outcomes.rs +++ b/rs/moq-net/src/stats/outcomes.rs @@ -18,6 +18,7 @@ pub enum Refusal { /// The LAN membership proof was missing or invalid. Lan, } + impl Refusal { /// Every currently defined kind, in display order. pub const ALL: &'static [Self] = &[ @@ -55,6 +56,7 @@ pub struct Refusals { /// The LAN membership proof was missing or invalid. pub lan: u64, } + impl Refusals { /// The cumulative count for this kind. pub fn count(&self, kind: Refusal) -> u64 { @@ -83,6 +85,7 @@ impl Refusals { || self.lan < previous.lan } } + #[derive(Debug, Default)] pub(super) struct RefusalCounters { refused: AtomicU64, @@ -91,6 +94,7 @@ pub(super) struct RefusalCounters { forbidden: AtomicU64, lan: AtomicU64, } + impl RefusalCounters { pub(super) fn counter(&self, kind: Refusal) -> &AtomicU64 { match kind { @@ -125,6 +129,7 @@ pub enum Failure { /// The session credential expired. Expired, } + impl Failure { /// Every currently defined kind, in display order. pub const ALL: &'static [Self] = &[Self::IdleTimeout, Self::Transport, Self::Application, Self::Expired]; @@ -153,6 +158,7 @@ pub struct Failures { /// The session credential expired. pub expired: u64, } + impl Failures { /// The cumulative count for this kind. pub fn count(&self, kind: Failure) -> u64 { @@ -178,6 +184,7 @@ impl Failures { || self.expired < previous.expired } } + #[derive(Debug, Default)] pub(super) struct FailureCounters { idle_timeout: AtomicU64, @@ -185,6 +192,7 @@ pub(super) struct FailureCounters { application: AtomicU64, expired: AtomicU64, } + impl FailureCounters { pub(super) fn counter(&self, kind: Failure) -> &AtomicU64 { match kind { diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 691d55b265..f6709c5707 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -532,14 +532,14 @@ impl Lease { pub(crate) fn close_session(self, error: &moq_net::Error, bytes: Bytes) { let stats = self.stats.clone(); - let reason = if crate::connection::failure(error).is_none() { - "closed".to_string() - } else { - error.to_string() + let failure = crate::connection::failure(error); + let reason = match failure { + Some(_) => error.to_string(), + None => "closed".to_string(), }; - let reason = self.close(reason, bytes); - if reason != lease::Reason::Expired - && let Some(kind) = crate::connection::failure(error) + // An expired lease already recorded its own failure kind. + if self.close(reason, bytes) != lease::Reason::Expired + && let Some(kind) = failure { stats.fail(kind); } diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 8c25db6281..9a87336e3a 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -2228,9 +2228,6 @@ mod tests { drop(broadcast); } - /// `?cost=` is read and stripped (it rides SETUP, not the URL), other - /// query params survive, and a garbage value is an error rather than a - /// silent default. #[test] fn remote_logical_outcome_ignores_retries_and_counts_one_terminal_close() { use moq_net::stats::{Failure, Registry}; @@ -2278,6 +2275,9 @@ mod tests { assert_eq!(presence.sessions_failed.application, 0); } + /// `?cost=` is read and stripped (it rides SETUP, not the URL), other + /// query params survive, and a garbage value is an error rather than a + /// silent default. #[test] fn cost_param_is_consumed() { let mut url = Url::parse("https://peer.example/?jwt=abc&cost=0").unwrap(); diff --git a/rs/moq-tokio/src/transport/owned.rs b/rs/moq-tokio/src/transport/owned.rs index 895dbefe90..857cf35a62 100644 --- a/rs/moq-tokio/src/transport/owned.rs +++ b/rs/moq-tokio/src/transport/owned.rs @@ -133,9 +133,9 @@ impl poll::RecvStream for super::RecvStream { } fn close_kind(error: &(dyn std::error::Error + 'static)) -> moq::CloseKind { - #[cfg(feature = "noq")] - if let Some(error) = error.downcast_ref::() { - use web_transport_moq::{SessionError, noq::ConnectionError}; + #[cfg(feature = "quic")] + if let Some(error) = error.downcast_ref::() { + use crate::quic::{driver::ConnectionError, web::SessionError}; return match error { SessionError::ConnectionError(ConnectionError::TimedOut) => moq::CloseKind::IdleTimeout, SessionError::ConnectionError(ConnectionError::LocallyClosed) => moq::CloseKind::Closed, @@ -154,11 +154,11 @@ fn close_kind(error: &(dyn std::error::Error + 'static)) -> moq::CloseKind { error.source().map(close_kind).unwrap_or_default() } -#[cfg(all(test, feature = "noq"))] +#[cfg(all(test, feature = "quic"))] mod outcome_tests { use super::*; + use crate::quic::{driver::ConnectionError, web::SessionError}; use moq_net::transport::Error as _; - use web_transport_moq::{SessionError, noq::ConnectionError}; #[test] fn typed_idle_and_local_closes_survive_session_and_stream_wrappers() { @@ -167,7 +167,7 @@ mod outcome_tests { moq_net::Error::from_transport(idle), moq_net::Error::IdleTimeout )); - let idle = Error(web_transport_moq::ReadError::SessionError( + let idle = Error(crate::quic::web::ReadError::SessionError( SessionError::ConnectionError(ConnectionError::TimedOut), )); assert_eq!(idle.close_kind(), moq::CloseKind::IdleTimeout); diff --git a/rs/moq-uring/src/quic/driver/mod.rs b/rs/moq-uring/src/quic/driver/mod.rs index 8ff7806428..4cb8d0168f 100644 --- a/rs/moq-uring/src/quic/driver/mod.rs +++ b/rs/moq-uring/src/quic/driver/mod.rs @@ -81,8 +81,7 @@ impl From for Error { ConnectionError::TimedOut => Self::TimedOut, ConnectionError::LocallyClosed => Self::Closed, // The rest (a stateless reset, an unsupported version, exhausted - // ids) have no code to report, and a local close is published - // where it happens rather than waited for as an event. + // ids) have no code to report. err => Self::Quic(err.to_string()), } } diff --git a/rs/moq-uring/src/transport/adapter.rs b/rs/moq-uring/src/transport/adapter.rs index 2d8e89baaa..5ed5ab03d4 100644 --- a/rs/moq-uring/src/transport/adapter.rs +++ b/rs/moq-uring/src/transport/adapter.rs @@ -154,7 +154,7 @@ impl poll::RecvStream for RecvStream { } fn close_kind(error: &(dyn std::error::Error + 'static)) -> moq::CloseKind { - #[cfg(all(target_os = "linux", feature = "noq"))] + #[cfg(all(target_os = "linux", feature = "quic"))] if let Some(error) = error.downcast_ref::() { return match error { crate::quic::Error::TimedOut => moq::CloseKind::IdleTimeout, @@ -327,7 +327,7 @@ mod tests { } } -#[cfg(all(test, target_os = "linux", feature = "noq"))] +#[cfg(all(test, target_os = "linux", feature = "quic"))] mod outcome_tests { use super::*; diff --git a/rs/moq-uring/tests/transport.rs b/rs/moq-uring/tests/transport.rs index ed93d67e99..b5a44691b6 100644 --- a/rs/moq-uring/tests/transport.rs +++ b/rs/moq-uring/tests/transport.rs @@ -1,4 +1,4 @@ -#[cfg(all(target_os = "linux", feature = "noq"))] +#[cfg(all(target_os = "linux", feature = "quic"))] use moq_uring::quic; // Exercise the transport adapter on native hosts without a browser or ring. From 50e067b85026f9c83a54b4faebf444eeb91de587 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 12:11:45 -0700 Subject: [PATCH 07/16] fix(relay): count failed SETUP replies and exhausted peer reconnects Admission starts the stats session, so a SETUP reply that fails before supervision now records its failure kind on the native and io_uring paths. A cluster peer link that gave up reconnecting after it had connected counts as a transport failure. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/cluster.rs | 24 ++++++++++++++++++++++ rs/moq-relay/src/connection.rs | 37 ++++++++++++++++++++++++++++++++-- rs/moq-relay/src/uring.rs | 9 +++++++-- 3 files changed, 66 insertions(+), 4 deletions(-) diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 9a87336e3a..c6c7289194 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -2064,6 +2064,8 @@ fn remote_status( stats.fail(kind); } } + // The retry window ran out after the link was lost, so the link failed to recover. + Err(moq_tokio::Error::Reconnect(_)) if *established => stats.fail(moq_net::stats::Failure::Transport), _ => {} } status @@ -2275,6 +2277,28 @@ mod tests { assert_eq!(presence.sessions_failed.application, 0); } + #[test] + fn remote_reconnect_exhaustion_fails_only_an_established_link() { + use moq_net::stats::{Failure, Registry}; + let exhausted = |established: bool| { + let registry = Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session(""); + let mut connected = false; + if established { + remote_status(&stats, &mut connected, Ok(moq_tokio::Status::Connected)).unwrap(); + } + let gave_up = Err(moq_tokio::Error::Reconnect("gave up".into())); + let _ = remote_status(&stats, &mut connected, gave_up); + drop(stats); + registry.snapshot().sessions()[0] + .1 + .sessions_failed + .count(Failure::Transport) + }; + assert_eq!(exhausted(false), 0); + assert_eq!(exhausted(true), 1); + } + /// `?cost=` is read and stripped (it rides SETUP, not the URL), other /// query params survive, and a garbage value is an error rather than a /// silent default. diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index a9dfd6fad8..4c3418a586 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -131,7 +131,10 @@ 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 = self + .request + .with_stats(stats.clone()) + .with_limits(self.cluster.limits(&lease)); if let Some(subscriber) = subscriber { request = request.with_publisher(subscriber); } @@ -144,7 +147,14 @@ impl Connection { .shutdown .until_deadline(request.ok()) .await - .ok_or_else(|| anyhow::anyhow!("the SETUP reply was still pending at the drain deadline"))??; + .ok_or_else(|| anyhow::anyhow!("the SETUP reply was still pending at the drain deadline"))? + .inspect_err(|err| { + let cause = match err { + moq_tokio::Error::MoqNet(err) => Some(err), + _ => None, + }; + setup_failed(&stats, cause); + })?; tracing::info!(version = %session.version(), %transport, "negotiated"); @@ -283,6 +293,14 @@ pub(crate) fn session_bytes(session: &moq_net::Session) -> moq_auth::Bytes { } } +/// Admission already started the stats session, so a failed SETUP reply ends it. +/// A cause without a MoQ error is the transport's. +pub(crate) fn setup_failed(stats: &moq_net::stats::Session, cause: Option<&moq_net::Error>) { + if let Some(kind) = cause.map_or(Some(moq_net::stats::Failure::Transport), failure) { + stats.fail(kind); + } +} + /// Close classification comes only from typed protocol and transport causes. pub(crate) fn failure(error: &moq_net::Error) -> Option { use moq_net::stats::Failure; @@ -304,6 +322,21 @@ mod tests { use super::*; use moq_net::{Error, SessionError, stats::Failure}; + #[test] + fn a_failed_setup_reply_ends_the_admitted_session_by_its_cause() { + let outcome = |cause: Option<&Error>| { + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(moq_net::stats::Tier::default()).session("root"); + setup_failed(&stats, cause); + drop(stats); + registry.snapshot().sessions()[0].1.sessions_failed + }; + assert_eq!(outcome(Some(&Error::Cancel)), Default::default()); + assert_eq!(outcome(None).count(Failure::Transport), 1); + assert_eq!(outcome(Some(&Error::IdleTimeout)).count(Failure::IdleTimeout), 1); + assert_eq!(outcome(Some(&Error::Timeout)).count(Failure::Application), 1); + } + #[test] fn clean_closes_are_not_failures_and_typed_causes_keep_their_kind() { for error in [ diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 623b23ecce..52a61bc354 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -821,7 +821,9 @@ async fn serve_connection( stats, } = admitted; - let mut request = request.with_stats(stats).with_limits(serve.cluster.limits(&lease)); + let mut request = request + .with_stats(stats.clone()) + .with_limits(serve.cluster.limits(&lease)); if let Some(subscriber) = subscriber { request = request.with_publisher(subscriber); } @@ -829,7 +831,10 @@ async fn serve_connection( request = request.with_subscriber(publisher); } let lease = lease.preparing(request.auth(), request.role())?; - let (session, driver) = deadline.bound(request.ok()).await?; + let (session, driver) = deadline + .bound(request.ok()) + .await + .inspect_err(|err| crate::connection::setup_failed(&stats, err.downcast_ref()))?; let driver_handle = handle.clone(); let driver = async move { match driver_handle.run(driver).await { From ef8589b8f7210edd1477129247c7dee8674e09fa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 19:04:21 -0700 Subject: [PATCH 08/16] test(uring): resolve crate::quic at the merged integration test root #5689 linked the integration tests into one binary, so the adapter source included by the transport module now resolves crate::quic at main.rs. Co-Authored-By: Claude Opus 5.5 --- rs/moq-uring/tests/integration/main.rs | 4 ++++ rs/moq-uring/tests/integration/transport.rs | 3 --- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/rs/moq-uring/tests/integration/main.rs b/rs/moq-uring/tests/integration/main.rs index ae4d797d51..39715fff97 100644 --- a/rs/moq-uring/tests/integration/main.rs +++ b/rs/moq-uring/tests/integration/main.rs @@ -5,6 +5,10 @@ //! `allocations.rs` stays apart because its counting global allocator would //! apply to every test in the binary. +// `transport` includes the adapter source, which names `crate::quic`. +#[cfg(all(target_os = "linux", feature = "quic"))] +use moq_uring::quic; + mod credit; mod echo; mod ecn; diff --git a/rs/moq-uring/tests/integration/transport.rs b/rs/moq-uring/tests/integration/transport.rs index 792e0812b7..eb3353e6f2 100644 --- a/rs/moq-uring/tests/integration/transport.rs +++ b/rs/moq-uring/tests/integration/transport.rs @@ -1,6 +1,3 @@ -#[cfg(all(target_os = "linux", feature = "quic"))] -use moq_uring::quic; - // Exercise the transport adapter on native hosts without a browser or ring. #[path = "../../src/transport/adapter.rs"] mod adapter; From 7f347d21ee865fa2dd941cdc479ad3800711b613 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 22:08:22 -0700 Subject: [PATCH 09/16] fix(relay): classify a refreshed session's end by its union, not its first credential A replacement credential can outlive the first credential's expiry, but the lease consumer keeps that first close reason. A later clean close then counted as an expired failure, and hid a real transport failure. A refreshed session now classifies by the reason that ended the session. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/auth.rs | 22 +++++----- rs/moq-relay/src/auth/refresh.rs | 69 +++++++++++++++++++++++++++----- 2 files changed, 73 insertions(+), 18 deletions(-) diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 829f2ba3c6..f0a3a756d8 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -652,18 +652,22 @@ impl Lease { } /// End the lease with the session's close classification and the totals it - /// moved, and learn what it ended with: that, or the decider's reason if it - /// revoked first. Dropping the consumer reports zero bytes. + /// moved, and learn what ended the session: that, or for a gateway the + /// decider's reason if it revoked first. Dropping the consumer reports zero bytes. pub fn close(self, reason: impl Into, bytes: Bytes) -> lease::Reason { let mut reason = reason.into(); // A gateway has one credential, so its revocation caused the transfer to - // end. A refreshed session instead reports the result of its whole union. - if self.refresh.is_none() - && let std::task::Poll::Ready(ended) = self.consumer.poll_closed(&moq_net::kio::Waiter::noop()) - { - reason = ended; - } - let reason = self.consumer.close(reason, bytes); + // end. A refreshed session instead reports the result of its whole union: + // its first credential may have ended with a reason a replacement outlived. + let reason = if self.refresh.is_none() { + if let std::task::Poll::Ready(ended) = self.consumer.poll_closed(&moq_net::kio::Waiter::noop()) { + reason = ended; + } + self.consumer.close(reason, bytes) + } else { + self.consumer.close(reason.clone(), bytes); + reason + }; if reason == lease::Reason::Expired { self.stats.fail(moq_net::stats::Failure::Expired); } diff --git a/rs/moq-relay/src/auth/refresh.rs b/rs/moq-relay/src/auth/refresh.rs index a9b126a93b..1a5b53cc69 100644 --- a/rs/moq-relay/src/auth/refresh.rs +++ b/rs/moq-relay/src/auth/refresh.rs @@ -489,7 +489,8 @@ mod tests { .unwrap() } - async fn sessions(lease: super::super::Lease) -> (moq_net::Session, tokio::task::JoinHandle) { + /// A client session and the server session `lease` was prepared for, before supervision. + async fn connect(lease: super::super::Lease) -> (moq_net::Session, moq_net::Session, super::super::Lease) { let (client, server) = transports().await; let now = tokio::time::Instant::now().into_std(); let client = async { @@ -517,17 +518,23 @@ mod tests { ) .await .unwrap(); - let mut lease = lease.preparing(request.auth(), request.role()).unwrap(); + let lease = lease.preparing(request.auth(), request.role()).unwrap(); let (session, driver) = request.ok().await.unwrap(); tokio::spawn(moq_net::time::run(driver)); - (tokio::spawn(async move { - let reason = lease.ended().await; - session.abort(super::super::error(&reason)); - lease.close(reason.clone(), Default::default()); - reason - }),) + (session, lease) }; - let (client, (server,)) = tokio::join!(client, server); + let (client, (session, lease)) = tokio::join!(client, server); + (client, session, lease) + } + + async fn sessions(lease: super::super::Lease) -> (moq_net::Session, tokio::task::JoinHandle) { + let (client, session, mut lease) = connect(lease).await; + let server = tokio::spawn(async move { + let reason = lease.ended().await; + session.abort(super::super::error(&reason)); + lease.close(reason.clone(), Default::default()); + reason + }); (client, server) } @@ -660,6 +667,50 @@ mod tests { let _ = supervising.await; } + /// A replacement keeps the session alive past the first credential's expiry, so a + /// later clean close is not an expired session. + #[tokio::test(start_paused = true)] + async fn a_superseded_expiry_does_not_fail_a_clean_close() { + let (auth, mut admissions) = super::super::Auth::embedded("test"); + let (producer, consumer) = lease::Producer::new(grant("a/**", 3000)); + let request = auth.request(moq_auth::Transport::Quic, "/room"); + let (lease, ()) = tokio::join!(auth.admit(request), async { + admissions.next().await.unwrap().grant(consumer); + }); + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(moq_net::stats::Tier::default()).session("room"); + let (client, session, mut lease) = connect(lease.unwrap().with_stats(stats.clone())).await; + let (close, closing) = tokio::sync::oneshot::channel::<()>(); + let server = tokio::spawn(async move { + tokio::select! { + reason = lease.ended() => panic!("the replacement did not keep the session: {reason}"), + _ = closing => lease.close_session(&moq_net::Error::Cancel, Default::default()), + } + }); + + let auth = client.auth(); + let (token, ()) = tokio::join!(auth.add("replacement"), async { + admissions + .next() + .await + .unwrap() + .grant(lease::Consumer::fixed(grant("a/**", 3000))); + }); + // Held, since dropping the handle withdraws the replacement. + let _token = token.unwrap(); + producer.revoke(lease::Reason::Expired); + tokio::time::advance(Duration::from_secs(1)).await; + assert!(!server.is_finished(), "the first credential's expiry ended the session"); + + close.send(()).unwrap(); + server.await.unwrap(); + drop((session, stats)); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_ended, 1); + assert_eq!(presence.sessions_failed, moq_net::stats::Failures::default()); + client.abort(moq_net::Error::Cancel); + } + #[tokio::test(start_paused = true)] async fn an_outage_refuses_transiently_and_never_extends_the_expiry() { let (lease, mut admissions) = admission(grant("a/**", 30)).await; From 1d4989d47413a1fb7ce6d65b5982b3d430f28753 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 22:21:05 -0700 Subject: [PATCH 10/16] fix(relay): count a lease that ends its session as a failure A revoked, refused, narrowed, or dropped grant aborts the session, but only an expired one counted. Lease::ended now records the end as expired or application. Co-Authored-By: Claude Opus 5.5 --- doc/concept/stats.md | 5 +++-- rs/moq-relay/src/auth.rs | 28 ++++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 590d8d16f1..fa1b349ac3 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -223,8 +223,9 @@ resolved. `sessions_failed` counts admitted sessions that ended abnormally, by kind: a subset of `sessions_ended`, so the live count is unchanged. A clean close by -either side, a drain, or a shutdown is not a failure. A reader treats a missing -object as zeros. +either side, a drain, or a shutdown is not a failure. A grant that expires +mid-session counts as `expired`; one revoked or narrowed out from under it +counts as `application`. A reader treats a missing object as zeros. `requests` counts admitted request/response exchanges under the token's root and tier, including HTTP `/fetch` and `/announced`. They never start or end a session. diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index f0a3a756d8..ed285937d7 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -585,7 +585,19 @@ impl Lease { /// /// Once the relay answers the session's AUTH requests itself, each of /// these ends only that credential's grant; the lease ends when no credential remains. + /// + /// The session's [stats](Self::with_stats) record the end as a failure: + /// `expired` for an expired grant, `application` for any other. pub async fn ended(&mut self) -> lease::Reason { + let reason = self.uncovered().await; + self.stats.fail(match reason { + lease::Reason::Expired => moq_net::stats::Failure::Expired, + _ => moq_net::stats::Failure::Application, + }); + reason + } + + async fn uncovered(&mut self) -> lease::Reason { if let Some(refresh) = &mut self.refresh { return refresh.ended().await; } @@ -1418,6 +1430,22 @@ mod tests { mod outcome_tests { use super::*; + #[tokio::test] + async fn a_revoked_lease_fails_its_session_as_application() { + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session("tenant"); + let (producer, consumer) = lease::Producer::new(Grant::default()); + let mut lease = Lease::new("/tenant", consumer).with_stats(stats.clone()); + producer.revoke(lease::Reason::Refused); + let why = lease.ended().await; + lease.close(why, Bytes::default()); + drop(stats); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_failed.application, 1); + assert_eq!(presence.sessions_failed.expired, 0); + assert_eq!(presence.sessions_ended, 1); + } + #[test] fn typed_expiry_wins_a_racing_transport_close_without_string_classification() { let registry = moq_net::stats::Registry::new(Default::default()); From 9ed7287d06b5f5e0bffc0baae6248f702f433f91 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 22:52:47 -0700 Subject: [PATCH 11/16] fix(relay): count an admitted session refused while preparing its lease A grant that gains an unenforceable cap between admission and SETUP fails Lease::preparing, which returned before the SETUP-reply classification. The admitted session then ended clean. Classify the preparing error the same way. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/auth.rs | 4 +++- rs/moq-relay/src/auth/refresh.rs | 11 ++++++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index ed285937d7..b8584f40c5 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -524,7 +524,9 @@ impl Lease { handle: moq_net::auth::Handle, role: Option, ) -> Result { - self.refresh = Some(refresh::Refresh::new(&mut self, handle, role)?); + let refresh = refresh::Refresh::new(&mut self, handle, role) + .inspect_err(|err| crate::connection::setup_failed(&self.stats, Some(err)))?; + self.refresh = Some(refresh); Ok(self) } diff --git a/rs/moq-relay/src/auth/refresh.rs b/rs/moq-relay/src/auth/refresh.rs index 1a5b53cc69..a092579df0 100644 --- a/rs/moq-relay/src/auth/refresh.rs +++ b/rs/moq-relay/src/auth/refresh.rs @@ -580,7 +580,9 @@ mod tests { #[tokio::test(start_paused = true)] async fn a_cap_added_before_protocol_setup_is_refused() { let (producer, consumer) = lease::Producer::new(grant("a/**", 3000)); - let lease = super::super::Lease::new("/room", consumer); + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(moq_net::stats::Tier::default()).session("room"); + let lease = super::super::Lease::new("/room", consumer).with_stats(stats.clone()); let (client_transport, server) = transports().await; let now = tokio::time::Instant::now().into_std(); let origin = moq_tokio::origin::spawn(); @@ -621,6 +623,13 @@ mod tests { let ((session, driver), ()) = tokio::join!(connecting, refusing); assert!(driver.await.unwrap().to_string().contains("unauthorized")); drop(session); + drop(stats); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_ended, 1); + assert_eq!( + presence.sessions_failed.application, 1, + "an admitted session refused before SETUP is a failure" + ); } #[tokio::test(start_paused = true)] From dcccb6e2a62e47fd2605fa4dc0818ad889f3f8d7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 23:07:23 -0700 Subject: [PATCH 12/16] fix(relay): count any typed revocation that races a clean close Lease::close recorded only an expired revocation, so a refused, invalid, narrowed, or dropped grant that beat a clean peer close ended the session clean. It now records every revocation, and close_session relies on the first recorded kind winning. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/auth.rs | 40 ++++++++++++++++++++++++++++++---------- 1 file changed, 30 insertions(+), 10 deletions(-) diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index b8584f40c5..cc8ba26548 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -544,10 +544,9 @@ impl Lease { Some(_) => error.to_string(), None => "closed".to_string(), }; - // An expired lease already recorded its own failure kind. - if self.close(reason, bytes) != lease::Reason::Expired - && let Some(kind) = failure - { + // A lease the decider ended first records its kind in `close`, and the first kind wins. + self.close(reason, bytes); + if let Some(kind) = failure { stats.fail(kind); } } @@ -592,10 +591,8 @@ impl Lease { /// `expired` for an expired grant, `application` for any other. pub async fn ended(&mut self) -> lease::Reason { let reason = self.uncovered().await; - self.stats.fail(match reason { - lease::Reason::Expired => moq_net::stats::Failure::Expired, - _ => moq_net::stats::Failure::Application, - }); + self.stats + .fail(revoked(&reason).unwrap_or(moq_net::stats::Failure::Application)); reason } @@ -682,13 +679,22 @@ impl Lease { self.consumer.close(reason.clone(), bytes); reason }; - if reason == lease::Reason::Expired { - self.stats.fail(moq_net::stats::Failure::Expired); + if let Some(kind) = revoked(&reason) { + self.stats.fail(kind); } reason } } +/// The failure a lease end records when the grant, not the session, ended it. +fn revoked(reason: &lease::Reason) -> Option { + match reason { + lease::Reason::Expired => Some(moq_net::stats::Failure::Expired), + lease::Reason::Shutdown | lease::Reason::Session(_) => None, + _ => Some(moq_net::stats::Failure::Application), + } +} + /// Patterns as one comma-separated line, for logs. fn join(patterns: &Patterns) -> String { patterns @@ -1448,6 +1454,20 @@ mod outcome_tests { assert_eq!(presence.sessions_ended, 1); } + #[test] + fn a_revocation_wins_a_racing_clean_close() { + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session("tenant"); + let (producer, consumer) = lease::Producer::new(Grant::default()); + let lease = Lease::new("/tenant", consumer).with_stats(stats.clone()); + producer.revoke(lease::Reason::Refused); + lease.close_session(&moq_net::Error::Cancel, Bytes::default()); + drop(stats); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_failed.application, 1); + assert_eq!(presence.sessions_ended, 1); + } + #[test] fn typed_expiry_wins_a_racing_transport_close_without_string_classification() { let registry = moq_net::stats::Registry::new(Default::default()); From e454fb83901215011e883f59a2d33f984102fdc7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 23:17:05 -0700 Subject: [PATCH 13/16] docs(net): say which error a code-less close surfaces as Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/error.rs | 4 ++++ rs/moq-net/src/session.rs | 9 +++++---- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index beabefd745..eb86c552f4 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -572,6 +572,10 @@ impl Error { /// [`StreamError::from_code`]. Reading a stream reset with the session table (or the /// reverse) silently mistranslates, since e.g. 0 is "no error" for a session but an /// internal error for a stream. + /// + /// Without either code, the transport's [`close_kind`](crate::transport::Error::close_kind) + /// decides: an idle timeout is [`Error::IdleTimeout`], a local close is + /// `Error::Session(SessionError::Cancel)`, and anything else is [`Error::Transport`]. pub fn from_transport(err: impl crate::transport::Error) -> Self { if let Some((code, _reason)) = err.session_error() { return SessionError::from_code(code).into(); diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index 4a825756d5..5e1e7e5b4f 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -226,10 +226,11 @@ impl Session { /// /// A close code the peer sent is decoded through the session registry (so an auth /// rejection arrives as `Error::Session(SessionError::Unauthorized)`); every peer code is - /// preserved as [`Error::Session`], and a close carrying no application code surfaces as - /// [`Error::Transport`]. See [`Error::from_transport`]. If the runtime drops - /// the driver instead of running it to completion, this resolves with - /// [`Error::Cancel`]. + /// preserved as [`Error::Session`]. A close carrying no application code surfaces as + /// [`Error::IdleTimeout`] for an idle timeout, `Error::Session(SessionError::Cancel)` + /// for a local close, and [`Error::Transport`] otherwise. See [`Error::from_transport`]. + /// If the runtime drops the driver instead of running it to completion, this resolves + /// with [`Error::Cancel`]. pub async fn closed(&self) -> Error { match self .closed From 6cc4a7d7a13afa7fc0cd4d23b44d3eed50a39fd4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 23:32:26 -0700 Subject: [PATCH 14/16] fix(relay): keep a lease ended by shutdown out of the failures Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/auth.rs | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index cc8ba26548..6d908f4bf5 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -588,11 +588,14 @@ impl Lease { /// these ends only that credential's grant; the lease ends when no credential remains. /// /// The session's [stats](Self::with_stats) record the end as a failure: - /// `expired` for an expired grant, `application` for any other. + /// `expired` for an expired grant, nothing for a shutdown, `application` for any other. pub async fn ended(&mut self) -> lease::Reason { let reason = self.uncovered().await; - self.stats - .fail(revoked(&reason).unwrap_or(moq_net::stats::Failure::Application)); + // Any other reason here, the decider's own included, still ends a live session. + if reason != lease::Reason::Shutdown { + self.stats + .fail(revoked(&reason).unwrap_or(moq_net::stats::Failure::Application)); + } reason } @@ -1454,6 +1457,21 @@ mod outcome_tests { assert_eq!(presence.sessions_ended, 1); } + #[tokio::test] + async fn a_lease_ended_by_shutdown_is_clean() { + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session("tenant"); + let (producer, consumer) = lease::Producer::new(Grant::default()); + let mut lease = Lease::new("/tenant", consumer).with_stats(stats.clone()); + producer.revoke(lease::Reason::Shutdown); + let why = lease.ended().await; + lease.close(why, Bytes::default()); + drop(stats); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_failed, moq_net::stats::Failures::default()); + assert_eq!(presence.sessions_ended, 1); + } + #[test] fn a_revocation_wins_a_racing_clean_close() { let registry = moq_net::stats::Registry::new(Default::default()); From e75a0f109661e780c7442a0667cab83055317d1c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 23:43:26 -0700 Subject: [PATCH 15/16] fix(relay): count a refreshed union that ended unobserved, and any terminal peer link error A refreshed lease's close now polls its union once, so a last credential revoked just before a clean close still fails the session. An established cluster link now counts every terminal error except a refused GOAWAY redirect, which is the peer draining. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/auth.rs | 29 +++++++++++++++++---------- rs/moq-relay/src/auth/refresh.rs | 22 +++++++++++++++++++++ rs/moq-relay/src/cluster.rs | 34 +++++++++++++++++++++++++------- 3 files changed, 68 insertions(+), 17 deletions(-) diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 6d908f4bf5..01b95e7b31 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -668,19 +668,28 @@ impl Lease { /// End the lease with the session's close classification and the totals it /// moved, and learn what ended the session: that, or for a gateway the /// decider's reason if it revoked first. Dropping the consumer reports zero bytes. - pub fn close(self, reason: impl Into, bytes: Bytes) -> lease::Reason { + pub fn close(mut self, reason: impl Into, bytes: Bytes) -> lease::Reason { + use futures::FutureExt; + let mut reason = reason.into(); // A gateway has one credential, so its revocation caused the transfer to - // end. A refreshed session instead reports the result of its whole union: - // its first credential may have ended with a reason a replacement outlived. - let reason = if self.refresh.is_none() { - if let std::task::Poll::Ready(ended) = self.consumer.poll_closed(&moq_net::kio::Waiter::noop()) { - reason = ended; + // end. A refreshed session instead reports the result of its whole union, + // which may have ended before the supervisor saw it: its first credential + // may have ended with a reason a replacement outlived. + let reason = match &mut self.refresh { + None => { + if let std::task::Poll::Ready(ended) = self.consumer.poll_closed(&moq_net::kio::Waiter::noop()) { + reason = ended; + } + self.consumer.close(reason, bytes) + } + Some(refresh) => { + if let Some(ended) = refresh.ended().now_or_never() { + reason = ended; + } + self.consumer.close(reason.clone(), bytes); + reason } - self.consumer.close(reason, bytes) - } else { - self.consumer.close(reason.clone(), bytes); - reason }; if let Some(kind) = revoked(&reason) { self.stats.fail(kind); diff --git a/rs/moq-relay/src/auth/refresh.rs b/rs/moq-relay/src/auth/refresh.rs index a092579df0..025d2900d4 100644 --- a/rs/moq-relay/src/auth/refresh.rs +++ b/rs/moq-relay/src/auth/refresh.rs @@ -676,6 +676,28 @@ mod tests { let _ = supervising.await; } + /// A union that ended before the supervisor saw it still fails a racing clean close. + #[tokio::test(start_paused = true)] + async fn an_unobserved_union_end_fails_a_racing_clean_close() { + let (auth, mut admissions) = super::super::Auth::embedded("test"); + let (producer, consumer) = lease::Producer::new(grant("a/**", 3000)); + let request = auth.request(moq_auth::Transport::Quic, "/room"); + let (lease, ()) = tokio::join!(auth.admit(request), async { + admissions.next().await.unwrap().grant(consumer); + }); + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(moq_net::stats::Tier::default()).session("room"); + let (client, session, lease) = connect(lease.unwrap().with_stats(stats.clone())).await; + + producer.revoke(lease::Reason::Refused); + lease.close_session(&moq_net::Error::Cancel, Default::default()); + drop((session, stats)); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_ended, 1); + assert_eq!(presence.sessions_failed.application, 1); + client.abort(moq_net::Error::Cancel); + } + /// A replacement keeps the session alive past the first credential's expiry, so a /// later clean close is not an expired session. #[tokio::test(start_paused = true)] diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 88727c47bf..11abe4ac78 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -2064,6 +2064,10 @@ fn remote_status( } // The retry window ran out after the link was lost, so the link failed to recover. Err(moq_tokio::Error::Reconnect(_)) if *established => stats.fail(moq_net::stats::Failure::Transport), + // The peer sent GOAWAY and is draining; not following its redirect is no failure. + Err(moq_tokio::Error::RefusedRedirect(_) | moq_tokio::Error::PinnedRedirect) => {} + // Any other terminal error, such as a reconnect the peer refused, ends a live link. + Err(_) if *established => stats.fail(moq_net::stats::Failure::Application), _ => {} } status @@ -2240,7 +2244,7 @@ mod tests { #[test] fn remote_logical_outcome_ignores_retries_and_counts_one_terminal_close() { - use moq_net::stats::{Failure, Registry}; + use moq_net::stats::Registry; let registry = Registry::new(Default::default()); let stats = registry.tier(Tier::default()).session(""); let mut established = false; @@ -2257,13 +2261,9 @@ mod tests { ] { remote_status(&stats, &mut established, Ok(status)).unwrap(); } - let _ = remote_status(&stats, &mut established, Err(moq_tokio::Error::ConnectFailed)); assert_eq!( - registry.snapshot().sessions()[0] - .1 - .sessions_failed - .count(Failure::Transport), - 0 + registry.snapshot().sessions()[0].1.sessions_failed, + moq_net::stats::Failures::default() ); let _ = remote_status( &stats, @@ -2307,6 +2307,26 @@ mod tests { assert_eq!(exhausted(true), 1); } + #[test] + fn remote_terminal_errors_fail_an_established_link_except_a_refused_redirect() { + use moq_net::stats::{Failures, Registry}; + let ended = |err: moq_tokio::Error| { + let registry = Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session(""); + let mut connected = false; + remote_status(&stats, &mut connected, Ok(moq_tokio::Status::Connected)).unwrap(); + let _ = remote_status(&stats, &mut connected, Err(err)); + drop(stats); + registry.snapshot().sessions()[0].1.sessions_failed + }; + assert_eq!(ended(moq_tokio::Error::ConnectFailed).application, 1); + assert_eq!(ended(moq_tokio::Error::PinnedRedirect), Failures::default()); + assert_eq!( + ended(moq_tokio::Error::RefusedRedirect("policy".into())), + Failures::default() + ); + } + /// `?cost=` is read and stripped (it rides SETUP, not the URL), other /// query params survive, and a garbage value is an error rather than a /// silent default. From 1d1589ad1df13150157d472c0481be1459efae41 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 23:53:27 -0700 Subject: [PATCH 16/16] fix(relay): classify an unobserved union end like Lease::ended The refresh poll in Lease::close took the union's reason but classified it as the session's own close, so a cap the relay could not enforce ended clean. Both paths now share one classifier. Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/auth.rs | 18 ++++++++++++++---- rs/moq-relay/src/auth/refresh.rs | 24 ++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 01b95e7b31..1e9878a799 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -591,10 +591,8 @@ impl Lease { /// `expired` for an expired grant, nothing for a shutdown, `application` for any other. pub async fn ended(&mut self) -> lease::Reason { let reason = self.uncovered().await; - // Any other reason here, the decider's own included, still ends a live session. - if reason != lease::Reason::Shutdown { - self.stats - .fail(revoked(&reason).unwrap_or(moq_net::stats::Failure::Application)); + if let Some(kind) = uncovered(&reason) { + self.stats.fail(kind); } reason } @@ -685,6 +683,9 @@ impl Lease { } Some(refresh) => { if let Some(ended) = refresh.ended().now_or_never() { + if let Some(kind) = uncovered(&ended) { + self.stats.fail(kind); + } reason = ended; } self.consumer.close(reason.clone(), bytes); @@ -698,6 +699,15 @@ impl Lease { } } +/// The failure a lease records when it stops covering a live session. Any reason but a +/// shutdown, the relay's or the decider's own words included, ends that session. +fn uncovered(reason: &lease::Reason) -> Option { + match reason { + lease::Reason::Shutdown => None, + _ => Some(revoked(reason).unwrap_or(moq_net::stats::Failure::Application)), + } +} + /// The failure a lease end records when the grant, not the session, ended it. fn revoked(reason: &lease::Reason) -> Option { match reason { diff --git a/rs/moq-relay/src/auth/refresh.rs b/rs/moq-relay/src/auth/refresh.rs index 025d2900d4..3f73ac78d3 100644 --- a/rs/moq-relay/src/auth/refresh.rs +++ b/rs/moq-relay/src/auth/refresh.rs @@ -698,6 +698,30 @@ mod tests { client.abort(moq_net::Error::Cancel); } + /// The relay's own reason for ending an unobserved union fails a racing clean close too. + #[tokio::test(start_paused = true)] + async fn an_unobserved_cap_fails_a_racing_clean_close() { + let (auth, mut admissions) = super::super::Auth::embedded("test"); + let (producer, consumer) = lease::Producer::new(grant("a/**", 3000)); + let request = auth.request(moq_auth::Transport::Quic, "/room"); + let (lease, ()) = tokio::join!(auth.admit(request), async { + admissions.next().await.unwrap().grant(consumer); + }); + let registry = moq_net::stats::Registry::new(Default::default()); + let stats = registry.tier(moq_net::stats::Tier::default()).session("room"); + let (client, session, lease) = connect(lease.unwrap().with_stats(stats.clone())).await; + + let mut capped = grant("a/**", 3000); + capped.publish.rate = Some(moq_auth::Rate::from_bps(8000).unwrap()); + producer.update(capped); + lease.close_session(&moq_net::Error::Cancel, Default::default()); + drop((session, stats)); + let presence = registry.snapshot().sessions()[0].1; + assert_eq!(presence.sessions_ended, 1); + assert_eq!(presence.sessions_failed.application, 1); + client.abort(moq_net::Error::Cancel); + } + /// A replacement keeps the session alive past the first credential's expiry, so a /// later clean close is not an expired session. #[tokio::test(start_paused = true)]