diff --git a/cpp/obs/src/moq-dock.cpp b/cpp/obs/src/moq-dock.cpp index d527bd78a1..ebf44e9d7c 100644 --- a/cpp/obs/src/moq-dock.cpp +++ b/cpp/obs/src/moq-dock.cpp @@ -174,6 +174,9 @@ QString ExplainFailure(int code, const std::string &reason) return QStringLiteral("Connect failed · the relay did not accept the session"); case MoQError::Kind::Offline: return {}; + case MoQError::Kind::TooManyRequests: + return QStringLiteral( + "Relay request limit reached · reduce tracks or ask the operator to raise the limit"); case MoQError::Kind::Other: return QString::fromStdString(reason); } diff --git a/cpp/obs/src/moq-error.h b/cpp/obs/src/moq-error.h index fb2ae1973e..4782c7ae2c 100644 --- a/cpp/obs/src/moq-error.h +++ b/cpp/obs/src/moq-error.h @@ -5,13 +5,15 @@ namespace MoQError { -enum class Kind { Unauthorized, Forbidden, Certificate, Timeout, Network, Connect, Offline, Other }; +enum class Kind { Unauthorized, Forbidden, Certificate, Timeout, Network, Connect, Offline, Other, TooManyRequests }; // What the output recorded a failure as, beside its text. None means no failure. -enum Code : int { None = 0, Unauthorized, Forbidden, Connect, Offline, Other }; +enum Code : int { None = 0, Unauthorized, Forbidden, Connect, Offline, Other, TooManyRequests }; inline Kind Classify(int code, std::string reason) { + if (code == Code::TooManyRequests) + return Kind::TooManyRequests; for (char &c : reason) { if (c >= 'A' && c <= 'Z') c = static_cast(c - 'A' + 'a'); diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 9ae0466914..5651a8c4f4 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -65,6 +65,8 @@ MoQError::Code FailureCode(const moq::Error &error) if (std::holds_alternative(variant)) return MoQError::Code::Connect; if (const auto *protocol = std::get_if(&variant)) { + if (protocol->details.kind == moq::ProtocolKind::kTooManyRequests) + return MoQError::Code::TooManyRequests; if (protocol->details.kind == moq::ProtocolKind::kUnauthorized) return MoQError::Code::Unauthorized; } diff --git a/cpp/obs/test/moq-error-test.cpp b/cpp/obs/test/moq-error-test.cpp index 082ef4a82d..40b7cf85a7 100644 --- a/cpp/obs/test/moq-error-test.cpp +++ b/cpp/obs/test/moq-error-test.cpp @@ -22,7 +22,8 @@ int main() Kind::Certificate || Classify(MoQError::Unauthorized, "reconnect timed out after 10s") != Kind::Unauthorized || Classify(MoQError::Forbidden, "forbidden") != Kind::Forbidden || - Classify(MoQError::Offline, "offline") != Kind::Offline) { + Classify(MoQError::Offline, "offline") != Kind::Offline || + Classify(MoQError::TooManyRequests, "reconnect timed out") != Kind::TooManyRequests) { std::fprintf(stderr, "FAIL: connection failure classification\n"); return 1; } diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 6aa3ba5a2b..e17f587691 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -4249,6 +4249,7 @@ enum MoqProtocolKind { timestampMismatch, app, unknown, + tooManyRequests, } class FfiConverterMoqProtocolKind { @@ -4299,6 +4300,8 @@ class FfiConverterMoqProtocolKind { return LiftRetVal(MoqProtocolKind.app, 4); case 22: return LiftRetVal(MoqProtocolKind.unknown, 4); + case 23: + return LiftRetVal(MoqProtocolKind.tooManyRequests, 4); default: throw UniffiInternalError( UniffiInternalError.unexpectedEnumCase, diff --git a/doc/bin/obs.md b/doc/bin/obs.md index 237d0e7b30..0938af7048 100644 --- a/doc/bin/obs.md +++ b/doc/bin/obs.md @@ -68,3 +68,8 @@ stream with the reason in the log rather than silently using a default. The plugin is C++ over the [generated C++ bindings](/lib/cpp/) and ships with every moq-cpp release. + +If the relay closes the session for too many requests, reduce the broadcasts +or subscribed tracks, or have the operator adjust its +[session limits](/bin/relay/config#session-and-peer). Reconnecting with the same +workload does not resolve a request cap. diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index dbd3aa7025..ee7a24c595 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -79,6 +79,31 @@ how much unread data one peer can make the relay buffer. Keep `stream_receive_window` well under it so one slow group cannot starve the connection. +## \[session] and \[peer] + +Request caps apply separately to each session on every transport. The relay +uses `[peer]` for an admitted grant marked `peer` or `upstream`, including +verified LAN membership, and for links it dials through the cluster. A client +certificate, route hop, or cluster-looking URL alone does not select peer caps. + +```toml +[session] +announces = 1000 # Simultaneous broadcasts or namespaces. +subscriptions = 1000 # Simultaneously held subscriptions. + +[peer] +announces = 100000 +subscriptions = 10000 +``` + +These are the defaults. For example, `--peer-subscriptions=50000` raises the +subscription cap on cluster links. Size peer caps for the combined traffic +they carry, and use session peak counters to check headroom. Exceeding a cap +closes the session with `TOO_MANY_REQUESTS`; reconnecting with the same load +will exceed it again. Zero admits no entries of that kind, not unlimited +entries. On moq-transport drafts 14 to 16 these caps also size the request +window, so very small values can prevent other requests from progressing. + ## \[runtime] By default one work-stealing runtime serves every connection off one UDP diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 302609e51f..398f536500 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -183,10 +183,11 @@ on the new. `announces_peak` and `subscriptions_peak` are the most broadcasts announced to the relay, and subscriptions held on it, by any one session. Compare them with -the relay's per-session limits: a session that goes past one is closed with -`TOO_MANY_REQUESTS`. On `sessions.json` a peak never goes down while the root -has a session; on the totals and presence tracks it never goes down while the -track lives. Summing nodes or roots takes the largest. +the relay's [client or peer session limits](/bin/relay/config#session-and-peer): +a session that goes past one is closed with `TOO_MANY_REQUESTS`. On +`sessions.json` a peak never goes down while the root has a session; on the +totals and presence tracks it never goes down while the track lives. Summing +nodes or roots takes the largest. ### Counters diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 6831149da4..81ce7c8ce3 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -57,7 +57,7 @@ here, from media publish and consume to raw pixels and PCM with the codec inside. The header is the reference: each function carries a doc comment. - **Handles and callbacks.** Every object is an integer handle, and every async result arrives on a `moq_status_callback` with your `void *user_data`. A status `> 0` is a live result, `0` a clean close, and `< 0` an error. The last two are terminal: moq-c never touches `user_data` again, so free it there. `*_close` and `*_cancel` only request shutdown; the terminal callback still fires. -- **Errors.** Negative return codes are named by `MOQ_ERROR_*`, and `moq_error()` gives the reason on the calling thread. Auth rejections have their own codes so you don't retry them. For the peer's protocol code use `moq_error_protocol()`; do not parse `moq_error()`. +- **Errors.** Negative return codes are named by `MOQ_ERROR_*`, and `moq_error()` gives the reason on the calling thread. Auth rejections have their own codes so you don't retry them. For the peer's protocol code use `moq_error_protocol()`; do not parse `moq_error()`. A request-cap closure has kind `MOQ_PROTOCOL_KIND_TOO_MANY_REQUESTS`; reduce the workload or adjust the relay's [session limits](/bin/relay/config#session-and-peer) before reconnecting. - **Threading.** Call any function from any thread. Raw publish calls block until the codec takes the frame, which paces a publisher. - **Zeroed config is the default.** A zeroed `moq_client_config` means the default for every field, so new fields don't break callers. A field whose default isn't zero has a `has_*` flag: `backoff_timeout_us = 0` means "retry forever" only with `has_backoff_timeout = true`. - **Audio needs cuts.** Video groups at its keyframes, but audio forms a group only where you call `moq_publish_media_cut`: after every frame, or at a segment cadence to align with video. diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index 043767dd08..bcbe1accc0 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -202,6 +202,12 @@ gives the message Rust's `Display` does, for logs; branch on the case, not the text. A Rust panic or a misused future (`get()` twice) aborts with a message on stderr instead of throwing. +A session closed for exceeding the relay's request caps reports the protocol +kind `ProtocolKind::kTooManyRequests`. Reduce +the simultaneous announcements or subscriptions, or ask the relay operator +to adjust the [session limits](/bin/relay/config#session-and-peer), before +reconnecting with the same workload. + Check a result before reading it. With exceptions off below C++23, reading the value of an error is undefined, through `*result` and `value()` alike, since the bundled `tl::expected` has nothing to throw. diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 359acf0670..6e50c44da3 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -87,6 +87,12 @@ host still poll in place; independent native wake callbacks can run concurrently The generated bindings deliver native wakeups to the Dart isolate. Keep UI work on its owning isolate. +A session closed for exceeding the relay's request caps reports the protocol +kind `ProtocolKind.tooManyRequests`. Reduce +the simultaneous announcements or subscriptions, or ask the relay operator +to adjust the [session limits](/bin/relay/config#session-and-peer), before +reconnecting with the same workload. + ## Reference - API reference: [pub.dev/documentation/moq](https://pub.dev/documentation/moq/latest/) diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 706cf5c58e..a9540a3512 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -116,6 +116,12 @@ host still poll in place; independent native wake callbacks can run concurrently The generated bridge synchronizes wakeups before resuming the awaiting goroutine. Application goroutines remain responsible for serializing their own shared state. +A session closed for exceeding the relay's request caps reports the protocol +kind `ProtocolKindTooManyRequests`. Reduce +the simultaneous announcements or subscriptions, or ask the relay operator +to adjust the [session limits](/bin/relay/config#session-and-peer), before +reconnecting with the same workload. + ## Reference - API reference: [pkg.go.dev/moq.dev/moq](https://pkg.go.dev/moq.dev/moq) diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 50609f3ed3..6edf5c3619 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -91,6 +91,12 @@ host still poll in place; independent native wake callbacks can run concurrently The generated bindings resume Kotlin coroutines through their dispatcher. Use the UI dispatcher when updating UI state. +A session closed for exceeding the relay's request caps reports the protocol +kind `ProtocolKind.TOO_MANY_REQUESTS`. Reduce +the simultaneous announcements or subscriptions, or ask the relay operator +to adjust the [session limits](/bin/relay/config#session-and-peer), before +reconnecting with the same workload. + ## Reference - API reference: [javadoc.io/doc/dev.moq/moq](https://javadoc.io/doc/dev.moq/moq) diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index a6f1338dd2..aacdd92c01 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -106,6 +106,12 @@ host still poll in place; independent native wake callbacks can run concurrently The generated bindings wake Python futures through the asyncio event loop. The package exit hook joins native workers before interpreter finalization. +A session closed for exceeding the relay's request caps reports the protocol +kind `ProtocolKind.TOO_MANY_REQUESTS`. Reduce +the simultaneous announcements or subscriptions, or ask the relay operator +to adjust the [session limits](/bin/relay/config#session-and-peer), before +reconnecting with the same workload. + ## Reference - API reference: [moq-rs.readthedocs.io](https://moq-rs.readthedocs.io) diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 84de1787c9..2a8fffabd3 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -92,6 +92,12 @@ host still poll in place; independent native wake callbacks can run concurrently The generated bindings resume Swift continuations. Use the appropriate actor when updating UI state. +A session closed for exceeding the relay's request caps reports the protocol +kind `ProtocolKind.tooManyRequests`. Reduce +the simultaneous announcements or subscriptions, or ask the relay operator +to adjust the [session limits](/bin/relay/config#session-and-peer), before +reconnecting with the same workload. + ## Reference - API reference: [Swift Package Index (DocC)](https://swiftpackageindex.com/moq-dev/moq-swift/documentation/moq) diff --git a/go/wrapper/errors_test.go b/go/wrapper/errors_test.go index 627f6bb694..211f3516d4 100644 --- a/go/wrapper/errors_test.go +++ b/go/wrapper/errors_test.go @@ -107,3 +107,13 @@ func TestErrorSentinels(t *testing.T) { }) } } + +func TestSessionLimitProtocolKind(t *testing.T) { + err := ffi.NewMoqErrorProtocol(ffi.MoqProtocolError{ + Scope: moq.ErrorScopeSession, Code: 7, Kind: moq.ProtocolKindTooManyRequests, Message: "request limit", + }) + details, ok := moq.ProtocolError(err) + if !ok || details.Kind != moq.ProtocolKindTooManyRequests || moq.IsAuthError(err) { + t.Fatalf("session limit lost its kind: %+v, %v", details, err) + } +} diff --git a/go/wrapper/types.go b/go/wrapper/types.go index f03cc84637..53394b7b57 100644 --- a/go/wrapper/types.go +++ b/go/wrapper/types.go @@ -215,6 +215,8 @@ const ( ProtocolKindApp = ffi.MoqProtocolKindApp // ProtocolKindUnknown identifies unknown. ProtocolKindUnknown = ffi.MoqProtocolKindUnknown + // ProtocolKindTooManyRequests identifies an exceeded session request cap. + ProtocolKindTooManyRequests = ffi.MoqProtocolKindTooManyRequests ) // AudioSampleFormat values: the raw PCM sample layout fed to or returned from the diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index d082dce0ba..0fc98a2787 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -120,6 +120,7 @@ async def reject(): def test_protocol_error_helper_covers_known_app_and_unknown(): cases = [ (moq.ErrorScope.SESSION, 0x2, moq.ProtocolKind.UNAUTHORIZED, True), + (moq.ErrorScope.SESSION, 0x7, moq.ProtocolKind.TOO_MANY_REQUESTS, False), (moq.ErrorScope.SESSION, 64 + 404, moq.ProtocolKind.APP, False), (moq.ErrorScope.SESSION, 0x1F, moq.ProtocolKind.UNKNOWN, False), (moq.ErrorScope.STREAM, 64 + 7, moq.ProtocolKind.APP, False), diff --git a/quest/m1/README.md b/quest/m1/README.md index 83f44048ba..6447fa587a 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -103,7 +103,6 @@ demand (#5214), each ranked beside the work it shares code with. - [IETF fill timeout](/quest/m1/ietf-fill-timeout.md) - native FILL_TIMEOUT bounds upstream waiting on drafts 18-22, alongside local content-age enforcement - [IETF group ingest refuses loudly](/quest/m1/ietf-ingest-refusal.md) - a subgroup other than 0 or a datagram past object 0 aborts the track, and a peer's malformed group is refused rather than dropped at debug - [Draft 17-18 withdraw by reset](/quest/m1/ietf-withdraw-reset.md) - withdrawing a PUBLISH_NAMESPACE on drafts 17 and 18 resets the request instead of FIN-ing it -- [Relay session limits](/quest/m1/relay-session-limits.md) - moq-relay sets per-session request limits, tighter for clients than peers, and the bindings name a refused request - [Catalog track identity](/quest/m1/catalog-tracks.md) - a track's codec and description never change for its name; resolution changes in band below ceilings fixed at creation, and anything else mints a new rendition or epoch - [Archive](/quest/m1/archive/README.md) - record selected tracks to any object_store and replay them over FETCH or derived HLS; the catalog entry and format may break in place, since no archives exist - [In-band auth](/quest/m1/auth/README.md) - a session tells its peer what it may publish and subscribe to, unions tokens presented in band, and fails loud on an out-of-scope publish diff --git a/quest/m1/quic/peer-limits.md b/quest/m1/quic/peer-limits.md index f26f7dfff5..e246a8dbeb 100644 --- a/quest/m1/quic/peer-limits.md +++ b/quest/m1/quic/peer-limits.md @@ -18,10 +18,11 @@ setters provide this seam; the transport integration waits for [the fork switch](/quest/m1/quic/fork/README.md), so moq-tokio and moq-uring run on `moq-quic` instead of `moq-noq`. -Decided 2026-10-08: [relay session limits](/quest/m1/relay-session-limits.md) -lands first and introduces the relay's peer config surface (classifying a -session as a cluster peer, and a peer table); this quest extends that -surface with the QUIC values rather than adding its own. +Decided 2026-10-08: the relay's peer config surface already exists. A +session is a cluster peer when its admitted grant is marked `peer`, and the +`[peer]` table (beside `[session]` for ordinary clients) holds its request +caps; this quest extends that table with the QUIC values rather than adding +its own. - moq-net's transport trait gains `set_limits(Limits)` on the session, `Limits` carrying the three values, implemented by the moq-tokio and @@ -47,6 +48,5 @@ viewer session does not; the io_uring path applies the same values; a ## Required -- [Relay session limits](/quest/m1/relay-session-limits.md) - introduces the peer classification and config table this extends - [Hard fork](/quest/m1/quic/fork/README.md) - moq-tokio and moq-uring run on `moq-quic`, whose setters this calls diff --git a/quest/m1/relay-session-limits.md b/quest/m1/relay-session-limits.md deleted file mode 100644 index b865d3e2ea..0000000000 --- a/quest/m1/relay-session-limits.md +++ /dev/null @@ -1,37 +0,0 @@ -# [S] The relay configures per-session request limits - -## Goal - -`moq-relay` sets moq-net's `session::Limits` from its config, with a tighter -default for client sessions than for cluster peers, and documents it in -`doc/bin/relay/`. The bindings report a refused request as its own error -kind instead of mapping `TooManyRequests` to Unknown. - -## Plan - -Request caps (#4820) set generous defaults in moq-net (100,000 announces and -10,000 subscriptions per session), sized for relay meshes, so viewer sessions -get loose limits. The relay knows -which sessions are peers, so it picks the limit per session. Propose the flag -and TOML names in the PR. - -Decided 2026-10-08: this quest lands before -[peer limits](/quest/m1/quic/peer-limits.md) and introduces the peer config -surface both use: how the relay classifies a session as a cluster peer, and a -peer table holding the request limits. Peer limits adds its QUIC values to -that table. - -Past a cap the session closes with TOO_MANY_REQUESTS, so until cluster peers -get higher caps (or none), a link carrying more than 10,000 subscriptions -closes and flaps on reconnect. This does not gate a release (decided -2026-10-08). - -Add the error kind to `rs/moq-ffi` and every wrapper per the Cross-Package -Sync table. - -Public API: a relay config field and flag, and a new error kind in moq-ffi -and the bindings. Wire: none. - -## Related - -- [Peer limits](/quest/m1/quic/peer-limits.md) - extends the peer table with QUIC stream and data limits diff --git a/quest/m1/test-flakes-2/interop-close-code.md b/quest/m1/test-flakes-2/interop-close-code.md index 759e2d85fd..7e9591ae8f 100644 --- a/quest/m1/test-flakes-2/interop-close-code.md +++ b/quest/m1/test-flakes-2/interop-close-code.md @@ -38,4 +38,15 @@ observation had the same cause. The server HTTP/3 control-stream lifetime is still another candidate. Preserve this quest until a failing browser trace paired with relay logs establishes that cause and its regression. +Seen again 2026-10-11 on #5310's Interop run +[38121507310](https://github.com/moq-dev/moq/actions/runs/38121507310), whose +`interop-failure` artifact keeps the run directory and browser traces. The +relay log pins the refusal at admission, before the relay answered SETUP: +connection 64 closed with "the auth server refused the session" +(07:50:07.583), through `Connection::run`'s reject path, and the harness +reported "Connection lost" by 07:50:07.79. Pair this relay log with the +artifact's browser trace to tell whether the close capsule raced the +server's HTTP/3 control stream or connection teardown. #5310's next push +passed the same check. + Public API: none expected. Wire: none expected. diff --git a/rs/moq-c/src/error.rs b/rs/moq-c/src/error.rs index 165a2dde75..5eef58e683 100644 --- a/rs/moq-c/src/error.rs +++ b/rs/moq-c/src/error.rs @@ -104,6 +104,8 @@ pub enum moq_protocol_kind { MOQ_PROTOCOL_KIND_APP = 23, /// Unknown. MOQ_PROTOCOL_KIND_UNKNOWN = 24, + /// The session exceeded its announcement or subscription cap. + MOQ_PROTOCOL_KIND_TOO_MANY_REQUESTS = 25, } /// A protocol failure a peer sent: scope, verbatim wire code, and recognized kind. @@ -377,6 +379,7 @@ fn session_kind(err: &moq_net::SessionError) -> moq_protocol_kind { moq_net::SessionError::GoawayTimeout => MOQ_PROTOCOL_KIND_GOAWAY_TIMEOUT, moq_net::SessionError::Timeout => MOQ_PROTOCOL_KIND_TIMEOUT, moq_net::SessionError::Version => MOQ_PROTOCOL_KIND_VERSION, + moq_net::SessionError::TooManyRequests => MOQ_PROTOCOL_KIND_TOO_MANY_REQUESTS, moq_net::SessionError::App(_) => MOQ_PROTOCOL_KIND_APP, moq_net::SessionError::Unknown(_) => MOQ_PROTOCOL_KIND_UNKNOWN, _ => MOQ_PROTOCOL_KIND_UNKNOWN, @@ -455,3 +458,17 @@ impl ffi::ReturnCode for Error { } } } + +#[cfg(test)] +mod tests { + #[test] + fn session_limit_has_a_known_kind() { + let error = super::from_session(&moq_net::SessionError::TooManyRequests); + assert_eq!(error.scope, super::moq_error_scope::MOQ_ERROR_SCOPE_SESSION as u32); + assert_eq!(error.code, moq_net::SessionError::TooManyRequests.to_code()); + assert_eq!( + error.kind, + super::moq_protocol_kind::MOQ_PROTOCOL_KIND_TOO_MANY_REQUESTS as u32 + ); + } +} diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index d73304027b..cfad358068 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -65,6 +65,8 @@ pub enum MoqProtocolKind { App, /// A code this version does not recognize; [`MoqProtocolError::code`] is the value. Unknown, + /// The session exceeded its announcement or subscription cap. + TooManyRequests, } /// A protocol failure a peer sent (or this side will send): scope, verbatim code, kind, @@ -121,6 +123,7 @@ fn session_kind(err: &moq_net::SessionError) -> MoqProtocolKind { moq_net::SessionError::GoawayTimeout => MoqProtocolKind::GoawayTimeout, moq_net::SessionError::Timeout => MoqProtocolKind::Timeout, moq_net::SessionError::Version => MoqProtocolKind::Version, + moq_net::SessionError::TooManyRequests => MoqProtocolKind::TooManyRequests, moq_net::SessionError::App(_) => MoqProtocolKind::App, moq_net::SessionError::Unknown(_) => MoqProtocolKind::Unknown, _ => MoqProtocolKind::Unknown, @@ -420,6 +423,14 @@ from_message! { mod tests { use super::*; + #[test] + fn session_limit_has_a_known_kind() { + let error = super::MoqProtocolError::from_session(&moq_net::SessionError::TooManyRequests); + assert_eq!(error.scope, super::MoqErrorScope::Session); + assert_eq!(error.code, moq_net::SessionError::TooManyRequests.to_code()); + assert_eq!(error.kind, super::MoqProtocolKind::TooManyRequests); + } + #[test] fn protocol_codes_survive_container_errors() { let error = moq_mux::Error::Cmaf(moq_mux::container::fmp4::Error::Moq( diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 42f680efc6..86d2c1efc1 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -890,6 +890,12 @@ where self } + /// Set request caps after authorizing the peer and before answering SETUP. + pub fn with_limits(mut self, limits: crate::session::Limits) -> Self { + self.inner_mut().server.limits = limits; + self + } + fn inner_mut(&mut self) -> &mut RequestInner { self.inner.as_mut().expect("request already responded") } diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 8b056a3402..0ba8df8fb0 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -813,6 +813,10 @@ struct LanAuth { #[derive(Default)] #[non_exhaustive] pub struct Options { + /// Request caps for ordinary client sessions. + pub session: crate::limits::Session, + /// Request caps for authenticated peers and outbound cluster links. + pub peer: crate::limits::Peer, /// Cluster identity, peers, and discovery. pub config: Config, @@ -906,6 +910,8 @@ struct Work { /// client can serve local sessions but cannot dial remote peers. #[derive(Clone)] pub struct Cluster { + session_limits: moq_net::session::Limits, + peer_limits: moq_net::session::Limits, config: Config, client: Option, /// Dial template for LAN peers that pin a fingerprint; static and API @@ -967,6 +973,15 @@ pub struct Admitted { } impl Cluster { + /// The request caps for an admitted session: only a `peer` grant gets the peer policy. + pub(crate) fn limits(&self, lease: &auth::Lease) -> moq_net::session::Limits { + if lease.token().peer { + self.peer_limits + } else { + self.session_limits + } + } + /// The origin ID used by this relay on the wire. pub fn id(&self) -> u64 { self.origin.hop().id() @@ -987,7 +1002,12 @@ impl Cluster { /// Errors if `config.id` is set but invalid: it must be non-zero and below /// 2^62 (the wire varint limit). An unset id picks a fresh random origin. pub fn new(options: Options) -> anyhow::Result { - let Options { config, cache } = options; + let Options { + config, + cache, + session, + peer, + } = options; let id = match config.id { Some(0) => anyhow::bail!("--cluster-id must be non-zero"), Some(id) if id >= 1 << 62 => { @@ -1011,6 +1031,8 @@ impl Cluster { let nodes = crate::nodes::Nodes::default(); tracing::info!(hop_id = %origin.hop(), configured = config.id.is_some(), "cluster initialized"); Ok(Cluster { + session_limits: session.resolve(), + peer_limits: peer.resolve(), config, client: None, connect: None, @@ -1720,7 +1742,8 @@ impl Cluster { 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(self.stats.tier(self.cluster_tier()).session("")) + .with_limits(self.peer_limits); if let Some(cost) = target.cost { client = client.with_cost(cost); } @@ -1959,6 +1982,26 @@ mod tests { use super::*; use crate::Config as RelayConfig; + #[tokio::test] + async fn request_limits_follow_the_admitted_grant() { + let mut options = Options::default(); + options.session.announces = 2; + options.session.subscriptions = 3; + options.peer.announces = 20; + options.peer.subscriptions = 30; + let cluster = Cluster::new(options).unwrap(); + for (peer, upstream, expected) in [(false, false, (2, 3)), (true, false, (20, 30)), (true, true, (20, 30))] { + let mut grant = moq_auth::Grant::new(Default::default(), Default::default()); + grant.peer = peer; + grant.upstream = upstream; + let (_producer, consumer) = moq_auth::lease::Producer::new(grant); + // A cluster-looking path alone never selects the peer policy. + let lease = auth::Lease::new("/.cluster/untrusted", consumer); + let limits = cluster.limits(&lease); + assert_eq!((limits.announces, limits.subscriptions), expected); + } + } + /// A millisecond track: the frames written here carry timestamps. fn timed() -> moq_net::track::Info { moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI) diff --git a/rs/moq-relay/src/config.rs b/rs/moq-relay/src/config.rs index 71f14cc4c9..e2087e66f7 100644 --- a/rs/moq-relay/src/config.rs +++ b/rs/moq-relay/src/config.rs @@ -20,6 +20,14 @@ use crate::{auth, cache, cluster, internal, stats, web}; #[serde(deny_unknown_fields, default)] #[non_exhaustive] pub struct Config { + /// Request caps for ordinary client sessions. + #[usage(flatten)] + pub session: crate::limits::Session, + + /// Request caps for authenticated peers and outgoing cluster links. + #[usage(flatten)] + pub peer: crate::limits::Peer, + /// The QUIC/TLS configuration for the server. #[usage(flatten)] #[serde(default)] @@ -128,6 +136,8 @@ pub struct Config { impl Default for Config { fn default() -> Self { Self { + session: Default::default(), + peer: Default::default(), listen: Default::default(), server: None, connect: Default::default(), @@ -360,6 +370,39 @@ mod tests { use super::*; use crate::test_env::EnvGuard; + #[test] + fn session_limits_merge_without_crossing_peer_policy() { + let _env = EnvGuard::clear(&[ + "MOQ_SESSION_ANNOUNCES", + "MOQ_SESSION_SUBSCRIPTIONS", + "MOQ_PEER_ANNOUNCES", + "MOQ_PEER_SUBSCRIPTIONS", + ]); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("limits.toml"); + std::fs::write( + &path, + "[session]\nannounces = 7\nsubscriptions = 8\n[peer]\nannounces = 90\nsubscriptions = 91\n", + ) + .unwrap(); + let config = Config::parse_and_merge([std::ffi::OsString::from("moq-relay"), path.clone().into()]).unwrap(); + assert_eq!(config.session.resolve().announces, 7); + assert_eq!(config.session.resolve().subscriptions, 8); + assert_eq!(config.peer.resolve().announces, 90); + assert_eq!(config.peer.resolve().subscriptions, 91); + let config = Config::parse_and_merge([ + std::ffi::OsString::from("moq-relay"), + path.into(), + "--session-announces=0".into(), + "--peer-subscriptions=1000".into(), + ]) + .unwrap(); + assert_eq!(config.session.resolve().announces, 0); + assert_eq!(config.session.resolve().subscriptions, 8); + assert_eq!(config.peer.resolve().announces, 90); + assert_eq!(config.peer.resolve().subscriptions, 1000); + } + #[cfg(not(feature = "iroh"))] #[test] fn iroh_environment_requires_the_feature() { diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index 36b748ba04..6019fefa1e 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -113,7 +113,7 @@ impl Connection { // // moq-net defaults the unset side to a fresh no-op origin, which is fine for a // publish-only or subscribe-only session. - let mut request = self.request.with_stats(stats); + let mut request = self.request.with_stats(stats).with_limits(self.cluster.limits(&lease)); if let Some(subscriber) = subscriber { request = request.with_publisher(subscriber); } diff --git a/rs/moq-relay/src/lib.rs b/rs/moq-relay/src/lib.rs index 1d89001a98..ac19b67c99 100644 --- a/rs/moq-relay/src/lib.rs +++ b/rs/moq-relay/src/lib.rs @@ -19,6 +19,7 @@ mod duration; mod fetch; mod http_client; pub mod internal; +pub mod limits; mod listener; mod nodes; mod refusals; diff --git a/rs/moq-relay/src/limits.rs b/rs/moq-relay/src/limits.rs new file mode 100644 index 0000000000..af5fa4b149 --- /dev/null +++ b/rs/moq-relay/src/limits.rs @@ -0,0 +1,86 @@ +//! Request caps selected after the relay authorizes a session. + +use serde::{Deserialize, Serialize}; + +/// Request caps for ordinary client sessions. +#[derive(usage::Args, Clone, Debug, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +#[non_exhaustive] +pub struct Session { + /// Maximum simultaneously announced broadcasts or namespaces. + #[usage( + long = "session-announces", + env = "MOQ_SESSION_ANNOUNCES", + default = "1000", + setting = "session.announces" + )] + pub announces: usize, + /// Maximum simultaneously held subscriptions. + #[usage( + long = "session-subscriptions", + env = "MOQ_SESSION_SUBSCRIPTIONS", + default = "1000", + setting = "session.subscriptions" + )] + pub subscriptions: usize, +} + +impl Default for Session { + fn default() -> Self { + Self { + announces: 1000, + subscriptions: 1000, + } + } +} + +impl Session { + pub(crate) fn resolve(&self) -> moq_net::session::Limits { + resolve(self.announces, self.subscriptions) + } +} + +/// Request caps for authenticated cluster peer sessions. +#[derive(usage::Args, Clone, Debug, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +#[non_exhaustive] +pub struct Peer { + /// Maximum simultaneously announced broadcasts or namespaces. + #[usage( + long = "peer-announces", + env = "MOQ_PEER_ANNOUNCES", + default = "100000", + setting = "peer.announces" + )] + pub announces: usize, + /// Maximum simultaneously held subscriptions. + #[usage( + long = "peer-subscriptions", + env = "MOQ_PEER_SUBSCRIPTIONS", + default = "10000", + setting = "peer.subscriptions" + )] + pub subscriptions: usize, +} + +impl Default for Peer { + fn default() -> Self { + Self { + announces: 100000, + subscriptions: 10000, + } + } +} + +impl Peer { + pub(crate) fn resolve(&self) -> moq_net::session::Limits { + resolve(self.announces, self.subscriptions) + } +} + +fn resolve(announces: usize, subscriptions: usize) -> moq_net::session::Limits { + let mut limits = moq_net::session::Limits::default(); + limits.announces = announces; + limits.subscriptions = subscriptions; + limits +} diff --git a/rs/moq-relay/src/relay.rs b/rs/moq-relay/src/relay.rs index 213fba085d..e9acd065aa 100644 --- a/rs/moq-relay/src/relay.rs +++ b/rs/moq-relay/src/relay.rs @@ -279,7 +279,10 @@ impl Relay { if generated && let Some(fingerprint) = certificates.fingerprints().into_iter().next() { advertise = advertise.with_fingerprint(fingerprint); } - let cluster = cluster::Cluster::new(cluster::Options::new(config.cluster).with_cache(cache))? + let mut cluster_options = cluster::Options::new(config.cluster).with_cache(cache); + cluster_options.session = config.session; + cluster_options.peer = config.peer; + let cluster = cluster::Cluster::new(cluster_options)? .with_client(client.clone()) .with_client_tls(config.connect.tls.build()?) .with_connect(config.connect.clone(), config.quic.clone()) diff --git a/rs/moq-relay/src/settings.rs b/rs/moq-relay/src/settings.rs index 4b615dd727..282467c64e 100644 --- a/rs/moq-relay/src/settings.rs +++ b/rs/moq-relay/src/settings.rs @@ -9,6 +9,10 @@ /// Every setting `moq-relay` merges, flattened tokio groups included. #[derive(usage::Config)] pub struct Settings { + #[usage(flatten)] + session: Session, + #[usage(flatten)] + peer: Peer, #[usage(flatten)] listen: moq_tokio::settings::Listen, @@ -207,3 +211,21 @@ struct Internal { #[usage(env = "MOQ_INTERNAL_LISTEN", cli("--internal-listen"))] listen: Option, } + +#[derive(usage::Config)] +#[usage(prefix = "session")] +struct Session { + #[usage(env = "MOQ_SESSION_ANNOUNCES", cli("--session-announces"))] + announces: Option, + #[usage(env = "MOQ_SESSION_SUBSCRIPTIONS", cli("--session-subscriptions"))] + subscriptions: Option, +} + +#[derive(usage::Config)] +#[usage(prefix = "peer")] +struct Peer { + #[usage(env = "MOQ_PEER_ANNOUNCES", cli("--peer-announces"))] + announces: Option, + #[usage(env = "MOQ_PEER_SUBSCRIPTIONS", cli("--peer-subscriptions"))] + subscriptions: Option, +} diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 8fe421b101..b36ff97f75 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -756,7 +756,7 @@ async fn serve_connection( stats, } = admitted; - let mut request = request.with_stats(stats); + let mut request = request.with_stats(stats).with_limits(serve.cluster.limits(&lease)); if let Some(subscriber) = subscriber { request = request.with_publisher(subscriber); } diff --git a/rs/moq-relay/src/websocket.rs b/rs/moq-relay/src/websocket.rs index 056e2ccf57..fe9ceb93ec 100644 --- a/rs/moq-relay/src/websocket.rs +++ b/rs/moq-relay/src/websocket.rs @@ -68,6 +68,7 @@ pub(crate) async fn serve_ws( .admit(state.auth.admit(request.clone())) .await .inspect_err(|err| state.cluster.refusals.record(err.into()))?; + let limits = state.cluster.limits(&lease); let token = lease.token(); let publish = state.cluster.publisher(token); // A verified client certificate marks a cluster peer, which discovers hidden @@ -105,6 +106,7 @@ pub(crate) async fn serve_ws( alpn, versions, extensions, + limits, publish, subscribe, stats, @@ -125,6 +127,8 @@ struct SessionInputs { versions: moq_net::Versions, /// The moq-transport extensions this side offers in SETUP. extensions: moq_net::setup::Extensions, + /// The request caps selected by the admitted grant. + limits: moq_net::session::Limits, publish: Option, subscribe: Option, stats: Session, @@ -162,6 +166,7 @@ where alpn, versions, extensions, + limits, publish, subscribe, stats, @@ -199,7 +204,8 @@ where let mut server = moq_net::Server::new() .with_versions(versions) .with_extensions(extensions) - .with_stats(stats); + .with_stats(stats) + .with_limits(limits); if let Some(subscribe) = subscribe { server = server.with_publisher(subscribe); } @@ -993,6 +999,7 @@ mod tests { let frozen = Arc::new(AtomicBool::new(false)); let session = SessionInputs { + limits: Default::default(), id: 0, session: String::new(), remote: "127.0.0.1:0".parse().unwrap(), @@ -1060,6 +1067,7 @@ mod tests { let timeout = Duration::from_secs(10); let session = SessionInputs { + limits: Default::default(), id: 0, session: String::new(), remote: "127.0.0.1:0".parse().unwrap(), diff --git a/rs/moq-relay/tests/auth_lifetime.rs b/rs/moq-relay/tests/auth_lifetime.rs index 3fc0696f76..8d72f17012 100644 --- a/rs/moq-relay/tests/auth_lifetime.rs +++ b/rs/moq-relay/tests/auth_lifetime.rs @@ -168,8 +168,15 @@ async fn spawn_relay_with( /// Stand up the relay's axum web stack with WebSocket enabled and return the /// port plus an abort handle. async fn spawn_ws_relay(auth: moq_relay::auth::Auth) -> (u16, tokio::task::JoinHandle<()>) { - let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); let cluster = cluster::Cluster::new(cluster::Options::default()).expect("cluster init"); + spawn_ws_relay_with(auth, cluster).await +} + +async fn spawn_ws_relay_with( + auth: moq_relay::auth::Auth, + cluster: cluster::Cluster, +) -> (u16, tokio::task::JoinHandle<()>) { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); // Stream listeners bind lazily, so this server never opens a socket; only // its certificate handle is used. @@ -1373,3 +1380,65 @@ async fn quic_and_webtransport_report_distinct_auth_transports() { relay.abort(); let _ = relay.await; } + +/// Both admission paths apply the selected cap before starting the session driver. +#[tokio::test] +async fn relay_request_limits_distinguish_clients_and_peers() { + for scheme in ["tcp", "ws"] { + for peer in [false, true] { + let mut grant = Grant::new(all(), all()); + grant.peer = peer; + let script = Script::new(grant); + let auth = build_auth(script.spawn().await); + let mut options = cluster::Options::default(); + options.session.announces = 1; + options.peer.announces = 2; + let cluster = cluster::Cluster::new(options).unwrap(); + let consumer = cluster.origin.consume(); + let mut observed = consumer.announced(); + let (port, relay) = match scheme { + "tcp" => spawn_relay_with(auth, cluster).await, + _ => spawn_ws_relay_with(auth, cluster).await, + }; + let origin = moq_tokio::origin::spawn(); + let session = client() + .with_publisher(origin.consume()) + .with_reconnect(false) + .connect(room_url(scheme, port)) + .established() + .await + .unwrap(); + let first = origin.create_broadcast("one").unwrap(); + first.announce(Default::default()).unwrap(); + tokio::time::timeout(TIMEOUT, next_update(&mut observed)) + .await + .unwrap() + .unwrap(); + let second = origin.create_broadcast("two").unwrap(); + second.announce(Default::default()).unwrap(); + if peer { + let (route, active) = tokio::time::timeout(TIMEOUT, next_update(&mut observed)) + .await + .unwrap() + .unwrap(); + assert!(active, "the peer must retain its second announcement: {route:?}"); + } else { + let error = tokio::time::timeout(TIMEOUT, session.closed()) + .await + .expect("the second client announcement exceeds the cap"); + assert!( + matches!( + error, + Err(moq_tokio::Error::MoqNet(moq_net::Error::Session( + moq_net::SessionError::TooManyRequests + ))) + ), + "{error:?}" + ); + } + drop(session); + relay.abort(); + let _ = relay.await; + } + } +} diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index 0767b1bf67..6440f75d68 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -219,6 +219,12 @@ impl Client { self } + /// Cap the announcements and subscriptions this peer can hold on a session. + pub fn with_limits(mut self, limits: moq_net::session::Limits) -> Self { + self.moq = self.moq.with_limits(limits); + self + } + /// Attach a per-connection [`moq_net::stats::Session`] context to all sessions /// opened by this client. pub fn with_stats(mut self, stats: moq_net::stats::Session) -> Self { diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 85a6609d1c..3d42bb5b50 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -1619,6 +1619,12 @@ impl Request { self } + /// Set request caps after authorizing the peer and before answering SETUP. + pub fn with_limits(mut self, limits: moq_net::session::Limits) -> Self { + self.kind = request_map!(self.kind, request => request.with_limits(limits)); + self + } + /// Accept the session, starting the MoQ session loops. /// /// Bounded by what remains of the handshake deadline: answering SETUP can wait