diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 5144fe94bd..0db52a217b 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -239,9 +239,11 @@ and auth root, split by a **tier** label chosen by the auth server's grant or `--cluster-tier`, which is what makes billing per customer or per region possible. Each run, and each group returning after its linger, announces under a fresh [epoch](/concept/moq-lite#publisher-epochs) on its route, so a restart -is a new broadcast at the same path. -[Stats](/concept/stats) describes the paths, tracks, and encodings; -read them with the [`moq-stats`](https://docs.rs/moq-stats) crate. +is a new broadcast at the same path. Each `sessions.json` row also carries +`announces_peak` and `subscriptions_peak`: the most any one session under that +root held against the per-session limits, past which a session 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. ## \[iroh] diff --git a/doc/concept/stats.md b/doc/concept/stats.md index d37d6844b5..b7a2a1f705 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -129,7 +129,7 @@ should prefer the canonical name and fall back to the legacy one. ### Presence ```json -{ "acme": { "sessions_started": 12, "sessions_ended": 10, "sessions": 12, "sessions_closed": 10 } } +{ "acme": { "sessions_started": 12, "sessions_ended": 10, "sessions": 12, "sessions_closed": 10, "announces_peak": 40, "subscriptions_peak": 900 } } ``` `sessions_started` and `sessions_ended` count connects and disconnects under an @@ -137,6 +137,12 @@ auth root on the tier, whether or not any data flows. `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. +`announces_peak` and `subscriptions_peak` are the most broadcasts announced to +the relay, and subscriptions held on it, by any one session under the root. +Compare them with the relay's per-session limits: a session that goes past one +is closed with `TOO_MANY_REQUESTS`. A peak never goes down while the root has a +session, and summing nodes takes the largest. + ### Counters Every counter is a cumulative, monotonic unsigned integer. A rate is the diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 30bd479c33..cff233dda9 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -85,6 +85,25 @@ received tracks cleanly after their delivered groups. A peer close ends received tracks and open group readers with the session error, preserving its close code. +### Session limits + +`Client::with_limits` and `Server::with_limits` take a `session::Limits`, which +caps what the peer can make one session hold: `announces` (broadcasts, or +moq-transport namespaces) and `subscriptions`. The defaults (100,000 and +10,000) suit a relay mesh; lower them for untrusted peers. A peer that goes +past either cap loses the session, closed with `TOO_MANY_REQUESTS`. Each +auth root's stats `sessions` row reports `announces_peak` and +`subscriptions_peak`, the most any one session held, so an operator sees how +close sessions come. On moq-transport drafts 14 to 16 the limits also size the +`MAX_REQUEST_ID` window advertised in SETUP, granted back as requests close. +That window counts every request, FETCH and SUBSCRIBE\_UPDATE included, so very +low limits can starve it. A request ID past it also closes the session with +`TOO_MANY_REQUESTS`, as the draft requires. + +Every length a peer declares is capped before it is buffered: 65,535 bytes for +a moq-lite control message (`ANNOUNCE_INIT` on lite 01 and 02 excepted) and +64 KiB for an object's property block. + `origin::Producer::new` returns a driver with the same `time::Driver` interface. It calls `cache::Pool::gc(now)` after each poll and folds the next cleanup time into its returned deadline. A standalone pool needs `gc(now)` diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index f93a2e639f..f3468e88ad 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -286,6 +286,8 @@ Sent when terminating the session, via the transport's session close. | ------- | ------------- | ----------- | | 0x6 | KEY_VALUE_FORMATTING_ERROR | A key-value pair was malformed, or repeated more than allowed. | | ------- | ------------- | ----------- | +| 0x7 | TOO_MANY_REQUESTS | The peer held more announcements or subscriptions than the endpoint allows. | +| ------- | ------------- | ----------- | | 0x10 | GOAWAY_TIMEOUT | The peer did not close within the GOAWAY drain deadline. | | ------- | ------------- | ----------- | | 0x11 | CONTROL_MESSAGE_TIMEOUT | The peer took too long to respond to a control message. | @@ -707,6 +709,7 @@ Most messages are prefixed with a variable-length integer indicating the number This length field does not include the length of the varint length itself. An implementation SHOULD close the connection with a PROTOCOL_VIOLATION if it receives a message with an unexpected length. +Except in FRAME, a Message Length over 65,535 bytes is unexpected: a sender MUST NOT exceed it, and a receiver MAY reject it based on the length prefix alone. The version and extensions should be used to support new fields, not the message length. ## STREAM_TYPE {#stream_type} @@ -741,7 +744,7 @@ Setup Parameter { } ~~~ -The Message Length MUST NOT exceed 65,536 bytes; a receiver MUST treat a longer SETUP as a protocol violation and MAY reject it based on the length prefix alone. +SETUP follows the [Message Length](#message-length) cap; a receiver MUST treat a longer SETUP as a protocol violation. **Parameter Count**: The number of Setup Parameters that follow. @@ -1401,7 +1404,8 @@ The `Message Length` describes the payload size on the wire. - The Subscribe Stream FIN now follows once every counted Group Stream has finished or been reset. - Added the Spread Hash tie-break after the shortest path: a hash of the requested path and the route's Hop IDs, so equal-cost advertisers of one prefix share its paths. - Added announce compression: ANNOUNCE_START gains `Path Base` and `Path Keep` to copy the head of a live advertisement's suffix, and ANNOUNCE_START and ANNOUNCE_UPDATE gain `Hop Base` and `Hop Keep` to copy the tail of a live advertisement's Hop ID list. -- Capped the SETUP Message Length at 65,536 bytes. +- Capped the Message Length of every message except FRAME at 65,535 bytes. +- Added the TOO_MANY_REQUESTS (0x7) session code, closing a session whose peer goes past the endpoint's bound on subscriptions or announcements. ## moq-lite-06 @@ -1576,7 +1580,7 @@ GOAWAY carries an optional New Session URI that asks the peer to reconnect elsew Hop IDs (see [ANNOUNCE_OK](#announce-ok) and [ANNOUNCE_START](#announce-start)) expose the relay path of a broadcast, which may reveal internal topology. A relay that does not wish to disclose its position MAY use the reserved value 0 ("unknown") instead of a stable identifier, at the cost of losing loop detection through itself (see [Routing](#routing)). The Hop ID announcement filter (see [Hop Parameter](#hop-parameter)) exists for loop avoidance, not access control: a subscriber cannot verify that a publisher honored it, so it MUST NOT be relied upon to hide a broadcast from a peer that declared its Hop ID. ## Resource Exhaustion -A peer can open many streams (subscriptions, announcements, fetches), request large announce prefixes, or advertise broad routes. Implementations SHOULD bound the number of concurrent subscriptions, announce matches, and cached groups, and SHOULD rely on QUIC flow control and stream limits to backpressure a misbehaving peer (see [ANNOUNCE_REQUEST](#announce-request)). Expiration (see [Expiration](#expiration)) bounds how long stale groups consume memory and flow control. A broad route invites a request for any covered path, each of which may start work: an advertiser SHOULD bound the work it starts, withdrawing its route before it runs out, and every refusal is terminal, so a flood of requests costs the mesh one round trip each rather than a search (see [Resolution](#resolution)). +A peer can open many streams (subscriptions, announcements, fetches), request large announce prefixes, or advertise broad routes. Implementations SHOULD bound the number of concurrent subscriptions, announce matches, and cached groups; an endpoint whose peer goes past its bound on subscriptions or announcements SHOULD close the session with TOO_MANY_REQUESTS, and SHOULD rely on QUIC flow control and stream limits to backpressure a misbehaving peer (see [ANNOUNCE_REQUEST](#announce-request)). Expiration (see [Expiration](#expiration)) bounds how long stale groups consume memory and flow control. A broad route invites a request for any covered path, each of which may start work: an advertiser SHOULD bound the work it starts, withdrawing its route before it runs out, and every refusal is terminal, so a flood of requests costs the mesh one round trip each rather than a search (see [Resolution](#resolution)). ## Datagram Injection Datagrams are routed to a subscription solely by Subscribe ID and carry no per-group authentication beyond that of the QUIC connection. On an unmodified QUIC/WebTransport connection this is sufficient, since datagrams are protected by the transport. A subscriber MUST silently drop any datagram with an unknown Subscribe ID and MUST deduplicate against groups received on streams (see [Datagrams](#datagrams)). diff --git a/js/net/src/error.ts b/js/net/src/error.ts index 6c103ecfd7..f13d7dc536 100644 --- a/js/net/src/error.ts +++ b/js/net/src/error.ts @@ -38,6 +38,8 @@ export const SessionCode = Object.freeze( ProtocolViolation: 0x3 as SessionCode, /** A key-value pair was malformed or repeated more than allowed. */ KeyValueFormatting: 0x6 as SessionCode, + /** The peer went past what the session allows: a request ID past MAX_REQUEST_ID, or too many announcements or subscriptions. */ + TooManyRequests: 0x7 as SessionCode, /** The peer did not close within the GOAWAY drain deadline. */ GoawayTimeout: 0x10 as SessionCode, /** A control message took too long. */ diff --git a/js/net/src/lite/announce.ts b/js/net/src/lite/announce.ts index c77b4b3ea0..f730530936 100644 --- a/js/net/src/lite/announce.ts +++ b/js/net/src/lite/announce.ts @@ -481,14 +481,17 @@ export class AnnounceInit { return new AnnounceInit(suffixes); } + // The whole initial set in one message, so it scales with the publisher's broadcasts. + static readonly MAX_SIZE = 64 * 1024 * 1024; + async encode(w: Writer, version: Version): Promise { AnnounceInit.#guard(version); - return Message.encode(w, this.#encode.bind(this)); + return Message.encode(w, this.#encode.bind(this), AnnounceInit.MAX_SIZE); } static async decode(r: Reader, version: Version): Promise { AnnounceInit.#guard(version); - return Message.decode(r, AnnounceInit.#decode); + return Message.decode(r, AnnounceInit.#decode, AnnounceInit.MAX_SIZE); } } diff --git a/js/net/src/lite/message.test.ts b/js/net/src/lite/message.test.ts new file mode 100644 index 0000000000..3e385e347f --- /dev/null +++ b/js/net/src/lite/message.test.ts @@ -0,0 +1,36 @@ +import { expect, test } from "bun:test"; +import { Reader, Writer } from "../stream.ts"; +import { AnnounceInit } from "./announce.ts"; +import { Subscribe } from "./subscribe.ts"; +import { Version } from "./version.ts"; + +async function prefix(size: number, version: Version): Promise { + const written: Uint8Array[] = []; + const writer = new Writer( + new WritableStream({ write: (chunk) => void written.push(new Uint8Array(chunk)) }), + version, + ); + await writer.u53(size); + writer.close(); + await writer.closed; + return written[0] ?? new Uint8Array(); +} + +test("control messages past 65,535 bytes are refused at the length prefix", async () => { + const version = Version.DRAFT_05; + const wire = await prefix(65_536, version); + await expect(Subscribe.decode(new Reader(undefined, wire, version), version)).rejects.toThrow("too large"); +}); + +test("ANNOUNCE_INIT keeps room for a large initial set", async () => { + const version = Version.DRAFT_02; + const wire = await prefix(1024 * 1024, version); + // Past the size check, so it fails waiting for a body that never arrives instead. + await expect(AnnounceInit.decode(new Reader(undefined, wire, version), version)).rejects.not.toThrow("too large"); +}); + +test("control message length 65,535 passes the prefix check", async () => { + const version = Version.DRAFT_05; + const wire = await prefix(65_535, version); + await expect(Subscribe.decode(new Reader(undefined, wire, version), version)).rejects.not.toThrow("too large"); +}); diff --git a/js/net/src/lite/message.ts b/js/net/src/lite/message.ts index b48552e554..885b7d5f7b 100644 --- a/js/net/src/lite/message.ts +++ b/js/net/src/lite/message.ts @@ -1,7 +1,10 @@ import { Reader, Writer } from "../stream.ts"; -/** The largest message body either side accepts, matching the Rust implementation. */ -const MAX_SIZE = 64 * 1024 * 1024; +/** + * The largest message body either side accepts, matching the Rust implementation: the + * same ceiling as SETUP, checked at the length prefix before the body is buffered. + */ +const MAX_SIZE = 0xffff; // Encodes a message with a varint size prefix, refusing a body the peer would reject. export async function encode(writer: Writer, f: (w: Writer) => Promise, max = MAX_SIZE) { diff --git a/js/net/src/lite/setup.test.ts b/js/net/src/lite/setup.test.ts index 7a7b805142..3b26a8b76c 100644 --- a/js/net/src/lite/setup.test.ts +++ b/js/net/src/lite/setup.test.ts @@ -166,9 +166,9 @@ test("SETUP rejects a path with malformed UTF-8", async () => { await expect(decodeParam(PARAM_PATH, new Uint8Array([0xc3, 0x28]))).rejects.toThrow(); }); -test("SETUP encode stops at the 64 KiB receive limit", async () => { +test("SETUP encode stops at the 65,535-byte receive limit", async () => { // Count, id, and a 4-byte length varint precede the path. - const atLimit = 64 * 1024 - 6; + const atLimit = 65_535 - 6; const msg = new Setup({ path: "a".repeat(atLimit) }); expect((await roundTrip(msg)).path).toBe(msg.path); @@ -178,7 +178,7 @@ test("SETUP encode stops at the 64 KiB receive limit", async () => { test("SETUP decode refuses an oversized length before reading the body", async () => { // Only the prefix is present, so reading the body would fail with a different error. - const prefix = await bytes((w) => w.u53(64 * 1024 + 1), Version.DRAFT_05); + const prefix = await bytes((w) => w.u53(65_536), Version.DRAFT_05); await expect(Setup.decode(new Reader(undefined, prefix, Version.DRAFT_05), Version.DRAFT_05)).rejects.toThrow( "too large", ); diff --git a/js/net/src/lite/setup.ts b/js/net/src/lite/setup.ts index 22b61b4268..932abec81c 100644 --- a/js/net/src/lite/setup.ts +++ b/js/net/src/lite/setup.ts @@ -24,7 +24,7 @@ const PARAM_HOP = 0x5n; const MAX_PARAMS = 64; /** SETUP only carries negotiation parameters, so it gets a far smaller cap than other messages. */ -const MAX_SETUP_SIZE = 64 * 1024; +const MAX_SETUP_SIZE = 0xffff; /** * The probe capability an endpoint advertises in SETUP. diff --git a/quest/m0/README.md b/quest/m0/README.md index fb3ae574b6..5926dc597a 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -50,6 +50,11 @@ quest, since nothing here waits on it. The [A/V clock](/quest/m1/av-clock.md) moved to m1 in the 2026-09-30 audit: it waits on the whole jitter line and is a published `@moq/watch` break. +Session caps: since #4820 a peer past a per-session cap loses the session, and +cluster links get the same 10,000-subscription default. The +[relay session limits](/quest/m0/relay-session-limits.md) quest gates the next +release (decided 2026-10-08), so a large mesh link can't flap on the cap. + ## Required - [Held group wakes](/quest/m0/held-group-wakes.md) - a group held across a route switch is woken when its successor's first timestamp or abort makes it stale, not at the next unrelated append @@ -60,7 +65,7 @@ a published `@moq/watch` break. - [End of Track Location](/quest/m0/ietf-end-of-track-location.md) - a relay forwards END_OF_TRACK at its upstream's Location, on the last group's stream, instead of re-sending it at the next group's object 0 - [End of Group status](/quest/m0/ietf-end-of-group-status.md) - an End of Group status on a stream whose header already marks the group's end is accepted, so imquic's last object per group arrives - [Subgroup at object 0](/quest/m0/ietf-first-object-zero.md) - a draft-18 subgroup with FIRST_OBJECT clear that starts at object 0 is read whole -- [Request caps](/quest/m0/request-caps.md) - lite message sizes, IETF request IDs, and per-session announces and subscriptions are bounded +- [Relay session limits](/quest/m0/relay-session-limits.md) - moq-relay sets per-session request limits, tighter for clients than peers, and the bindings name a refused request - [Prefix route fronts](/quest/m0/prefix-route-fronts.md) - a prefix route cannot be made to mint one front per requested path - [TRACK stream demand](/quest/m0/track-stream-demand.md) - a lite-05+ subscribe shows the publisher one `used` edge, not a flap between TRACK and SUBSCRIBE - [Idle fronts](/quest/m0/idle-fronts.md) - a front nobody has read for the linger ends with its per-path state, so a standing claim stops accumulating fronts and sources diff --git a/quest/m0/prefix-route-fronts.md b/quest/m0/prefix-route-fronts.md index 571e8bdd81..9f9433eb6f 100644 --- a/quest/m0/prefix-route-fronts.md +++ b/quest/m0/prefix-route-fronts.md @@ -16,11 +16,7 @@ not per viewer, so a single session can amplify it by naming paths. - Decide the bound (per route, per session, or both) and what a refusal looks like on each wire version; reuse the refusal shapes - [Request caps](/quest/m0/request-caps.md) settles. + that request caps (#4820) settled. - Cover it with a test that requests many distinct covered paths and checks the front count stays bounded, and extend the origin benchmark with a distinct-path axis. - -## Required - -- [Request caps](/quest/m0/request-caps.md) - settles the refusal shapes reused here diff --git a/quest/m1/relay-session-limits.md b/quest/m0/relay-session-limits.md similarity index 60% rename from quest/m1/relay-session-limits.md rename to quest/m0/relay-session-limits.md index 5406d9874a..ce7438af5c 100644 --- a/quest/m1/relay-session-limits.md +++ b/quest/m0/relay-session-limits.md @@ -9,19 +9,19 @@ kind instead of mapping `TooManyRequests` to Unknown. ## Plan -Blocked until request caps (#4820) lands. That PR proposes generous -defaults in moq-net (100,000 announces and 10,000 subscriptions per session), -sized for relay meshes, so viewer sessions would get loose limits. Confirm -the landed API and defaults before implementing this quest. The relay knows +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. +This gates the next release (decided 2026-10-08). 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. + 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. - -## Required - -- [Request caps](/quest/m0/request-caps.md) - the session limits and TooManyRequests error exist on main diff --git a/quest/m0/request-caps.md b/quest/m0/request-caps.md deleted file mode 100644 index b469c58adc..0000000000 --- a/quest/m0/request-caps.md +++ /dev/null @@ -1,36 +0,0 @@ -# [L] Peer-declared lengths and request counts are bounded - -## Goal - -Every length and count a peer declares is bounded before the relay buffers -or tracks it. That covers three things: - -- lite control messages, which may declare 64 MiB each - (`lite/message.rs`, buffered by `coding/reader.rs` `poll_read_more`); -- IETF FETCH object properties, decoded as a bare `Vec` with no cap; -- IETF incoming request IDs, which are never checked against the - `MAX_REQUEST_ID` we advertise on drafts 14 to 16 (`u32::MAX`, - `rs/moq-net/src/server.rs` and `client.rs`). - -Announces and subscriptions per session are capped as well. - -## Plan - -- Lite: per-message `MAX_SIZE` sized to the largest legitimate message - (SETUP already caps at 64 KiB). Anything larger is a decode error. -- Audit `rs/moq-net/src/coding/` for any other decode that allocates from a - peer length, and cap the fetch properties. This absorbs the deleted - `coding-reader-cap` quest. -- IETF drafts 14 to 16: advertise a finite `MAX_REQUEST_ID` window, grant more with - MAX_REQUEST_ID as requests close, and close with the draft's error on an ID - past it (`ietf/control.rs`, `ietf/adapter.rs`). Draft-17 removed - MAX_REQUEST_ID; there QUIC's bidi stream limit and the per-session cap - bound requests. -- A per-session cap on live announces and subscriptions on both protocols, - refused per request rather than session-fatal where the protocol allows. -- Benchmark per-session request churn swept over sessions and requests per - session, so the window refill shows no slope. -- Mirror the lite message cap in `js/net`. - -Public API: possibly caps on the session config; propose the shape in the PR. -Wire: behaviour within the drafts' existing limits. No format change. diff --git a/quest/m0/track-stream-demand.md b/quest/m0/track-stream-demand.md index cc51eb3558..ae6f57dd9c 100644 --- a/quest/m0/track-stream-demand.md +++ b/quest/m0/track-stream-demand.md @@ -41,10 +41,9 @@ Decisions (2026-10-07): parks the track and drops its copy at once, letting go upstream. Waiting for the first response costs one stream held for about a round trip. - A held TRACK stream is demand without a subscription, so it counts against - the per-session subscription cap from - [Request caps](/quest/m0/request-caps.md); a peer can't hold more interest - than it could by subscribing. A held TRACK and the SUBSCRIBE for the same - track on the same session share one slot: the SUBSCRIBE takes over the + the per-session subscription cap (`session::Limits`, from #4820); a peer + can't hold more interest than it could by subscribing. A held TRACK and the + SUBSCRIBE for the same track on the same session share one slot: the SUBSCRIBE takes over the TRACK's reservation instead of needing a second, so a session at the cap can still turn its held TRACKs into subscriptions (decided 2026-10-08 from review). @@ -60,16 +59,12 @@ stays subscribed, on lite-05, 06, and 07, direct and through one relay. It fails on `main` today. The relay case controls the ordering so the downstream TRACK FIN is handled before its SUBSCRIBE. A boundary test fills the per-session subscription cap with held TRACK streams, turns each into a -live SUBSCRIBE with no `unused` edge, and checks that one more TRACK is -refused. JS counterparts for the JS side. +live SUBSCRIBE with no `unused` edge, and checks that one more TRACK closes +the session with TOO_MANY_REQUESTS. JS counterparts for the JS side. Public API: none. Wire: semantics only (holding the TRACK stream open), no new fields. -## Required - -- [Request caps](/quest/m0/request-caps.md) - the per-session subscription cap a held TRACK stream counts against - ## Related - [Idle fronts](/quest/m0/idle-fronts.md) - found in the same transcode-pool report diff --git a/quest/m1/README.md b/quest/m1/README.md index 99c369dcf0..aa81d1f254 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -89,7 +89,6 @@ blocks. The quests that gated m0 lines moved under them. - [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - `@moq/net` sends and receives datagram groups over moq-transport, like Rust - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface - [JS request window](/quest/m1/js-request-window.md) - @moq/net grants MAX_REQUEST_ID as requests close, so a peer never stalls after ~21k requests -- [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 - [Churn with held subscriptions](/quest/m1/session-churn-held.md) - opening and closing a request costs the same with 1 or 1,024 held subscriptions - [Wide varint tests](/quest/m1/varint-test-gaps.md) - wide sequences and both varint forms are tested where they differ - [Export linger](/quest/m1/export-ts-linger.md) - `moq export ts --linger` fails on an export error while the broadcast is live, and lingers once it closes or is replaced diff --git a/quest/m1/auth/not-supported.md b/quest/m1/auth/not-supported.md index 236e81d736..21d565c768 100644 --- a/quest/m1/auth/not-supported.md +++ b/quest/m1/auth/not-supported.md @@ -31,8 +31,7 @@ Decided while planning the follow-ups of AUTH_ERROR. Rust `SessionError` and JS `SessionCode` round-trip it, so a session closed for an unsupported feature stops reading as INTERNAL_ERROR. - **The message ceiling is not this quest's** (decided in the 2026-10-06 - audit): [Request caps](/quest/m0/request-caps.md) (#4820) caps lite - messages at 65,535 bytes, moq-transport's 16-bit control Message Length, so + audit): request caps (#4820) caps lite messages at 65,535 bytes, moq-transport's 16-bit control Message Length, so a grant too large for AUTH_OK is too large on both wires. This quest only answers that case with NOT_SUPPORTED. - **Compatibility.** An old peer that receives 0x30 treats it as an @@ -50,5 +49,4 @@ code. ## Related -- [Request caps](/quest/m0/request-caps.md) - owns the 65,535-byte lite message ceiling that makes a grant too large for AUTH_OK - [AUTH violations](/quest/m1/auth/violations.md) - AUTH protocol violations close the session on the same path (AUTH endings landed in #4550) diff --git a/quest/m1/js-session-parity.md b/quest/m1/js-session-parity.md index add3385adb..2d29a038a6 100644 --- a/quest/m1/js-session-parity.md +++ b/quest/m1/js-session-parity.md @@ -7,10 +7,9 @@ relay behaves like a Rust one. ## Plan -- Per-session caps (#4820, still open): once Rust ships live announce and - subscription caps, add them to JS with the same defaults (100,000 and - 10,000 in #4820 today) and close the session with TOO_MANY_REQUESTS (0x7) - past them, as Rust does. JS's own request-ID window is +- Per-session caps (#4820, landed): add Rust's live announce and + subscription caps to JS with the same defaults (100,000 and 10,000) and + close the session with TOO_MANY_REQUESTS (0x7) past them, as Rust does. JS's own request-ID window is [JS request window](/quest/m1/js-request-window.md). - Pending tail (#4225, still open): once Rust readers of a received track wait for missing groups below a declared end until the tail settles, mirror @@ -23,10 +22,6 @@ relay behaves like a Rust one. Test each against the Rust behavior, with mocked time. -## Required - -- [Request caps](/quest/m0/request-caps.md) - #4820's per-session caps land in Rust first - ## Related - [JS request window](/quest/m1/js-request-window.md) - JS grants request IDs as requests close, the other half of JS request limits diff --git a/quest/m1/session-churn-held.md b/quest/m1/session-churn-held.md index 1c891c0864..124f172c99 100644 --- a/quest/m1/session-churn-held.md +++ b/quest/m1/session-churn-held.md @@ -4,24 +4,19 @@ Opening and closing a request on a session costs the same whether the session holds 1 or 1,024 subscriptions. The `session_churn_held` benchmark -proposed by request caps (#4820) is flat in held subscriptions. +added by request caps (#4820) is flat in held subscriptions. ## Plan -Blocked until request caps (#4820) lands with the benchmark. Measured on -that PR: about 70 µs per churn at held=1 and 10 to 20 ms at held=1024 across -4 sessions, equally on lite-06 and draft-22. Neither has a -request window, so the slope predates request caps. Profile first, find what +Measured on request caps (#4820): about 70 µs per churn at held=1 and 10 to +20 ms at held=1024 across 4 sessions, equally on lite-06 and draft-22. Neither +has a request window, so the slope predates request caps. Profile first, find what scans the held subscriptions on each churn, and make that path proportional to what it touches. Keep the bench swept over both sessions and held subscriptions. Public API: none. Wire: none. -## Required - -- [Request caps](/quest/m0/request-caps.md) - the held-subscription churn benchmark exists on main - ## Related - [Reproducible relay CPU and allocation profiles](/quest/m1/performance-profiles.md) - the profiling setup this can use diff --git a/quest/m2/ietf-malformed-close.md b/quest/m2/ietf-malformed-close.md index be38fccdfe..83ca65b7e1 100644 --- a/quest/m2/ietf-malformed-close.md +++ b/quest/m2/ietf-malformed-close.md @@ -132,5 +132,4 @@ behaviour moves closer to the drafts. ## Related - [Request tokens](/quest/m1/auth/request-token.md) - owns request-token decode and AUTH_TOKEN_CACHE_OVERFLOW -- [Request caps](/quest/m0/request-caps.md) - bounds peer lengths and counts; shares the decode paths - [moq-transport request codes](/quest/m2/ietf-request-codes.md) - the request-level half of the same validator report diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 0441c047fa..c949251303 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -20,6 +20,11 @@ //! one small frame per track per tick. Throughput counts tracks, so a flat //! elements/s across a sweep means the cost per track stays constant. //! +//! `session_churn_*` times one request opened and closed by a session that already +//! holds others open, swept over sessions and requests held per session. On +//! moq-transport 14 every close is credited back through the MAX_REQUEST_ID window, +//! so a slope here is that bookkeeping growing with the sessions or the requests. +//! //! Everything runs on one current-thread runtime, so the only work measured is //! the protocol and model code, never scheduling across threads. //! @@ -637,6 +642,107 @@ fn allocations() { } } +/// Versions for the churn sweep: the one draft with a MAX_REQUEST_ID window, the +/// newest draft without it, and lite. +const CHURN_VERSIONS: [&str; 3] = ["moq-transport-14", "moq-transport-22", "moq-lite-06"]; + +/// Sessions on one relay, each holding `held` subscriptions open while it churns one more. +#[derive(Clone, Copy)] +struct Churn { + sessions: usize, + held: usize, +} + +impl Churn { + fn id(&self, version: &str) -> String { + format!("{version}/sessions={}/held={}", self.sessions, self.held) + } +} + +/// One relay publishing a broadcast with `held` shared tracks plus one churn track per session. +struct ChurnRoom { + _cluster: Cluster, + _held: Vec, + _viewers: Vec, + /// Each session's view of the broadcast, to open the churned subscription on. + broadcasts: Vec, + /// The relay's side of each session's churn track, to wait out its close. + churn: Vec, +} + +impl ChurnRoom { + async fn new(version: &str, shape: Churn) -> Self { + let mut cluster = Cluster::new( + version, + Shape { + relays: 1, + publishers: 1, + viewers: 0, + ..Shape::BASE + }, + ) + .await; + let published = &cluster._broadcasts[0]; + let held: Vec<_> = (0..shape.held).map(|i| format!("held{i}")).collect(); + let tracks: Vec<_> = held + .iter() + .map(|name| published.create_track(name.as_str(), None).unwrap()) + .collect(); + let churn: Vec<_> = (0..shape.sessions) + .map(|i| published.create_track(format!("churn{i}"), None).unwrap()) + .collect(); + + let mut viewers = Vec::new(); + let mut broadcasts = Vec::new(); + for _ in 0..shape.sessions { + let mut viewer = cluster.join(0).await; + viewer.watch(&path(0), &held).await; + broadcasts.push(viewer.origin.consume().routed_broadcast(&path(0)).await.unwrap()); + viewers.push(viewer); + } + Self { + _cluster: cluster, + _held: tracks, + _viewers: viewers, + broadcasts, + churn, + } + } + + /// Open and close one subscription per iteration, round-robin over the sessions. + async fn measure(&self, iters: u64) -> Duration { + let start = Instant::now(); + for iter in 0..iters as usize { + let session = iter % self.broadcasts.len(); + let track = self.broadcasts[session].track(&format!("churn{session}")).unwrap(); + let subscriber = track.subscribe(None).await.unwrap(); + // The relay saw the request, rather than a cached track answering it locally. + self.churn[session].demand().used().await.unwrap(); + drop(subscriber); + drop(track); + self.churn[session].demand().unused().await.unwrap(); + } + start.elapsed() + } +} + +fn churn(c: &mut Criterion, name: &str, shapes: impl IntoIterator) { + let rt = runtime(); + let shapes: Vec<_> = shapes.into_iter().collect(); + let mut group = c.benchmark_group(format!("session_churn_{name}")); + group.throughput(Throughput::Elements(1)); + for version in CHURN_VERSIONS { + for shape in &shapes { + let mut room = None; + group.bench_function(BenchmarkId::from_parameter(shape.id(version)), |b| { + let room = room.get_or_insert_with(|| rt.block_on(ChurnRoom::new(version, *shape))); + b.iter_custom(|iters| rt.block_on(room.measure(iters))) + }); + } + } + group.finish(); +} + /// Withdraw a batch across a full mesh, sweeping both peers and prefixes. /// Paused protocol time lets the quiet-point timeout drain pending work without /// including a real wait in the measured duration. @@ -767,6 +873,9 @@ fn session(c: &mut Criterion) { ); join(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base })); + churn(c, "sessions", [1, 16, 128].map(|sessions| Churn { sessions, held: 16 })); + churn(c, "held", [1, 64, 1024].map(|held| Churn { sessions: 4, held })); + // Production (2026-09): ~34 nodes x >=5 projects x ~24 tracks. let prod = Dash { relays: 35, diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index c1e1b772f2..4a511e7345 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -18,6 +18,7 @@ pub struct Client { setup_authority: Option, cost: Option, peer_hop: Option, + limits: crate::session::Limits, } impl Client { @@ -58,6 +59,12 @@ impl Client { self.with_publisher(&origin).with_subscriber(origin) } + /// Cap what the server can make this session hold. Defaults to [`session::Limits::default`](crate::session::Limits::default). + pub fn with_limits(mut self, limits: crate::session::Limits) -> Self { + self.limits = limits; + self + } + /// Restrict which protocol versions to offer, in preference order. /// Defaults to every version this crate supports. pub fn with_versions(mut self, versions: Versions) -> Self { @@ -193,6 +200,7 @@ impl Client { let start = lite::start(lite::Config { runtime: runtime.clone(), + limits: self.limits, session: session.clone(), setup_stream: None, publish, @@ -270,6 +278,7 @@ impl Client { // We advertise the request path in our SETUP for URL-less transports. let (protocol, goaway, setup) = ietf::start(ietf::Config { runtime: runtime.clone(), + limits: self.limits, session: session.clone(), setup: None, request_id_max: None, @@ -352,7 +361,11 @@ impl Client { let ietf_encoding = ietf::Version::try_from(encoding).map_err(|_| Error::Version)?; let mut parameters = ietf::Parameters::default(); - parameters.set_varint(ietf::ParameterVarInt::MaxRequestId, u32::MAX as u64); + // The server's requests, admitted up to our limits and granted back as they close. + parameters.set_varint( + ietf::ParameterVarInt::MaxRequestId, + ietf::initial_max_request_id(self.limits.requests(), false), + ); parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); // Advertise the request path in-band (draft 14-16), same as the lite-05 SETUP. if let Some(path) = &self.setup_path { @@ -386,6 +399,7 @@ impl Client { let stream = stream.with_version(v); let start = lite::start(lite::Config { runtime: runtime.clone(), + limits: self.limits, session: session.clone(), setup_stream: Some(stream), publish: publish.clone(), @@ -423,6 +437,7 @@ impl Client { // Draft 14-16: the path rode in the bidi SETUP above, not the uni one. let (protocol, goaway, setup) = ietf::start(ietf::Config { runtime: runtime.clone(), + limits: self.limits, session: session.clone(), setup: Some(stream), request_id_max, diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index 9221fbef6c..0f9b03a98f 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -556,7 +556,7 @@ mod tests { fn oversized_setup_is_refused_at_the_prefix() { use crate::{Version, lite, setup}; for version in [lite::Version::Lite01, lite::Version::Lite02] { - for size in [65537u64, (1 << 40) + 1] { + for size in [65536u64, (1 << 40) + 1] { let mut prefix = vec![0x20]; Encoder::new(&mut prefix, version.into()).varint(size).unwrap(); refuses_frame::(&prefix, Version::Lite(version), true); @@ -566,7 +566,7 @@ mod tests { } for version in [lite::Version::Lite05, lite::Version::Lite06] { let mut prefix = Vec::new(); - Encoder::new(&mut prefix, version.into()).varint(65537).unwrap(); + Encoder::new(&mut prefix, version.into()).varint(65536).unwrap(); refuses_frame::(&prefix, version, true); } } @@ -576,7 +576,7 @@ mod tests { use crate::{Version, lite, setup}; for version in [lite::Version::Lite01, lite::Version::Lite02] { let mut prefix = vec![0x20]; - Encoder::new(&mut prefix, version.into()).varint(65536).unwrap(); + Encoder::new(&mut prefix, version.into()).varint(65535).unwrap(); assert!(matches!( setup::Client::decode_slice(&prefix, Version::Lite(version)), Err(DecodeError::Short) @@ -589,7 +589,7 @@ mod tests { } for version in [lite::Version::Lite05, lite::Version::Lite06] { let mut prefix = Vec::new(); - Encoder::new(&mut prefix, version.into()).varint(65536).unwrap(); + Encoder::new(&mut prefix, version.into()).varint(65535).unwrap(); assert!(matches!( lite::Setup::decode_slice(&prefix, version), Err(DecodeError::Short) diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index a606cdedcc..d9b38c9eae 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -32,6 +32,12 @@ pub enum SessionError { #[error("key-value formatting error")] KeyValueFormatting, + /// The peer went past what the session allows: a request ID at or past the + /// MAX_REQUEST_ID we advertised (moq-transport drafts 14 to 16), or more announcements + /// or subscriptions than its [`crate::session::Limits`]. + #[error("too many requests")] + TooManyRequests, + /// The peer did not close within the GOAWAY drain deadline. #[error("goaway timeout")] GoawayTimeout, @@ -62,6 +68,7 @@ impl SessionError { Self::Unauthorized => 0x2, Self::ProtocolViolation => 0x3, Self::KeyValueFormatting => 0x6, + Self::TooManyRequests => 0x7, Self::GoawayTimeout => 0x10, Self::Timeout => 0x11, Self::Version => 0x15, @@ -82,6 +89,7 @@ impl SessionError { 0x2 => Self::Unauthorized, 0x3 => Self::ProtocolViolation, 0x6 => Self::KeyValueFormatting, + 0x7 => Self::TooManyRequests, 0x10 => Self::GoawayTimeout, 0x11 => Self::Timeout, 0x15 => Self::Version, @@ -376,6 +384,11 @@ pub enum Error { #[error("too many parameters")] TooManyParameters, + /// The peer already holds as many requests, announcements, or subscriptions as this + /// session allows (see [`crate::session::Limits`]). + #[error("too many requests")] + TooManyRequests, + /// The peer offered an ALPN this endpoint doesn't recognize, so no version could be /// negotiated. A connect-time error. #[error("unknown ALPN: {0}")] @@ -538,6 +551,7 @@ impl From<&Error> for SessionError { Error::Unauthorized => Self::Unauthorized, Error::Version | Error::UnknownAlpn(_) => Self::Version, Error::TooManyParameters => Self::KeyValueFormatting, + Error::TooManyRequests => Self::TooManyRequests, Error::GoawayTimeout => Self::GoawayTimeout, Error::Timeout => Self::Timeout, Error::ProtocolViolation @@ -634,6 +648,7 @@ mod tests { SessionError::Unauthorized, SessionError::ProtocolViolation, SessionError::KeyValueFormatting, + SessionError::TooManyRequests, SessionError::GoawayTimeout, SessionError::Timeout, SessionError::Version, @@ -650,6 +665,7 @@ mod tests { // The moq-transport codes we reuse must keep moq-transport's values. assert_eq!(SessionError::Unauthorized.to_code(), 0x2); + assert_eq!(SessionError::TooManyRequests.to_code(), 0x7); assert_eq!(SessionError::GoawayTimeout.to_code(), 0x10); assert_eq!(SessionError::Version.to_code(), 0x15); diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index c0de8cb267..8c39957c75 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -12,7 +12,7 @@ use crate::{ ietf::{self, RequestId}, }; -use super::{Control, Message, Version}; +use super::{Control, Message, Version, control::RequestPermit}; // === Message Queues === @@ -120,6 +120,9 @@ pub struct VirtualRecvStream { /// The request this stream serves, known up front when the peer opened it and /// filled in by the first write when we did. request_id: Arc>>, + /// Holds the peer's request open against our MAX_REQUEST_ID window until the + /// handler drops this stream. Only on a stream the peer opened. + _permit: Option, } impl VirtualRecvStream { @@ -131,6 +134,7 @@ impl VirtualRecvStream { park: kio::Park::default(), shared: Arc::downgrade(&shared), request_id, + _permit: None, } } @@ -664,14 +668,20 @@ impl Shared { } /// Register the peer's new request and queue its stream for accept_bi. - fn open_incoming(self: &Arc, request_id: RequestId, raw: Bytes) -> Result<(), Error> { + fn open_incoming( + self: &Arc, + request_id: RequestId, + raw: Bytes, + permit: Option, + ) -> Result<(), Error> { let follow = Queue::new(); - let recv = VirtualRecvStream::new( + let mut recv = VirtualRecvStream::new( raw, follow.clone(), Arc::clone(self), Arc::new(Mutex::new(Some(request_id))), ); + recv._permit = permit; let send = VirtualSendStream::new(self.control.clone()); self.streams.lock().unwrap().insert(request_id, follow.writer()); if !self.incoming.push((send, recv)) { @@ -749,6 +759,10 @@ impl ControlStreamAdapter { if let Poll::Ready(res) = waiter.poll_future(read.as_mut()) { return Poll::Ready(res); } + // Queued before the write drains, so a grant goes out in the same turn. + while let Poll::Ready(max) = self.control.poll_grant(waiter) { + self.send_max_request_id(max); + } if let Poll::Ready(res) = waiter.poll_future(write.as_mut()) { return Poll::Ready(res); } @@ -796,6 +810,22 @@ impl ControlStreamAdapter { } } + /// Queue a MAX_REQUEST_ID granting the peer room up to `max`. + fn send_max_request_id(&self, max: RequestId) { + let mut raw = Vec::new(); + let mut w = crate::coding::Encoder::new(&mut raw, self.version.into()); + let msg = ietf::MaxRequestId { request_id: max }; + if let Err(err) = w + .varint(ietf::MaxRequestId::ID) + .and_then(|()| crate::coding::Encode::encode(&msg, &mut w, self.version)) + { + tracing::warn!(%err, "failed to encode MAX_REQUEST_ID"); + return; + } + // A closed control stream means the session is ending; there is no one to grant. + let _ = self.shared.control.push(raw.into()); + } + /// Writer task: drains the queue and writes to the control stream. async fn run_write(&self, mut writer: Writer) -> Result<(), Error> { while let Some(msg) = self.shared.control.pop().await { @@ -824,8 +854,17 @@ impl ControlStreamAdapter { // Classify and route match classify(type_id, &body, self.version, &self.shared.namespaces)? { - Route::NewRequest(request_id) => self.shared.open_incoming(request_id, raw)?, - Route::Response(request_id) | Route::FollowUp(request_id) => self.shared.push(request_id, raw), + Route::NewRequest(request_id) => { + let permit = self.control.accept(request_id)?; + self.shared.open_incoming(request_id, raw, permit)? + } + Route::Response(request_id) => self.shared.push(request_id, raw), + Route::FollowUp { own, target } => { + // SUBSCRIBE_UPDATE takes a request ID of its own (draft-14 section 9.1), + // done as soon as it is delivered, so it is granted straight back. + drop(self.control.accept(own)?); + self.shared.push(target, raw) + } Route::CloseStream(request_id) => self.shared.close(request_id, raw), Route::MaxRequestId(max) => self.control.max_request_id(max), Route::Ignore => {} @@ -972,8 +1011,9 @@ fn classify(type_id: u64, body: &Bytes, version: Version, namespaces: &Namespace // Follow-up messages: route to the request being updated, not the update's own id. ietf::SubscribeUpdate::ID => { - let id = decode_update_target(body, version)?; - Ok(Route::FollowUp(id)) + let own = decode_request_id(body, version)?; + let target = decode_update_target(body, version)?; + Ok(Route::FollowUp { own, target }) } // Close stream messages @@ -1033,7 +1073,9 @@ fn classify(type_id: u64, body: &Bytes, version: Version, namespaces: &Namespace let id = decode_request_id(body, version)?; Ok(Route::MaxRequestId(id)) } - ietf::RequestsBlocked::ID => Err(Error::UnexpectedMessage), + // The peer reached the MAX_REQUEST_ID we advertised and waits for a grant, which + // follows on its own as its requests close. + ietf::RequestsBlocked::ID => Ok(Route::Ignore), // Terminal ietf::GoAway::ID => Ok(Route::GoAway), @@ -1147,7 +1189,11 @@ impl crate::transport::poll::Session for Con enum Route { NewRequest(RequestId), Response(RequestId), - FollowUp(RequestId), + /// An update: its own request ID counts against the window, and it goes to `target`. + FollowUp { + own: RequestId, + target: RequestId, + }, CloseStream(RequestId), MaxRequestId(RequestId), /// Nothing to route: the message named a request that is already gone. @@ -1295,12 +1341,17 @@ mod tests { }; let body = encode_body(&update, version); let raw = encode_raw(ietf::SubscribeUpdate::ID, &body, version); - let Route::FollowUp(routed) = classify_msg(version, ietf::SubscribeUpdate::ID, &body).unwrap() else { + let Route::FollowUp { own, target: routed } = + classify_msg(version, ietf::SubscribeUpdate::ID, &body).unwrap() + else { panic!("{version} did not route the update as a follow-up"); }; + assert_eq!(own, update_id, "{version}"); let shared = Arc::new(Shared::default()); - shared.open_incoming(target_id, Bytes::from_static(b"request")).unwrap(); + shared + .open_incoming(target_id, Bytes::from_static(b"request"), None) + .unwrap(); let (_, mut recv) = shared.incoming.pop().await.unwrap(); assert_eq!( recv.read_chunk(usize::MAX).await.unwrap(), @@ -1490,7 +1541,9 @@ mod tests { ) .unwrap(); assert!(matches!(route, Route::NewRequest(id) if id == request_id)); - shared.open_incoming(request_id, encode_msg(&msg, version)).unwrap(); + shared + .open_incoming(request_id, encode_msg(&msg, version), None) + .unwrap(); let (_, mut recv) = shared.incoming.pop().await.unwrap(); @@ -1762,11 +1815,110 @@ mod tests { } } + /// A peer's control stream carrying a bare SUBSCRIBE for each request ID: the + /// adapter routes on the ID alone, so the rest of the body is never read. + async fn windowed( + window: u64, + ids: &[u64], + then: &[u8], + ) -> ( + ControlStreamAdapter, + Reader, + Writer, + crate::lite::test_transport::Log, + ) { + use crate::lite::test_transport::{Log, ScriptedSession, SinkSend}; + use crate::transport::poll::Session as _; + + const VERSION: Version = Version::Draft14; + let mut script = Vec::new(); + for id in ids { + let body = make_body_with_request_id(*id, VERSION); + script.extend_from_slice(&encode_raw(ietf::Subscribe::ID, &body, VERSION)); + } + script.extend_from_slice(then); + let mut session = ScriptedSession::new(script); + let (_, recv) = session.open_bi().await.unwrap(); + // We are the server, so the client's request IDs are even. + let control = Control::new(None, false).with_window(window, false); + let adapter = ControlStreamAdapter::new(session, control, VERSION); + let log = Log::default(); + let writer = Writer::new(SinkSend::new(log.clone()), VERSION); + (adapter, Reader::new(recv, VERSION), writer, log) + } + + /// A request ID at the MAX_REQUEST_ID we advertised closes the session with the + /// draft's TOO_MANY_REQUESTS, before the request is queued. + #[moq_net_sim::test] + async fn a_request_past_the_window_closes_the_session() { + let (adapter, reader, writer, _) = windowed(1, &[0, 2], &[]).await; + let (_, goaway) = crate::goaway::Handle::new(true); + let err = adapter.run(reader, writer, goaway).await.unwrap_err(); + assert!( + matches!(err, Error::Session(crate::SessionError::TooManyRequests)), + "{err:?}" + ); + } + + /// An update's own request ID counts against the window, not the subscription it targets. + #[moq_net_sim::test] + async fn an_update_past_the_window_closes_the_session() { + const VERSION: Version = Version::Draft14; + let update = ietf::SubscribeUpdate { + request_id: RequestId(4), + subscription_request_id: Some(RequestId(0)), + start_location: ietf::Location { group: 0, object: 0 }, + end_group: 0, + subscriber_priority: 1, + forward: true, + }; + let update = encode_raw(ietf::SubscribeUpdate::ID, &encode_body(&update, VERSION), VERSION); + // The target is in the window and the update's own ID is past it. + let (adapter, reader, writer, _) = windowed(2, &[0], &update).await; + let (_, goaway) = crate::goaway::Handle::new(true); + let err = adapter.run(reader, writer, goaway).await.unwrap_err(); + assert!( + matches!(err, Error::Session(crate::SessionError::TooManyRequests)), + "{err:?}" + ); + } + + /// Requests that end are granted back with MAX_REQUEST_ID, once half the window has. + #[moq_net_sim::test] + async fn retired_requests_are_granted_back() { + use crate::transport::poll::Session as _; + + const VERSION: Version = Version::Draft14; + let (mut adapter, reader, writer, log) = windowed(2, &[0, 2], &[]).await; + let (_, goaway) = crate::goaway::Handle::new(true); + let runner = adapter.clone(); + let mut run = std::pin::pin!(runner.run(reader, writer, goaway)); + assert!(futures::poll!(run.as_mut()).is_pending()); + + let first = adapter.accept_bi().await.unwrap(); + let _second = adapter.accept_bi().await.unwrap(); + assert!(log.writes.lock().unwrap().is_empty(), "nothing has ended yet"); + + drop(first); + assert!(futures::poll!(run.as_mut()).is_pending()); + + let mut expected = Vec::new(); + crate::coding::Encoder::new(&mut expected, VERSION.into()) + .varint(ietf::MaxRequestId::ID) + .unwrap(); + ietf::MaxRequestId { + request_id: RequestId(6), + } + .encode(&mut crate::coding::Encoder::new(&mut expected, VERSION.into()), VERSION) + .unwrap(); + assert_eq!(*log.writes.lock().unwrap(), expected.to_vec()); + } + #[test] fn queued_incoming_stream_does_not_keep_shared_alive() { let shared = Arc::new(Shared::default()); let weak = Arc::downgrade(&shared); - shared.open_incoming(RequestId(7), Bytes::new()).unwrap(); + shared.open_incoming(RequestId(7), Bytes::new(), None).unwrap(); drop(shared); assert!(weak.upgrade().is_none()); } diff --git a/rs/moq-net/src/ietf/control.rs b/rs/moq-net/src/ietf/control.rs index b7ebd1607a..3ba504d7e2 100644 --- a/rs/moq-net/src/ietf/control.rs +++ b/rs/moq-net/src/ietf/control.rs @@ -1,11 +1,35 @@ use std::task::Poll; -use crate::{Error, ietf::RequestId}; +use crate::{Error, SessionError, ietf::RequestId}; + +/// The MAX_REQUEST_ID that first admits `window` requests from a peer. +/// +/// A client's request IDs are even from 0 and a server's odd from 1, each stepping by 2. +pub(crate) fn initial_max_request_id(window: u64, peer_client: bool) -> u64 { + let parity = if peer_client { 0 } else { 1 }; + window.saturating_mul(2).saturating_add(parity).min((1 << 62) - 1) +} struct ControlState { request_id_next: RequestId, /// None means no flow control (draft17 removed MaxRequestId). request_id_max: Option, + /// The peer's requests, flow controlled on drafts 14 to 16 only. + incoming: Option, +} + +/// The MAX_REQUEST_ID window we grant the peer. +struct Incoming { + /// How many requests the peer may hold open at once. + window: u64, + /// The MAX_REQUEST_ID in force: every request ID the peer uses must be below it. + max: u64, + /// The last MAX_REQUEST_ID the peer was told, so each grant goes out once. + sent: u64, + /// Requests accepted and not yet retired. + live: u64, + /// Requests retired since the last grant. + retired: u64, } #[derive(Clone)] @@ -19,15 +43,81 @@ impl Control { state: kio::Shared::new(ControlState { request_id_next: if client { RequestId(0) } else { RequestId(1) }, request_id_max, + incoming: None, }), } } + /// Flow control the peer's requests (drafts 14 to 16), admitting `window` at once. + /// + /// `client` is our own role, as passed to [`Self::new`]; the window starts at what + /// [`initial_max_request_id`] advertised in our SETUP. + pub fn with_window(self, window: u64, client: bool) -> Self { + let max = initial_max_request_id(window, !client); + self.state.lock().incoming = Some(Incoming { + window, + max, + sent: max, + live: 0, + retired: 0, + }); + self + } + pub fn max_request_id(&self, max: RequestId) { // Mutating through the guard wakes any `next_request_id` waiting on the limit. self.state.lock().request_id_max = Some(max); } + /// Admit a request the peer opened, held open until the returned permit drops. + /// + /// Closes the session with TOO_MANY_REQUESTS when the ID is at or past the + /// MAX_REQUEST_ID we advertised, or the peer reused IDs to hold more than the window. + /// `None` when the peer's requests are not flow controlled. + pub fn accept(&self, id: RequestId) -> Result, Error> { + let mut state = self.state.lock(); + let Some(incoming) = &mut state.incoming else { + return Ok(None); + }; + if id.0 >= incoming.max || incoming.live >= incoming.window { + tracing::warn!(id = %id, max = incoming.max, live = incoming.live, "peer exceeded MAX_REQUEST_ID"); + return Err(SessionError::TooManyRequests.into()); + } + incoming.live += 1; + Ok(Some(RequestPermit(self.clone()))) + } + + /// Retire one accepted request, growing the window once half of it has retired. + /// + /// Batching keeps the MAX_REQUEST_ID traffic a fraction of the requests it grants. + fn retire(&self) { + let mut state = self.state.lock(); + let Some(incoming) = &mut state.incoming else { + return; + }; + incoming.live -= 1; + incoming.retired += 1; + if incoming.retired.saturating_mul(2) >= incoming.window { + let grant = incoming.retired.saturating_mul(2); + incoming.max = incoming.max.saturating_add(grant).min((1 << 62) - 1); + incoming.retired = 0; + } + } + + /// Poll for a MAX_REQUEST_ID the peer has not been told yet. + /// + /// Pending forever when the peer's requests are not flow controlled. + pub fn poll_grant(&self, waiter: &kio::Waiter) -> Poll { + let ready = self.state.poll(waiter, |state| match &state.incoming { + Some(incoming) if incoming.max > incoming.sent => Poll::Ready(()), + _ => Poll::Pending, + }); + let mut state = std::task::ready!(ready); + let incoming = state.incoming.as_mut().expect("checked above"); + incoming.sent = incoming.max; + Poll::Ready(RequestId(incoming.max)) + } + /// Allocate the next request_id, blocking until MAX_REQUEST_ID allows it. /// /// `runtime` arms the give-up timer, so the wait cannot outlive the caller's @@ -58,3 +148,72 @@ impl Control { .await } } + +/// One request the peer holds open, retired (and eventually granted back) on drop. +pub(super) struct RequestPermit(Control); + +impl Drop for RequestPermit { + fn drop(&mut self) { + self.0.retire(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn granted(control: &Control) -> Option { + match control.poll_grant(&kio::Waiter::noop()) { + Poll::Ready(max) => Some(max.0), + Poll::Pending => None, + } + } + + /// The window admits exactly its size, refuses the next ID with TOO_MANY_REQUESTS, + /// and grants room back in batches as requests retire. + #[test] + fn window_refills_as_requests_retire() { + // We are the server, so the client's IDs are even: 0, 2, 4, 6. + let control = Control::new(None, false).with_window(4, false); + assert_eq!(initial_max_request_id(4, true), 8); + + let mut open: Vec<_> = (0..4) + .map(|i| control.accept(RequestId(i * 2)).unwrap().unwrap()) + .collect(); + assert!(matches!( + control.accept(RequestId(8)), + Err(Error::Session(SessionError::TooManyRequests)) + )); + assert_eq!(granted(&control), None, "nothing retired yet"); + + // One retired is below half the window, so nothing is granted yet. + open.pop(); + assert_eq!(granted(&control), None); + + // Half the window retired: two more IDs. + open.pop(); + assert_eq!(granted(&control), Some(12)); + assert_eq!(granted(&control), None, "each grant goes out once"); + + let _a = control.accept(RequestId(8)).unwrap(); + let _b = control.accept(RequestId(10)).unwrap(); + assert!(control.accept(RequestId(12)).is_err()); + } + + /// A peer reusing request IDs below the window still cannot hold more than it. + #[test] + fn reused_ids_cannot_exceed_the_window() { + let control = Control::new(None, false).with_window(2, false); + let _a = control.accept(RequestId(0)).unwrap(); + let _b = control.accept(RequestId(0)).unwrap(); + assert!(control.accept(RequestId(2)).is_err()); + } + + /// Draft-17+ has no MAX_REQUEST_ID: nothing is tracked or granted. + #[test] + fn unlimited_without_a_window() { + let control = Control::new(None, true); + assert!(control.accept(RequestId(u64::MAX)).unwrap().is_none()); + assert_eq!(granted(&control), None); + } +} diff --git a/rs/moq-net/src/ietf/fetch.rs b/rs/moq-net/src/ietf/fetch.rs index b16d0acd5b..b256878b75 100644 --- a/rs/moq-net/src/ietf/fetch.rs +++ b/rs/moq-net/src/ietf/fetch.rs @@ -594,7 +594,7 @@ impl Encode for FetchObject { } impl Decode for FetchObject { - fn decode(buf: &mut Decoder<'_>, _: Version) -> Result { + fn decode(buf: &mut Decoder<'_>, version: Version) -> Result { let flags = buf.varint()?; // Anything at or above 128 is a named value rather than a set of flags, and only @@ -637,7 +637,11 @@ impl Decode for FetchObject { }; let properties = match flags & flag::PROPERTIES != 0 { - true => Some(buf.bytes()?.to_vec()), + true => { + let super::group::ObjectExtensionsLength(size) = + super::group::ObjectExtensionsLength::decode(buf, version)?; + Some(buf.slice(size)?.to_vec()) + } false => None, }; @@ -1027,6 +1031,22 @@ mod object_tests { (buf.to_vec(), decoded) } + /// A fetch stream is read until a whole object header has arrived, so a properties + /// length the peer declares is refused at its prefix rather than buffered. + #[test] + fn oversized_properties_are_refused_at_the_prefix() { + let mut wire = Vec::new(); + Encoder::new(&mut wire, VERSION.into()) + .varint(flag::PROPERTIES) + .unwrap(); + Encoder::new(&mut wire, VERSION.into()) + .varint((super::super::group::MAX_OBJECT_EXTENSIONS + 1) as u64) + .unwrap(); + + let err = FetchObject::decode_slice(&wire, VERSION).unwrap_err(); + assert!(matches!(err, DecodeError::MessageTooLarge { .. }), "{err:?}"); + } + /// The first Object carries absolute IDs and a priority, because "same as the prior /// Object" has no prior to refer to. Byte-pinned: the flags declare exactly the fields /// that follow, in wire order. diff --git a/rs/moq-net/src/ietf/group.rs b/rs/moq-net/src/ietf/group.rs index 23fd779f1d..dc02dc34a3 100644 --- a/rs/moq-net/src/ietf/group.rs +++ b/rs/moq-net/src/ietf/group.rs @@ -20,6 +20,27 @@ const PROP_TIMESTAMP: u64 = 0x10; /// assigns 0x0A to Secure Objects private properties, so it is not accepted here. const PROP_TIMESTAMP_DRAFT03: u64 = 0x06; +/// Implementation limit for an object's extension (property) block, independent of the +/// IETF draft. Checked at the length prefix, before the block is buffered. +pub(super) const MAX_OBJECT_EXTENSIONS: usize = 64 * 1024; + +/// The length prefix of an object's extension block, refused past [`MAX_OBJECT_EXTENSIONS`]. +#[derive(Debug)] +pub(super) struct ObjectExtensionsLength(pub usize); + +impl Decode for ObjectExtensionsLength { + fn decode(buf: &mut Decoder<'_>, _: Version) -> Result { + let size = usize::try_from(buf.varint()?).map_err(|_| DecodeError::BoundsExceeded)?; + if size > MAX_OBJECT_EXTENSIONS { + return Err(DecodeError::MessageTooLarge { + size, + max: MAX_OBJECT_EXTENSIONS, + }); + } + Ok(Self(size)) + } +} + /// Encode a frame's presentation timestamp as a moq-transport Object Property. /// /// Matches the LOC encoding of the same registry id so a relay or LOC-aware peer diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index 6b93e0915a..bc2ef60538 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -38,6 +38,7 @@ mod track; mod version; use control::Control; +pub(crate) use control::initial_max_request_id; pub use datagram::*; pub use fetch::*; pub use filter::*; diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 044c645fcc..41d0b2f0a2 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -317,7 +317,7 @@ pub(super) struct Publisher { session: S, // Traffic stats are attributed through this tagged origin handle. origin: origin::Consumer, - control: Control, + pub(super) control: Control, // Our own Hop ID, stamped onto every advertisement we forward. Taken from the // origin we consume so it matches the local relay identity across every session, // which is what makes cross-session loop detection work. @@ -334,6 +334,8 @@ pub(super) struct Publisher { version: Version, // Dispatched finite serves, including those not yet polled. pub(super) owed: Arc, + // Subscriptions the peer may hold at once (`session::Limits::subscriptions`). + pub(super) subscriptions: crate::session::Slots, } struct Serve(Arc); @@ -399,6 +401,7 @@ where joins: Default::default(), version, owed: Default::default(), + subscriptions: Default::default(), } } @@ -610,6 +613,16 @@ where .reject_subscribe(stream, request_id, &Error::Unsupported, "range filters not supported") .await; } + // Held for the life of the subscription. A peer past its limits loses the session. + let _slot = match self.subscriptions.acquire() { + Ok(slot) => slot, + Err(err) => { + self.session + .clone() + .close(crate::SessionError::from(&err).to_code(), "too many subscriptions"); + return Err(err); + } + }; // Stats (subscriptions, viewer refcount, groups/frames/bytes) are counted in // the model, through the tagged `origin::Consumer` the broadcast resolves from. @@ -3933,6 +3946,33 @@ mod serve_tests { } } + /// A SUBSCRIBE past the session's cap closes the session with TOO_MANY_REQUESTS. + #[moq_net_sim::test] + async fn subscriptions_past_the_cap_close_the_session() { + for version in [Version::Draft14, Version::Draft16, Version::Draft20] { + let mut h = serve(version); + h.publisher.subscriptions = crate::session::Slots::new(0); + let stream = Stream::open(&mut h.session.clone(), version).await.unwrap(); + let mut body = Vec::new(); + subscribe(Filter::NextObject, None) + .encode_msg(&mut Encoder::new(&mut body, version.into()), version) + .unwrap(); + h.publisher + .clone() + .handle_stream(ietf::Subscribe::ID, ietf::Body(body.into()), stream) + .unwrap_or_else(|e| panic!("{version}: the request was not started: {e}")) + .await; + assert_eq!( + h.log.closes(), + vec![( + crate::SessionError::TooManyRequests.to_code(), + "too many subscriptions".to_string() + )], + "{version}" + ); + } + } + /// Legal requests we don't serve are refused NOT_SUPPORTED one at a time, and the /// session stays open for the next one. #[moq_net_sim::test] diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index b87101968a..89b1ead563 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -27,6 +27,10 @@ pub struct Config { pub request_id_max: Option, + /// What the peer may make this session hold. On drafts 14 to 16 it also sizes the + /// MAX_REQUEST_ID window advertised in our SETUP (see [`ietf::initial_max_request_id`]). + pub limits: crate::session::Limits, + /// Whether we dialed, which sets the request-id parity. pub client: bool, @@ -104,6 +108,7 @@ where mut session, setup, request_id_max, + limits, client, publish, subscribe, @@ -164,6 +169,10 @@ where // nothing, and an empty subscribe origin issues no SUBSCRIBE_NAMESPACE. let publish = publish.unwrap_or_else(|| origin::Producer::empty(Hop::random()).consume()); let subscribe = subscribe.unwrap_or_else(|| origin::Producer::empty(Hop::random())); + let subscriptions = crate::session::Slots::new(limits.subscriptions) + .with_stats(publish.stats(), crate::stats::Cap::Subscriptions); + let announces = + crate::session::Slots::new(limits.announces).with_stats(subscribe.stats(), crate::stats::Cap::Announces); let res = match version { Version::Draft14 | Version::Draft15 | Version::Draft16 => { @@ -172,7 +181,7 @@ where session.close(SessionError::from(&err).to_code(), "setup stream required"); return Err(err); }; - let control = Control::new(request_id_max, client); + let control = Control::new(request_id_max, client).with_window(limits.requests(), client); let adapter = ControlStreamAdapter::new(session.clone(), control.clone(), version); let mut publisher = Publisher::new( @@ -187,8 +196,9 @@ where let (tasks, mut task_set) = TaskSet::new(); publisher.withdrawal = withdrawing.clone(); publisher.owed = serving.clone(); + publisher.subscriptions = subscriptions; - let subscriber = Subscriber::new( + let mut subscriber = Subscriber::new( runtime.clone(), adapter.clone(), subscribe, @@ -201,6 +211,7 @@ where tasks.clone(), goaway.going_away.clone(), ); + subscriber.announces = announces; // GOAWAY send task: draft-14-16 carry GOAWAY on the shared control // stream. Parked on the drain trigger; races the transport close so @@ -355,8 +366,9 @@ where let (tasks, mut task_set) = TaskSet::new(); publisher.withdrawal = withdrawing.clone(); publisher.owed = serving.clone(); + publisher.subscriptions = subscriptions; - let subscriber = Subscriber::new( + let mut subscriber = Subscriber::new( runtime.clone(), session.clone(), subscribe, @@ -369,6 +381,7 @@ where tasks, goaway.going_away.clone(), ); + subscriber.announces = announces; let sub_ns_session = session.clone(); let sub_ns = subscriber.clone(); @@ -994,6 +1007,21 @@ where }; match id { + // Draft-16 moved SUBSCRIBE_NAMESPACE to its own stream, past the control stream + // that admits every other request, but it still takes a request ID from the + // MAX_REQUEST_ID window. Held until the request ends, like the rest. + ietf::SubscribeNamespaceLegacy::ID if version == Version::Draft16 => { + let request_id = RequestId::decode(&mut crate::coding::Decoder::new(&data.0, version.into()), version)?; + let permit = publisher.control.accept(request_id)?; + let task = publisher.handle_stream(id, data, stream)?; + tasks.push( + async move { + let _permit = permit; + task.await + } + .maybe_boxed(), + ); + } // Publisher handles: Subscribe, Fetch, SubscribeNamespace (0x50 modern / // 0x11 legacy), SubscribeTracks, TrackStatus ietf::Subscribe::ID @@ -1137,6 +1165,7 @@ mod tests { session, setup: None, request_id_max: None, + limits: Default::default(), client: true, publish: None, subscribe: Some(origin), @@ -1197,6 +1226,7 @@ mod tests { session, setup: None, request_id_max: None, + limits: Default::default(), client: true, publish: None, subscribe: Some(scoped), @@ -1250,6 +1280,7 @@ mod tests { session, setup: None, request_id_max: None, + limits: Default::default(), client: true, publish: Some(origin.consume()), subscribe: None, @@ -1358,6 +1389,7 @@ mod tests { session, setup: None, request_id_max: None, + limits: Default::default(), client: true, publish: None, subscribe: Some(origin), @@ -1398,6 +1430,7 @@ mod tests { session, setup: None, request_id_max: None, + limits: Default::default(), client: true, publish: None, subscribe: None, @@ -1686,6 +1719,7 @@ mod tests { session, setup: None, request_id_max: None, + limits: Default::default(), client: false, publish: None, subscribe: None, @@ -1789,6 +1823,7 @@ mod tests { session, setup: Some(setup), request_id_max: None, + limits: Default::default(), client: true, publish: None, subscribe: Some(origin), diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 693bf38b91..64b8b355ba 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -13,7 +13,7 @@ use crate::{ util::{MaybeBoxedExt, MaybeSendBox, TaskSet, Tasks}, }; -use super::{Message, Version, cluster, error::request, peer}; +use super::{Message, Version, cluster, error::request, group::ObjectExtensionsLength, peer}; use crate::tail::{Reading, Settle, Tail}; use kio::Lock; @@ -538,6 +538,9 @@ struct BroadcastState { // One minted source per requested path under the namespace, each closed // when its guard drops. sources: HashMap, + + // Counts this namespace against the session's announce cap until it is retracted. + _slot: crate::session::Slot, } /// What one advertisement said, once its parameters are resolved against the session. @@ -581,6 +584,8 @@ pub(super) struct Subscriber { going_away: crate::goaway::GoingAway, // What this session may allocate up front for objects still arriving. frames: frame::Budget, + // Namespaces the peer may have announced at once (`session::Limits::announces`). + pub(super) announces: crate::session::Slots, } /// The prefixes to issue SUBSCRIBE_NAMESPACE for: `origin`'s permitted scope, @@ -661,6 +666,7 @@ where version, going_away, frames: Default::default(), + announces: Default::default(), } } @@ -1531,6 +1537,12 @@ where Ok(()) } Entry::Vacant(entry) => { + // A peer past its limits loses the session, not just this namespace. + let slot = self.announces.acquire().inspect_err(|err| { + self.session + .clone() + .close(crate::SessionError::from(err).to_code(), "too many announcements"); + })?; // Propagates Error::Unauthorized if the namespace is out of scope. let dynamic = self.origin.dynamic(&path, route.clone())?; @@ -1539,6 +1551,7 @@ where dynamic, count: 1, sources: HashMap::new(), + _slot: slot, }); tracing::debug!(route = %self.origin.absolute(&path), "announce"); @@ -2786,25 +2799,6 @@ const END_OF_GROUP: u64 = 0x3; /// Object status: no object at or past this location exists (every implemented draft). const END_OF_TRACK: u64 = 0x4; -// Implementation limit for object extension blocks, independent of the IETF draft. -const MAX_OBJECT_EXTENSIONS: usize = 64 * 1024; - -#[derive(Debug)] -struct ObjectExtensionsLength(usize); - -impl Decode for ObjectExtensionsLength { - fn decode(buf: &mut Decoder<'_>, _: Version) -> Result { - let size = usize::try_from(buf.varint()?).map_err(|_| DecodeError::BoundsExceeded)?; - if size > MAX_OBJECT_EXTENSIONS { - return Err(DecodeError::MessageTooLarge { - size, - max: MAX_OBJECT_EXTENSIONS, - }); - } - Ok(Self(size)) - } -} - /// The start of a subgroup stream's first object, peeked before its group is created. #[derive(Debug, Clone, Copy)] struct FirstObject { @@ -3826,8 +3820,7 @@ async fn decode_fetch_object( let subgroup = stream.varint().await?; let object = stream.varint().await?; let _priority = stream.read_exact(1).await?; - let size = usize::try_from(stream.varint().await?) - .map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded))?; + let ObjectExtensionsLength(size) = stream.decode().await?; let properties = stream.read_exact(size).await?.to_vec(); return Ok(Some(FetchedObject { group: Some(group), @@ -4126,6 +4119,39 @@ mod tests { } } + /// Draft-14 frames a fetch object's properties with a bare length, which is refused at + /// the prefix rather than buffered while the peer trickles the rest in. + #[moq_net_sim::test] + async fn draft14_fetch_properties_are_capped() { + use crate::lite::test_transport::ScriptedSession; + use crate::transport::poll::Session as _; + use futures::FutureExt as _; + + const VERSION: Version = Version::Draft14; + let mut wire = Vec::new(); + for field in [0u64, 0, 0] { + crate::coding::Encoder::new(&mut wire, VERSION.into()) + .varint(field) + .unwrap(); + } + wire.push(0); + crate::coding::Encoder::new(&mut wire, VERSION.into()) + .varint((super::super::group::MAX_OBJECT_EXTENSIONS + 1) as u64) + .unwrap(); + + let mut session = ScriptedSession::new(wire); + let (_, recv) = session.open_bi().await.unwrap(); + let mut reader = Reader::new(recv, VERSION); + let result = decode_fetch_object(&mut reader, VERSION, true) + .now_or_never() + .expect("refused at the prefix, not parked on the body"); + assert!( + matches!(result, Err(Error::Decode(DecodeError::MessageTooLarge { .. }))), + "{:?}", + result.err() + ); + } + fn fin_responses(clean: bool) -> Vec { use crate::coding::Encode; let mut responses = Vec::new(); @@ -5383,6 +5409,57 @@ mod tests { assert!(!table.map.contains_key(&0), "the oldest tombstone is forgotten first"); } + /// A namespace past the session's cap closes the session with TOO_MANY_REQUESTS, a + /// repeat of a held namespace costs nothing, and a retraction frees its slot. + #[moq_net_sim::test] + async fn namespaces_past_the_cap_close_the_session() { + let session = crate::lite::test_transport::SinkSession::new(Default::default()); + let log = session.log.clone(); + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let (tasks, _task_set) = crate::util::TaskSet::new(); + let mut subscriber = Subscriber::new( + crate::time::Clock::sim(), + session, + origin, + Control::new(None, false), + None, + peer::PeerSetup::default(), + crate::Hop::new(1).unwrap(), + None, + Version::Draft14, + tasks, + Default::default(), + ); + subscriber.announces = crate::session::Slots::new(1); + + let advert = |subscriber: &Subscriber<_>| subscriber.route(None, &cluster::Peer::default()).expect("route"); + let a = crate::Path::new("a").to_owned(); + let b = crate::Path::new("b").to_owned(); + let (first, repeat, over, last) = ( + advert(&subscriber), + advert(&subscriber), + advert(&subscriber), + advert(&subscriber), + ); + subscriber.start_announce(a.clone(), first).unwrap(); + subscriber.start_announce(a.clone(), repeat).unwrap(); + assert!(matches!( + subscriber.start_announce(b.clone(), over), + Err(Error::TooManyRequests) + )); + assert_eq!( + log.closes(), + vec![( + crate::SessionError::TooManyRequests.to_code(), + "too many announcements".to_string() + )] + ); + + subscriber.stop_announce(a.clone()).unwrap(); + subscriber.stop_announce(a).unwrap(); + subscriber.start_announce(b, last).unwrap(); + } + /// moq-transport carries no hop ids, so a peer's broadcasts are marked /// anonymous (hop 0). An identity assigned via `Client::with_peer_hop` is /// stored as `via` for split-horizon and never written into the chain. diff --git a/rs/moq-net/src/lite/announce.rs b/rs/moq-net/src/lite/announce.rs index 7bfbf9578a..bc83e631f6 100644 --- a/rs/moq-net/src/lite/announce.rs +++ b/rs/moq-net/src/lite/announce.rs @@ -2,7 +2,10 @@ use num_enum::{IntoPrimitive, TryFromPrimitive}; use crate::{Epoch, Hop, Hops, Path, coding::*, origin::Cost}; -use super::{Message, Version, message::decode_size}; +use super::{ + Message, Version, + message::{MAX_MESSAGE_SIZE, decode_size}, +}; // lite-06 announce message types: an outer discriminator carried before the length // prefix, so each announcement is an independently-typed, length-delimited message @@ -255,6 +258,10 @@ impl Encode for AnnounceBroadcast<'_> { } Self::Ended { .. } | Self::Skipped => unreachable!("refused above"), } + if w.since(&prefix) > MAX_MESSAGE_SIZE { + w.discard(prefix); + return Err(EncodeError::TooLarge); + } return w.fill(prefix); } @@ -282,6 +289,10 @@ impl Encode for AnnounceBroadcast<'_> { return Err(EncodeError::Version); } } + if w.since(&prefix) > MAX_MESSAGE_SIZE { + w.discard(prefix); + return Err(EncodeError::TooLarge); + } w.fill(prefix) } } @@ -291,7 +302,7 @@ impl Decode for AnnounceBroadcast<'_> { if version.has_announce_id() { // Lite06+: outer type, then a size-prefixed body decoded within its bounds. let typ = buf.varint()?; - let size = decode_size(buf)?; + let size = decode_size(buf, MAX_MESSAGE_SIZE)?; let mut body = buf.sub(size)?; let msg = match typ { ANNOUNCE_START => Self::Active { @@ -320,7 +331,7 @@ impl Decode for AnnounceBroadcast<'_> { } // Older versions: a single size-prefixed ANNOUNCE_BROADCAST with an inner status. - let size = decode_size(buf)?; + let size = decode_size(buf, MAX_MESSAGE_SIZE)?; let mut body = buf.sub(size)?; let msg = Self::decode_legacy(&mut body, version)?; if !body.is_empty() { @@ -469,6 +480,9 @@ pub struct AnnounceInit<'a> { } impl Message for AnnounceInit<'_> { + // Published lite01/02 carry the whole initial set in one message. + const MAX_SIZE: usize = 64 * 1024 * 1024; + fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { match version { Version::Lite01 | Version::Lite02 => {} diff --git a/rs/moq-net/src/lite/message.rs b/rs/moq-net/src/lite/message.rs index 13cb0538c1..b38b2a910e 100644 --- a/rs/moq-net/src/lite/message.rs +++ b/rs/moq-net/src/lite/message.rs @@ -2,18 +2,19 @@ use crate::coding::{Decode, DecodeError, Decoder, Encode, EncodeError, Encoder}; use super::Version; -// Match the JavaScript reader's ceiling. Lite control messages are buffered before -// decoding, so the limit must be checked as soon as their length prefix arrives. -pub(super) const MAX_MESSAGE_SIZE: usize = 64 * 1024 * 1024; +// Lite control messages are buffered whole before decoding, so the limit is checked as +// soon as the length prefix arrives. The same ceiling as SETUP: paths, track names, and +// hop chains fit with room to spare, and the JavaScript reader matches it. +pub(super) const MAX_MESSAGE_SIZE: usize = u16::MAX as usize; -/// Read a lite message's varint size prefix, refusing one past [`MAX_MESSAGE_SIZE`]. -pub(super) fn decode_size(r: &mut Decoder<'_>) -> Result { +/// Read a lite message's varint size prefix, refusing one past `max`. +pub(super) fn decode_size(r: &mut Decoder<'_>, max: usize) -> Result { let size = r.varint()?; match usize::try_from(size) { - Ok(size) if size <= MAX_MESSAGE_SIZE => Ok(size), + Ok(size) if size <= max => Ok(size), _ => Err(DecodeError::MessageTooLarge { size: usize::try_from(size).unwrap_or(usize::MAX), - max: MAX_MESSAGE_SIZE, + max, }), } } @@ -48,13 +49,7 @@ impl Encode for T { impl Decode for T { fn decode(r: &mut Decoder<'_>, version: Version) -> Result { - let size = decode_size(r)?; - if size > Self::MAX_SIZE { - return Err(DecodeError::MessageTooLarge { - size, - max: Self::MAX_SIZE, - }); - } + let size = decode_size(r, Self::MAX_SIZE)?; let mut body = r.sub(size)?; // The body is complete, so running short inside it is malformed, not a wait for more. @@ -120,4 +115,36 @@ mod tests { let err = Empty::decode_slice(&wire, Version::Lite06).unwrap_err(); assert!(matches!(err, DecodeError::Short)); } + + /// A peer can no longer make a control stream buffer megabytes: every message past + /// the SETUP ceiling is refused at its length prefix, announcements included. + #[test] + fn control_messages_are_refused_at_the_prefix() { + let oversized = |prefix: &[u8]| { + let mut wire = prefix.to_vec(); + Encoder::new(&mut wire, Version::Lite06.into()).varint(65_536).unwrap(); + wire + }; + + let wire = oversized(&[]); + let err = super::super::Subscribe::decode_slice(&wire, Version::Lite06).unwrap_err(); + assert!(matches!(err, DecodeError::MessageTooLarge { .. }), "{err:?}"); + + // ANNOUNCE_START: the type, then the length. + let wire = oversized(&[0]); + let err = super::super::AnnounceBroadcast::decode_slice(&wire, Version::Lite06).unwrap_err(); + assert!(matches!(err, DecodeError::MessageTooLarge { .. }), "{err:?}"); + } + + /// ANNOUNCE_INIT carries the whole initial set in one message, so it keeps the room + /// a large origin needs. + #[test] + fn announce_init_waits_for_a_large_body() { + let mut wire = Vec::new(); + Encoder::new(&mut wire, Version::Lite02.into()) + .varint(1024 * 1024) + .unwrap(); + let err = super::super::AnnounceInit::decode_slice(&wire, Version::Lite02).unwrap_err(); + assert!(matches!(err, DecodeError::Short), "{err:?}"); + } } diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index b765a7cec8..21125b5b4f 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -39,6 +39,8 @@ pub(super) struct PublisherConfig { /// The origin (hop) id assigned to the peer, used whenever the peer doesn't /// declare one itself. See `Client::with_peer_hop`. pub peer_hop: Option, + /// Subscriptions the peer may hold at once (`session::Limits::subscriptions`). + pub subscriptions: crate::session::Slots, } /// Context shared by every control-stream child. @@ -64,6 +66,7 @@ struct Shared { goaway: crate::goaway::Protocol, // Control streams still serving the peer data, which a draining close waits for. owed: AtomicUsize, + subscriptions: crate::session::Slots, // The send time an untimed track's frames carry, since no lite version can mark them // untimed yet (see `wire_timestamp`). runtime: crate::time::Clock, @@ -170,6 +173,7 @@ impl Publisher { version: config.version, goaway: config.goaway, owed: AtomicUsize::new(0), + subscriptions: config.subscriptions, runtime: config.runtime.clone(), }), runtime: config.runtime, @@ -194,6 +198,7 @@ where shared: self.shared.clone(), runtime: self.runtime.clone(), state: ControlState::Start { stream }, + _slot: None, }); } Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), @@ -238,6 +243,8 @@ struct Control { // Handed to the children that arm timers (PROBE, announce linger). runtime: crate::time::Clock, state: ControlState, + // A subscription's place under the session's cap, held until the stream ends. + _slot: Option, } // A state machine's enum is its storage: one transient instance per stream, so the @@ -301,9 +308,20 @@ impl Control { lite::ControlType::Announce => { ControlState::Announce(AnnounceServe::new(self.shared.clone(), stream)) } - lite::ControlType::Subscribe => { - ControlState::Subscribe(RequestServe::new(self.shared.clone(), stream)) - } + lite::ControlType::Subscribe => match self.shared.subscriptions.acquire() { + Ok(slot) => { + self._slot = Some(slot); + ControlState::Subscribe(RequestServe::new(self.shared.clone(), stream)) + } + // A peer past its limits loses the session, not just this request. + Err(err) => { + self.shared + .session + .clone() + .close(crate::SessionError::from(&err).to_code(), "too many subscriptions"); + return Poll::Ready(Err(err)); + } + }, // The Track Stream and FETCH are lite-05+ only. lite::ControlType::Fetch | lite::ControlType::Track if !self.shared.version.has_track_stream() => @@ -3746,6 +3764,7 @@ mod tests { peer_setup, goaway, peer_hop: Some(assigned), + subscriptions: Default::default(), }); let serving = kio::wait(|waiter| publisher.shared.poll_serving_origin(waiter)).await; @@ -3911,6 +3930,7 @@ mod tests { peer_setup: crate::lite::PeerSetup::default(), goaway, peer_hop: None, + subscriptions: Default::default(), }); ProbeServe::new(publisher.shared.clone(), publisher.runtime.clone(), stream) } @@ -4083,6 +4103,7 @@ mod tests { peer_setup, goaway, peer_hop: None, + subscriptions: Default::default(), }); let mut script = Vec::new(); @@ -4206,6 +4227,7 @@ mod tests { peer_setup, goaway, peer_hop: None, + subscriptions: Default::default(), }); let mut script = Vec::new(); @@ -4260,4 +4282,40 @@ mod tests { }; assert_eq!(*run.track_priority_tx.read(), 7, "the update was lost"); } + + /// A subscription past the session's cap closes the session with TOO_MANY_REQUESTS. + #[moq_net_sim::test] + async fn subscriptions_past_the_cap_close_the_session() { + use crate::coding::Encode as _; + + const VERSION: Version = Version::Lite06; + let mut script = Vec::new(); + lite::ControlType::Subscribe + .encode(&mut crate::coding::Encoder::new(&mut script, VERSION.into()), VERSION) + .unwrap(); + let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(vec![script]); + let log = session.log.clone(); + + let origin = Hop::random().produce(); + let (_, goaway) = crate::goaway::Handle::new(true); + let mut publisher = Publisher::new(PublisherConfig { + runtime: crate::time::Clock::sim(), + session, + origin: origin.consume(), + version: VERSION, + peer_setup: crate::lite::PeerSetup::default(), + goaway, + peer_hop: None, + subscriptions: crate::session::Slots::new(0), + }); + + let _ = publisher.poll(&kio::Waiter::noop()); + assert_eq!( + log.closes(), + vec![( + crate::SessionError::TooManyRequests.to_code(), + "too many subscriptions".to_string() + )] + ); + } } diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 76f098e22a..fdb059dda9 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -171,6 +171,9 @@ pub struct Config { /// gated on the client's path via [`accept_setup`]). Seeds the peer-setup slot so /// the Setup Stream isn't expected again. `None` reads it from the wire as usual. pub peer_setup: Option>, + + /// What the peer may make this session hold. + pub limits: crate::session::Limits, } /// Start a lite session. @@ -190,6 +193,7 @@ where version, mut our_setup, peer_setup, + limits, } = config; let recv_bw = bandwidth::Producer::new(); @@ -225,6 +229,10 @@ where // subscribe origin issues no ANNOUNCE_PLEASE. let publish = publish.unwrap_or_else(|| origin::Producer::empty(Hop::random()).consume()); let subscribe = subscribe.unwrap_or_else(|| origin::Producer::empty(Hop::random())); + let subscriptions = + crate::session::Slots::new(limits.subscriptions).with_stats(publish.stats(), crate::stats::Cap::Subscriptions); + let announces = + crate::session::Slots::new(limits.announces).with_stats(subscribe.stats(), crate::stats::Cap::Announces); // Publisher and Subscriber each derive their identity from their own // attached origin (publish.info / subscribe.info). This is what gets @@ -270,8 +278,9 @@ where peer_setup: peer_setup.clone(), goaway: goaway.clone(), peer_hop, + subscriptions, }); - let subscriber = Subscriber::new(SubscriberConfig { + let mut subscriber = Subscriber::new(SubscriberConfig { runtime: runtime.clone(), session: session.clone(), origin: subscribe, @@ -285,6 +294,7 @@ where cost: our_cost, going_away: goaway.going_away.clone(), }); + subscriber.announces = announces; let driver = Driver { local_close: Default::default(), @@ -687,6 +697,7 @@ mod tests { version, our_setup: Setup::default(), peer_setup: None, + limits: Default::default(), }) .unwrap(); let _ = started.driver.poll(&kio::Waiter::noop()); @@ -725,6 +736,7 @@ mod tests { version, our_setup: Setup::default(), peer_setup: None, + limits: Default::default(), }) .unwrap(); let _ = started.driver.poll(&kio::Waiter::noop()); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index bd9ae5fe5d..70ff330b38 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -83,6 +83,8 @@ pub(super) struct Subscriber { going_away: crate::goaway::GoingAway, /// What this session may allocate up front for frames still arriving. frames: frame::Budget, + /// Broadcasts the peer may have announced at once (`session::Limits::announces`). + pub(super) announces: crate::session::Slots, } #[derive(Clone)] @@ -118,6 +120,7 @@ impl Subscriber { sources: kio::Queue::new(), going_away: config.going_away, frames: Default::default(), + announces: Default::default(), } } @@ -282,8 +285,9 @@ impl Subscriber { } // The peer holds this prefix now. Everything below either accepts the announcement, - // replacing this, or declines it and leaves it exactly as reserved. - announced.reserve(path.clone(), epoch); + // replacing this, or declines it and leaves it exactly as reserved. Past the + // session's cap this refuses the announce stream instead. + announced.reserve(path.clone(), epoch)?; if let Some(responder) = responder_origin { // A chain already naming the sender came back through it: a reflection, and @@ -1247,7 +1251,13 @@ impl AnnouncePrefix { } } + /// Fails the session on any error, [`Error::TooManyRequests`] included: a peer + /// announcing more than the session allows loses the session. fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { + self.poll_run(waiter) + } + + fn poll_run(&mut self, waiter: &kio::Waiter) -> Poll> { let mut cx = waiter.context(); loop { match &mut self.state { @@ -1318,7 +1328,7 @@ impl AnnouncePrefix { let run = PrefixRun { responder_origin, link_cost, - announced: Announced::default(), + announced: Announced::new(self.subscriber.announces.clone()), decoder: lite::AnnounceDecoder::default(), }; @@ -3159,6 +3169,58 @@ mod tests { cursor.assert_next_active("a"); cursor.assert_next_active("b"); } + + /// An announce past the session's cap fails the session with TOO_MANY_REQUESTS, and + /// the stream gives its slots back once the session drops it. + #[moq_net_sim::test] + async fn announces_past_the_cap_close_the_session() { + const VERSION: Version = Version::Lite06; + let start = |suffix| lite::AnnounceBroadcast::Active { + epoch: None, + suffix: lite::PathRef::literal(Path::new(suffix)), + hops: lite::HopsRef::literal(crate::Hops::new()), + cost: crate::origin::Cost::default(), + }; + let mut script = Vec::new(); + lite::AnnounceOk { + origin: crate::Hop::new(9).unwrap(), + active: 3, + } + .encode(&mut crate::coding::Encoder::new(&mut script, VERSION.into()), VERSION) + .unwrap(); + for suffix in ["a", "b", "c"] { + start(suffix) + .encode(&mut crate::coding::Encoder::new(&mut script, VERSION.into()), VERSION) + .unwrap(); + } + + let origin = origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let session = crate::lite::test_transport::ScriptedSession::new(script); + let mut subscriber = Subscriber::new(SubscriberConfig { + runtime: crate::time::Clock::sim(), + session, + origin, + recv_bandwidth: None, + version: VERSION, + peer_setup: Default::default(), + cost: Some(1), + peer_hop: None, + going_away: Default::default(), + }); + let slots = crate::session::Slots::new(2); + subscriber.announces = slots.clone(); + let mut prefix = AnnouncePrefix::new(subscriber, Path::new("").to_owned()); + + let err = kio::wait(|waiter| prefix.poll(waiter)) + .await + .expect_err("an announce past the cap must fail the session"); + assert_eq!(crate::SessionError::from(&err), crate::SessionError::TooManyRequests); + + // The session's end drops the stream, which gives its slots back. + drop(prefix); + let _a = slots.acquire().expect("the failed stream kept a slot"); + let _b = slots.acquire().expect("the failed stream kept a slot"); + } } /// The four wire fields a subscription's half-open range encodes to. @@ -3254,9 +3316,11 @@ enum Sub { /// /// A declined advertisement remains present with no route because the peer still /// owns its path and announce id until it retracts or restarts it. -#[derive(Default)] struct Announced { routes: HashMap>, + /// The session's announce cap, and one slot taken from it per entry in `routes`. + slots: crate::session::Slots, + held: Vec, /// The epoch each advertisement named, kept while it stands even when declined, /// so a restart that attaches it later serves the same broadcast. epochs: HashMap, @@ -3266,7 +3330,24 @@ struct Announced { ready: kio::Queue, } +#[cfg(test)] +impl Default for Announced { + fn default() -> Self { + Self::new(Default::default()) + } +} + impl Announced { + fn new(slots: crate::session::Slots) -> Self { + Self { + routes: HashMap::new(), + slots, + held: Vec::new(), + epochs: HashMap::new(), + ready: Default::default(), + } + } + fn contains(&self, path: &PathOwned) -> bool { self.routes.contains_key(path) } @@ -3297,12 +3378,16 @@ impl Announced { /// the peer still holds. /// Only valid on a prefix the peer does not already hold, which the caller establishes /// with [`Self::contains`]. Overwriting an attached route is [`Self::declined`]'s job. - fn reserve(&mut self, path: PathOwned, epoch: Option) { + /// + /// Fails with [`Error::TooManyRequests`] once the session holds as many as it allows. + fn reserve(&mut self, path: PathOwned, epoch: Option) -> Result<(), Error> { debug_assert!(!self.routes.contains_key(&path), "reserved a prefix already advertised"); + self.held.push(self.slots.acquire()?); if let Some(epoch) = epoch { self.epochs.insert(path.clone(), epoch); } self.routes.insert(path, None); + Ok(()) } /// The epoch the advertisement at `path` named, if any. @@ -3318,7 +3403,11 @@ impl Announced { /// session from the same peer. Dropping its sources closes their requests. fn withdraw(&mut self, path: &PathOwned) { self.epochs.remove(path); - if let Some(Some(entry)) = self.routes.remove(path) { + let Some(entry) = self.routes.remove(path) else { + return; + }; + self.held.pop(); + if let Some(entry) = entry { entry.dynamic.withdrawn(); } } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 2fa46ab68d..168652ecc8 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -1430,6 +1430,11 @@ impl Producer { (producer, driver) } + /// The ingress stats context this handle was tagged with. + pub(crate) fn stats(&self) -> &stats::Session { + &self.stats + } + /// Attach an ingress stats context: broadcasts created through this handle (and /// any handle derived from it) are attributed to `session` on the subscriber /// (ingress) side. Pass [`stats::Session::default`] to opt out. @@ -4024,6 +4029,11 @@ impl Consumer { self.hidden.include } + /// The egress stats context this handle was tagged with. + pub(crate) fn stats(&self) -> &stats::Session { + &self.stats + } + /// Attach an egress stats context: broadcasts handed out through this handle (and /// any handle derived from it) are attributed to `session` on the publisher /// (egress) side. Pass [`stats::Session::default`] to opt out. diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 4f615ed1bf..637d416157 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -20,6 +20,7 @@ pub struct Server { subscribe: Option, stats: stats::Session, versions: Versions, + limits: crate::session::Limits, } impl Server { @@ -58,6 +59,12 @@ impl Server { self.with_publisher(&origin).with_subscriber(origin) } + /// Cap what each client can make its session hold. Defaults to [`session::Limits::default`](crate::session::Limits::default). + pub fn with_limits(mut self, limits: crate::session::Limits) -> Self { + self.limits = limits; + self + } + /// Restrict which protocol versions to accept, in preference order. /// Defaults to every version this crate supports. pub fn with_versions(mut self, versions: Versions) -> Self { @@ -111,6 +118,7 @@ impl Server { let start = lite::start(lite::Config { runtime: runtime.clone(), + limits: self.limits, session: session.clone(), setup_stream: None, publish, @@ -522,6 +530,7 @@ where // for GOAWAY. A server never advertises a path, hence `None`. let (protocol, goaway, setup) = ietf::start(ietf::Config { runtime: runtime.clone(), + limits: server.limits, session: session.clone(), setup: None, request_id_max: None, @@ -591,7 +600,11 @@ where let parameters = match version { Version::Ietf(v) => { let mut parameters = ietf::Parameters::default(); - parameters.set_varint(ietf::ParameterVarInt::MaxRequestId, u32::MAX as u64); + // The client's requests, admitted up to our limits and granted back as they close. + parameters.set_varint( + ietf::ParameterVarInt::MaxRequestId, + ietf::initial_max_request_id(server.limits.requests(), true), + ); parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); ietf::solicit::into_setup(&mut parameters, v); ietf::hidden::into_setup(&mut parameters, v); @@ -613,6 +626,7 @@ where // Pre-lite-05: no Setup Stream, so nothing to advertise or seed. let start = lite::start(lite::Config { runtime: runtime.clone(), + limits: server.limits, session: session.clone(), setup_stream: Some(stream), publish, @@ -634,6 +648,7 @@ where // Draft 14-16: path came in the bidi SETUP, no uni SETUP to hand back. let (protocol, goaway, setup) = ietf::start(ietf::Config { runtime: runtime.clone(), + limits: server.limits, session: session.clone(), setup: Some(stream), request_id_max, diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index 181ce38f60..0d551b5b26 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -643,3 +643,102 @@ impl Drop for Withdrawing { self.0.0.lock().active -= 1; } } + +/// Caps on what one peer can make a session hold at once. +/// +/// A peer that goes past a cap loses the session, closed with `TOO_MANY_REQUESTS`. The +/// stats sessions rows report each root's peaks, so an operator sees how close sessions +/// come before one is closed. +/// +/// On moq-transport drafts 14 to 16 the caps also size the request window, which counts +/// every request (FETCH, TRACK_STATUS and SUBSCRIBE_UPDATE included), so very low caps +/// can starve it. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct Limits { + /// Broadcasts (moq-lite) or namespaces (moq-transport) the peer may have announced to us. + pub announces: usize, + /// Subscriptions the peer may hold on our broadcasts. + pub subscriptions: usize, +} + +impl Default for Limits { + /// Generous enough for a relay mesh carrying a large origin; lower them for untrusted peers. + fn default() -> Self { + Self { + announces: 100_000, + subscriptions: 10_000, + } + } +} + +impl Limits { + /// How many requests a moq-transport peer may hold open at once (drafts 14 to 16). + /// + /// Twice what the caps admit, so a peer within them is never blocked by request IDs + /// still waiting to be granted back. + pub(crate) fn requests(&self) -> u64 { + let caps = self.announces.saturating_add(self.subscriptions) as u64; + caps.saturating_mul(2) + } +} + +/// A count of live entries shared across a session, refused past its cap. +#[derive(Clone)] +pub(crate) struct Slots { + live: Arc, + max: usize, + /// Where the most this session has held is reported. + stats: Option<(crate::stats::Session, crate::stats::Cap)>, +} + +impl Slots { + pub(crate) fn new(max: usize) -> Self { + Self { + live: Default::default(), + max, + stats: None, + } + } + + /// Report how many are held to `stats`, as `cap`'s peak. + pub(crate) fn with_stats(mut self, stats: &crate::stats::Session, cap: crate::stats::Cap) -> Self { + stats.track_held(cap, &self.live); + self.stats = Some((stats.clone(), cap)); + self + } + + /// Take one slot until the returned guard drops, or [`Error::TooManyRequests`] at the cap. + /// + /// The caller closes the session on that error: a peer past its limits loses the session. + pub(crate) fn acquire(&self) -> Result { + use std::sync::atomic::Ordering; + let held = self + .live + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |live| { + (live < self.max).then_some(live + 1) + }) + .map_err(|_| Error::TooManyRequests)? + + 1; + if let Some((stats, cap)) = &self.stats { + stats.hold(*cap, held as u64); + } + Ok(Slot(self.live.clone())) + } +} + +impl Default for Slots { + /// Unlimited, for sessions and tests that configure nothing. + fn default() -> Self { + Self::new(usize::MAX) + } +} + +/// One taken [`Slots`] entry, released on drop. +pub(crate) struct Slot(Arc); + +impl Drop for Slot { + fn drop(&mut self) { + self.0.fetch_sub(1, std::sync::atomic::Ordering::AcqRel); + } +} diff --git a/rs/moq-net/src/setup.rs b/rs/moq-net/src/setup.rs index cb30ff5d2f..f8a12ef862 100644 --- a/rs/moq-net/src/setup.rs +++ b/rs/moq-net/src/setup.rs @@ -11,7 +11,7 @@ use crate::{ }; // SETUP only carries negotiation parameters. Bound it independently of control messages. -pub(crate) const MAX_SETUP_SIZE: usize = 64 * 1024; +pub(crate) const MAX_SETUP_SIZE: usize = u16::MAX as usize; const CLIENT_SETUP: u8 = 0x20; const SERVER_SETUP: u8 = 0x21; diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index fa2b7273d5..e440375360 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -59,6 +59,10 @@ //! Driven by [`Handle::session`] (the [`Session`] context); a //! [`Session::set_tier`] ends the session on the old tier and starts it on //! the new one. +//! * `announces_peak` / `subscriptions_peak` ([`Presence`]): the most +//! announcements, or subscriptions, any one session under the auth root held +//! at once, to compare against its [`crate::session::Limits`]. A peak never +//! goes down; a rollup takes the largest rather than the sum. //! //! Counters are strictly monotonic (only `fetch_add`); a counter going //! backwards across reads means the underlying entry was garbage collected @@ -111,8 +115,8 @@ use std::{ collections::HashMap, fmt, sync::{ - Arc, Mutex, - atomic::{AtomicU64, Ordering}, + Arc, Mutex, Weak, + atomic::{AtomicU64, AtomicUsize, Ordering}, }, }; @@ -264,9 +268,18 @@ impl Content { struct SessionCounters { sessions_started: AtomicU64, sessions_ended: AtomicU64, + announces_peak: AtomicU64, + subscriptions_peak: AtomicU64, } impl SessionCounters { + fn peak(&self, cap: Cap) -> &AtomicU64 { + match cap { + Cap::Announces => &self.announces_peak, + Cap::Subscriptions => &self.subscriptions_peak, + } + } + /// Read the gauge into a [`Presence`]. Ended is loaded with `Acquire` /// before started with `Relaxed`, the same pairing as [`Counters::snapshot`], /// so the readout never shows `ended > started`. @@ -276,6 +289,8 @@ impl SessionCounters { Presence { sessions_started, sessions_ended, + announces_peak: self.announces_peak.load(Ordering::Relaxed), + subscriptions_peak: self.subscriptions_peak.load(Ordering::Relaxed), } } } @@ -507,7 +522,8 @@ impl Traffic { /// Connected-session presence for one slice (an auth root on a tier, or any /// sum of such slices): cumulative connects and disconnects. `sessions_started -/// - sessions_ended` is the current live session count. +/// - sessions_ended` is the current live session count. The peaks show how close +/// one session came to its [`crate::session::Limits`]. /// /// Like [`Traffic`], this is also the wire shape of one entry on a published /// sessions track. Serialize writes both the canonical names and the legacy @@ -520,6 +536,11 @@ pub struct Presence { pub sessions_started: u64, /// Cumulative sessions disconnected. pub sessions_ended: u64, + /// The most broadcasts (moq-transport: namespaces) any one session has held announced + /// to us at once. + pub announces_peak: u64, + /// The most subscriptions any one session has held on our broadcasts at once. + pub subscriptions_peak: u64, } #[derive(Serialize)] @@ -528,6 +549,8 @@ struct PresenceSer { sessions: u64, sessions_ended: u64, sessions_closed: u64, + announces_peak: u64, + subscriptions_peak: u64, } impl From for PresenceSer { @@ -537,6 +560,8 @@ impl From for PresenceSer { sessions: p.sessions_started, sessions_ended: p.sessions_ended, sessions_closed: p.sessions_ended, + announces_peak: p.announces_peak, + subscriptions_peak: p.subscriptions_peak, } } } @@ -548,6 +573,8 @@ struct PresenceDe { sessions: Edge, sessions_ended: Edge, sessions_closed: Edge, + announces_peak: u64, + subscriptions_peak: u64, } impl From for Presence { @@ -555,6 +582,8 @@ impl From for Presence { Self { sessions_started: counter_edge(d.sessions_started, d.sessions), sessions_ended: counter_edge(d.sessions_ended, d.sessions_closed), + announces_peak: d.announces_peak, + subscriptions_peak: d.subscriptions_peak, } } } @@ -572,10 +601,12 @@ impl<'de> Deserialize<'de> for Presence { } impl Presence { - /// Fold another readout into this one. + /// 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_started += other.sessions_started; self.sessions_ended += other.sessions_ended; + self.announces_peak = self.announces_peak.max(other.announces_peak); + self.subscriptions_peak = self.subscriptions_peak.max(other.subscriptions_peak); } /// Sessions currently connected. @@ -1137,6 +1168,13 @@ pub struct Session { inner: Option>, } +/// Which of a session's [`crate::session::Limits`] a [`Session::hold`] counts against. +#[derive(Clone, Copy, Debug)] +pub(crate) enum Cap { + Announces, + Subscriptions, +} + /// The shared state behind a [`Session`]. Its `Drop` (on the last clone) records /// the session as closed. struct SessionInner { @@ -1151,6 +1189,9 @@ struct SessionInner { /// subscription this context opens for a broadcast bumps `broadcasts_started`, the last /// to close bumps `broadcasts_ended` on the same counters, even across a tier change. viewers: Mutex>, + /// What this session holds against each cap right now, so a tier change can seed the + /// new tier's peaks with slots it still holds. + held: Mutex)>>, } /// The tier a [`Session`] records under right now. @@ -1181,6 +1222,7 @@ impl Session { current: Mutex::new(Current { tier, presence }), generation: AtomicU64::new(0), viewers: Mutex::new(HashMap::new()), + held: Mutex::new(Vec::new()), })), } } @@ -1196,6 +1238,17 @@ impl Session { if let Some(presence) = &presence { presence.sessions_started.fetch_add(1, Ordering::Relaxed); } + if let Some(presence) = &presence { + // Slots still held count on the new tier, but not the old tier's history. + let mut held = inner.held.lock().expect("stats session poisoned"); + held.retain(|(cap, live)| { + let Some(live) = live.upgrade() else { return false }; + presence + .peak(*cap) + .fetch_max(live.load(Ordering::Acquire) as u64, Ordering::Relaxed); + true + }); + } if let Some(old) = std::mem::replace(&mut current.presence, presence) { // Release pairs with the readout's Acquire load of `sessions_ended`. old.sessions_ended.fetch_add(1, Ordering::Release); @@ -1204,6 +1257,24 @@ impl Session { inner.generation.fetch_add(1, Ordering::Release); } + /// Count `live` as what this session holds against `cap`, so a tier change carries it. + pub(crate) fn track_held(&self, cap: Cap, live: &Arc) { + let Some(inner) = &self.inner else { return }; + let mut held = inner.held.lock().expect("stats session poisoned"); + // A context reused across reconnects registers again each time: drop the gone ones. + held.retain(|(_, live)| live.strong_count() > 0); + held.push((cap, Arc::downgrade(live))); + } + + /// Record that this session holds `held` of what `cap` limits, raising that peak on its + /// tier and root. + pub(crate) fn hold(&self, cap: Cap, held: u64) { + let Some(inner) = &self.inner else { return }; + let current = inner.current.lock().expect("stats session poisoned"); + let Some(presence) = ¤t.presence else { return }; + presence.peak(cap).fetch_max(held, Ordering::Relaxed); + } + /// Egress (publisher / reads) scope for a broadcast path. The path is the /// absolute broadcast name. pub(crate) fn egress(&self, path: impl AsPath) -> Scope { @@ -1584,6 +1655,66 @@ mod tests { assert_eq!(regional_counters.subscriber.bytes.load(Relaxed), 7); } + /// Each root's sessions row carries the most any one of its sessions held against each + /// cap, which a slot released does not lower and a rollup does not sum. + #[test] + fn sessions_report_their_peaks_against_the_caps() { + use crate::session::Slots; + + let stats = test_stats(); + let tier = stats.tier(Tier::default()); + let (s1, s2) = (tier.session("acme"), tier.session("acme")); + let announces = Slots::new(10).with_stats(&s1, Cap::Announces); + let subscriptions = Slots::new(10).with_stats(&s2, Cap::Subscriptions); + let other = Slots::new(10).with_stats(&s2, Cap::Announces); + + let held: Vec<_> = (0..3).map(|_| announces.acquire().unwrap()).collect(); + drop(held); + let _one = announces.acquire().unwrap(); + let _subs: Vec<_> = (0..5).map(|_| subscriptions.acquire().unwrap()).collect(); + let _two: Vec<_> = (0..2).map(|_| other.acquire().unwrap()).collect(); + + let (_, presence) = stats + .snapshot() + .sessions() + .into_iter() + .find(|(t, _)| *t == Tier::default()) + .expect("sessions row"); + assert_eq!(presence.announces_peak, 3, "the peak of one session, not the sum"); + assert_eq!(presence.subscriptions_peak, 5); + + let json = serde_json::to_string(&presence).unwrap(); + assert_eq!(serde_json::from_str::(&json).unwrap(), presence); + } + + /// A session moving tiers brings the slots it still holds to the new tier's peak, but + /// not the old tier's history. + #[test] + fn a_tier_change_carries_the_slots_still_held() { + use crate::session::Slots; + + let stats = test_stats(); + let session = stats.tier(Tier::default()).session("acme"); + let subscriptions = Slots::new(10).with_stats(&session, Cap::Subscriptions); + let mut held: Vec<_> = (0..5).map(|_| subscriptions.acquire().unwrap()).collect(); + held.truncate(3); + + let gold = Tier::new("gold"); + session.set_tier(gold.clone()); + + let peak = |tier: &Tier| { + let (_, presence) = stats + .snapshot() + .sessions() + .into_iter() + .find(|(t, _)| t == tier) + .expect("sessions row"); + presence.subscriptions_peak + }; + assert_eq!(peak(&Tier::default()), 5); + assert_eq!(peak(&gold), 3, "the new tier sees what is still held"); + } + #[test] fn snapshot_rolls_up_by_tier_role_and_sessions() { let stats = test_stats(); @@ -2233,6 +2364,7 @@ mod tests { Presence { sessions_started: 3, sessions_ended: 1, + ..Default::default() } } diff --git a/rs/moq-net/tests/request_caps.rs b/rs/moq-net/tests/request_caps.rs new file mode 100644 index 0000000000..3415481390 --- /dev/null +++ b/rs/moq-net/tests/request_caps.rs @@ -0,0 +1,167 @@ +//! A server's [`moq_net::session::Limits`] bound what one client can make it hold: a client +//! past a cap loses the session with TOO_MANY_REQUESTS, and on moq-transport drafts 14 to 16 +//! the MAX_REQUEST_ID window they size is granted back as requests close, so a client that +//! keeps opening and closing requests within the caps never stalls. + +mod support; + +use std::time::Duration; + +use moq_net::{Hop, Version, session::Limits}; +use support::harness::{MockConnectOptions, MockPair, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +/// Drafts with MAX_REQUEST_ID, a draft without it, and the lite draft production runs. +const VERSIONS: [&str; 5] = [ + "moq-transport-14", + "moq-transport-15", + "moq-transport-16", + "moq-transport-17", + "moq-lite-06", +]; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +/// A server publishing `bcast` with `tracks`, and a client subscribed to its origin. +struct Setup { + client: moq_net::origin::Producer, + _server: moq_net::origin::Producer, + _broadcast: moq_net::broadcast::Producer, + tracks: Vec, + _pair: MockPair, +} + +async fn setup(version: &str, limits: Limits, tracks: usize) -> Setup { + let version: Version = version.parse().unwrap(); + let server = produce_origin(1); + let client = produce_origin(2); + + let broadcast = server.create_broadcast("bcast").unwrap(); + let tracks = (0..tracks) + .map(|i| broadcast.create_track(format!("t{i}"), None).unwrap()) + .collect(); + broadcast.announce(Default::default()).unwrap(); + + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(server.consume()); + options.client_subscribe = Some(client.clone()); + options.server_limits = Some(limits); + let pair = connect_mock(options).await; + + Setup { + client, + _server: server, + _broadcast: broadcast, + tracks, + _pair: pair, + } +} + +fn limits(announces: usize, subscriptions: usize) -> Limits { + let mut limits = Limits::default(); + limits.announces = announces; + limits.subscriptions = subscriptions; + limits +} + +/// Many more requests than the window admits at once, opened and closed in turn, all +/// succeed: each one that closes is granted back. +#[moq_net_sim::test] +async fn closed_requests_are_granted_back() { + const ROUNDS: usize = 64; + + for version in VERSIONS { + // A window of 8 requests, far fewer than the rounds below. The cap leaves room + // for an UNSUBSCRIBE still in flight when the next SUBSCRIBE lands. + let setup = setup(version, limits(0, 4), ROUNDS).await; + let broadcast = moq_net_sim::timeout(TIMEOUT, setup.client.consume().routed_broadcast("bcast")) + .await + .expect("announce timeout") + .unwrap(); + + for round in 0..ROUNDS { + let track = broadcast.track(&format!("t{round}")).unwrap(); + let subscriber = moq_net_sim::timeout(TIMEOUT, track.subscribe(None)) + .await + .unwrap_or_else(|_| panic!("{version}: round {round} stalled")) + .unwrap_or_else(|err| panic!("{version}: round {round} refused: {err}")); + drop(subscriber); + drop(track); + + // Wait for the server to see it end, so the next round is a fresh request. + let published = &setup.tracks[round]; + moq_net_sim::timeout(TIMEOUT, published.demand().unused()) + .await + .unwrap_or_else(|_| panic!("{version}: round {round} never ended")) + .unwrap(); + } + } +} + +/// A subscription past the cap closes the session with TOO_MANY_REQUESTS. +#[moq_net_sim::test] +async fn a_subscription_past_the_cap_closes_the_session() { + for version in VERSIONS { + let setup = setup(version, limits(1, 1), 2).await; + let broadcast = moq_net_sim::timeout(TIMEOUT, setup.client.consume().routed_broadcast("bcast")) + .await + .expect("announce timeout") + .unwrap(); + + let _held = moq_net_sim::timeout(TIMEOUT, broadcast.track("t0").unwrap().subscribe(None)) + .await + .expect("subscribe timeout") + .unwrap_or_else(|err| panic!("{version}: the first subscription was refused: {err}")); + + // moq-lite hands the subscriber over once TRACK_INFO answers, so the close + // arrives on the first read instead. + let refused = moq_net_sim::timeout(TIMEOUT, async { + let mut subscriber = broadcast.track("t1").unwrap().subscribe(None).await?; + subscriber.recv_group().await.map(|_| ()) + }) + .await + .expect("subscribe timeout"); + assert!(refused.is_err(), "{version}: a second subscription was admitted"); + + let err = moq_net_sim::timeout(TIMEOUT, setup._pair.client.closed()) + .await + .unwrap_or_else(|_| panic!("{version}: the session stayed open")); + assert!( + matches!(err, moq_net::Error::Session(moq_net::SessionError::TooManyRequests)), + "{version}: closed with {err}" + ); + } +} + +/// A broadcast announced past the cap closes the session with TOO_MANY_REQUESTS. +#[moq_net_sim::test] +async fn an_announce_past_the_cap_closes_the_session() { + for version in VERSIONS { + let version: Version = version.parse().unwrap(); + let client = produce_origin(1); + let server = produce_origin(2); + let a = client.create_broadcast("a").unwrap(); + let b = client.create_broadcast("b").unwrap(); + a.announce(Default::default()).unwrap(); + b.announce(Default::default()).unwrap(); + + let mut options = MockConnectOptions::new(version); + options.client_publish = Some(client.consume()); + options.server_subscribe = Some(server.clone()); + options.server_limits = Some(limits(1, 1)); + let pair = connect_mock(options).await; + + let err = moq_net_sim::timeout(TIMEOUT, pair.client.closed()) + .await + .unwrap_or_else(|_| panic!("{version}: the session stayed open")); + assert!( + matches!(err, moq_net::Error::Session(moq_net::SessionError::TooManyRequests)), + "{version}: closed with {err}" + ); + } +} diff --git a/rs/moq-net/tests/support/harness.rs b/rs/moq-net/tests/support/harness.rs index a45f6a93b2..59877a1cf7 100644 --- a/rs/moq-net/tests/support/harness.rs +++ b/rs/moq-net/tests/support/harness.rs @@ -44,6 +44,8 @@ pub struct MockConnectOptions { pub server_subscribe: Option, /// One-way delay for stream data in each direction. pub latency: std::time::Duration, + /// What the client may make the server's session hold, when not the default. + pub server_limits: Option, } impl MockConnectOptions { @@ -56,6 +58,7 @@ impl MockConnectOptions { server_publish: None, server_subscribe: None, latency: std::time::Duration::ZERO, + server_limits: None, } } } @@ -101,6 +104,9 @@ pub async fn connect_mock(opts: MockConnectOptions) -> MockPair { if let Some(subscribe) = opts.server_subscribe { server = server.with_subscriber(subscribe); } + if let Some(limits) = opts.server_limits { + server = server.with_limits(limits); + } // Run both handshakes concurrently and spawn each side's driver // the moment its handshake resolves: on draft-17+ the server's accept blocks diff --git a/test/drill/mutations/subscriber-leaks-broadcasts.patch b/test/drill/mutations/subscriber-leaks-broadcasts.patch index 328b418eb7..ce36fd64da 100644 --- a/test/drill/mutations/subscriber-leaks-broadcasts.patch +++ b/test/drill/mutations/subscriber-leaks-broadcasts.patch @@ -20,10 +20,11 @@ index 8a0259fec..92febbcdf 100644 // Lite06+: announce ids. Each received `active` implicitly assigns the next // per-stream ordinal; `ended`/`restart` reference it instead of repeating the // path, and lite-07 bases name it too. Tracked even for announces we drop -@@ -1233,5 +1234,5 @@ impl AnnouncePrefix { +@@ -1373,6 +1374,6 @@ impl AnnouncePrefix { + let run = PrefixRun { responder_origin, link_cost, -- announced: Announced::default(), -+ announced: std::mem::ManuallyDrop::new(Announced::default()), +- announced: Announced::new(self.subscriber.announces.clone()), ++ announced: std::mem::ManuallyDrop::new(Announced::new(self.subscriber.announces.clone())), decoder: lite::AnnounceDecoder::default(), - }; + landing,