diff --git a/doc/concept/standard.md b/doc/concept/standard.md index cc5d798caa..46a1f5d1ce 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -30,7 +30,7 @@ subset drops. What a peer actually observes against this implementation: - **Timing.** A track whose `SUBSCRIBE_OK` declares no `TIMESCALE` is untimed, as is every track on drafts 14 through 16. A Rust reader that never subscribed takes the units from `TRACK_STATUS` on draft 18 and later. - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. On draft 17 and later, a `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session. Message parameters follow the negotiated draft's lists: drafts 14 and 15 ignore an unknown or misplaced parameter, draft 16 ignores one defined only for another message but closes the session on an unknown one, and draft 17 and later close it on both. Other malformed control input closes the session with the draft's code too, such as a server `SETUP` carrying `PATH` or `AUTHORITY`, a request cut short by FIN, or a second `GOAWAY` on one stream; a first `GOAWAY` on a request stream is legal. - **One `SETUP` credential**, forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second `SETUP` token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. Where the Rust side authorizes the peer or verifies tokens, a token on a `SUBSCRIBE` or `PUBLISH_NAMESPACE` is verified like an in-band session token and authorizes that one request alone, even where that authorization already covers it; it never widens the session. It is refused with `NOT_SUPPORTED` when nothing verifies it, when it is empty, or when its Token Type is anything but 0 (out of band). A side that does neither, JavaScript included, ignores the token. Rust refuses a request carrying more than one distinct token with `NOT_SUPPORTED`, since it verifies one; JavaScript, which verifies none, ignores them all. A token-authorized `SUBSCRIBE` still stays within the grant the peer gave us. A token-authorized `PUBLISH_NAMESPACE` needs a grant covering everything beneath the namespace, and the tracks we subscribe under it rely on that announce instead of our limit, ending with it. An update renews the token, except on a drafts 14 and 15 `PUBLISH_NAMESPACE`, which has no update, and a refused renewal ends only that request. Rust and JavaScript send the token a route names (`Route::auth`) on its `PUBLISH_NAMESPACE`, including the one drafts 14 and 15 answer a solicited `SUBSCRIBE_NAMESPACE` with, and swapping it renews the request with `REQUEST_UPDATE` from draft 17, or ends and re-sends it before; the inline `NAMESPACE` that answers a solicited `SUBSCRIBE_NAMESPACE` from draft 16 carries no token. The token belongs to the session that added it, so a reconnect means adding it again and re-announcing. A token on any other request is decoded and grants nothing, and a request repeating one is refused with `MALFORMED_AUTH_TOKEN`. `SUBSCRIBE_TRACKS` reads its namespace before refusal; its token is not decoded. On draft 17 and later, the negotiated [MoQ Auth extension](/concept/moq-lite#authorization) accepts additional session tokens in band. -- **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `FORWARD=0`, `SUBSCRIBE_TRACKS`, and the fetch forms above get `NOT_SUPPORTED`. Range Filters get `INVALID_FILTER` on draft 19 and later because the advertised filter limit is zero. On draft 21 and later, namespaces whose first tuple field is exactly `.` or `.session` get `DOES_NOT_EXIST` before application lookup; other dot prefixes still reach the application. Subscription updates accept priority and delivery timeouts; Rust also renews request tokens and judges `NEW_GROUP_REQUEST` as a join (see dynamic groups below). Other supported-but-unimplemented preferences end that subscription; a `TRACK_PROPERTY_FILTER` there closes the session with `PROTOCOL_VIOLATION`, since it belongs only to `SUBSCRIBE_TRACKS`. On draft 18 and later a namespace update may change its prefix; suffixes after the acknowledgement are relative to that new prefix. +- **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `FORWARD=0`, `SUBSCRIBE_TRACKS`, and the fetch forms above get `NOT_SUPPORTED`. Range Filters get `INVALID_FILTER` on draft 19 and later because the advertised filter limit is zero. On draft 21 and later, namespaces whose first tuple field is exactly `.` or `.session` get `DOES_NOT_EXIST` before application lookup; other dot prefixes still reach the application. Subscription updates accept priority, negotiated subscriber max age, and delivery timeouts; Rust also renews request tokens and judges `NEW_GROUP_REQUEST` as a join (see dynamic groups below). Other supported-but-unimplemented preferences end that subscription; a `TRACK_PROPERTY_FILTER` there closes the session with `PROTOCOL_VIOLATION`, since it belongs only to `SUBSCRIBE_TRACKS`. On draft 18 and later a namespace update may change its prefix; suffixes after the acknowledgement are relative to that new prefix. - **Objects and datagrams.** Only subgroup 0 is accepted: a stream for any other subgroup, whether its header gives the ID or takes it from the first Object ID, is stopped and the session stays up. Rust relays preserve Object IDs, gaps, object properties, FIRST\_OBJECT, and END\_OF\_GROUP status objects between moq-transport peers for subgroup 0. Datagrams retain their IDs, properties, priority, and status. Plain Rust readers and moq-lite peers see consecutive frames and normal datagram payloads; JavaScript supports only normal datagrams at Object ID 0. A malformed datagram closes the session; a relayed one the downstream draft cannot carry is dropped. A relayed Timestamp leaves in the units the downstream response declares, and is dropped where it declares none (drafts 14 through 16), as for a plain frame; an immutable one that would change or vanish is refused. Only property 0x10 is read as a Timestamp. Property 0x06 stays opaque and keeps its value regardless of timestamp units; it is dropped across the draft-19 boundary, or refused when immutable, because draft 19 assigns it to SUBGROUP\_DELIVERY\_TIMEOUT. From draft 15, a status object carrying properties is refused, and one relayed from draft 14 leaves without them. Draft 14 allows them, but an END\_OF\_TRACK carrying properties ends the subscription as unsupported, since a track end has nowhere to keep them. A `FETCH` carries the END\_OF\_GROUP object on drafts 14 and 15; later drafts have no FETCH status, so their response ends at the last frame. A group cannot resume across route changes by frame ordinal once Object IDs skip ahead of the ordinals on either route; recovery refuses it instead of splicing different objects. - **Dynamic groups.** A track advertises `DYNAMIC_GROUPS` when its publisher sets a refresh ratio, and a track received with `DYNAMIC_GROUPS` gets a ratio of 0. A `NEW_GROUP_REQUEST` (draft 15 and later) counts as one more subscriber joining, which a `SUBSCRIBE` already is: it asks for a new group only when the newest group's bytes after its first frame exceed the ratio times the first frame. Joins coalesce until the asked-for group arrives. A Rust relay sends `NEW_GROUP_REQUEST` upstream, on the `SUBSCRIBE` or in a `REQUEST_UPDATE`, once a join wants a group past the newest it holds; a relay holding nothing yet asks for none. On a track without a ratio, an update carrying one ends the subscription with `UPDATE_FAILED`. On drafts 15 and 16, whose `TRACK_STATUS` answer has no room for it, a relay copy first learned from `TRACK_STATUS` (a fetch-only reader) does not advertise it to later subscribers. moq-lite hops do not carry requests yet. - **Priority.** Higher is served first. The IETF default of 128 is this stack's 127, and a track that never sets one is 127. A relay's upstream subscription carries the highest priority its subscribers ask for and updates it as they come and go; its group order stays as the `SUBSCRIBE` asked, since no draft lets an update change it. @@ -101,10 +101,14 @@ choices. They follow the product model. namespace prefixes are coalesced internally, so `DUPLICATE_SUBSCRIPTION` and `PREFIX_OVERLAP` are never sent. -- **Delivery timeouts are advisory.** `OBJECT_DELIVERY_TIMEOUT` and - `SUBGROUP_DELIVERY_TIMEOUT` are accepted on subscriptions and their updates, - but do not drop objects or reset subgroups. Latency enforcement uses - presentation time instead of wall-clock delivery deadlines. +- **Subscriber max age is negotiated.** The [Timestamp extension](/draft/moq-timestamp) + carries a subscriber's `max_delay` in `SUBSCRIBE` and `REQUEST_UPDATE` only + when the peer advertises support. Superseded groups expire against media time + and the hop's clock; the newest group, including a quiet catalog, survives any + budget. Relays forward their most tolerant subscriber's budget. Without the + parameter, the publisher preserves its retained backlog. `OBJECT_DELIVERY_TIMEOUT` + and `SUBGROUP_DELIVERY_TIMEOUT` are accepted and ignored: their per-object and + post-FIN deadlines would also cut the newest group, so neither is a max-age budget. - **A relay judges `NEW_GROUP_REQUEST` instead of forwarding it.** [Draft 22 ยง9.20.19](https://www.ietf.org/archive/id/draft-ietf-moq-transport-22.html#section-9.20.19) diff --git a/drafts/draft-lcurley-moq-timestamp.md b/drafts/draft-lcurley-moq-timestamp.md index 2dfdb530d7..39fb4f3862 100644 --- a/drafts/draft-lcurley-moq-timestamp.md +++ b/drafts/draft-lcurley-moq-timestamp.md @@ -27,7 +27,7 @@ informative: This document specifies the transport-level use of the TIMESTAMP and TIMESCALE properties registered by {{loc}}, independent of the LOC container itself. A track-level Timescale property establishes the units, and an object-level Timestamp property carries the presentation time of each object. Exposing media time to the transport lets relays make consistent age-based decisions (e.g. dropping stale objects) without parsing the media container, and it remains consistent across hops regardless of buffering or jitter. -No new code points are requested: an endpoint implementing this document is on the wire indistinguishable from a LOC endpoint that carries only these two properties. +A negotiated subscriber max age bounds delivery of superseded groups without expiring the newest group. --- note_Note_to_Readers @@ -123,16 +123,55 @@ A receiver MUST NOT substitute a time for an untimed object, such as its arrival Arrival time differs at every hop and changes when a route fails over, so a substituted time would not be the same timeline downstream. Age-based dropping ({{age}}) does not apply to an untimed object. -A relay MAY instead drop a group of an untimed track once a later group arrived at it longer ago than the threshold, measured on its own clock. +A relay MAY instead drop a group of an untimed track once its immediate successor arrived at it at least the threshold ago, measured on its own clock ({{max-age}}). ## Age-Based Dropping {#age} Given two objects on the same track, both with TIMESTAMP, a relay computes their relative age as the difference of their timestamps divided by the Timescale. -A relay serving a live subscription MAY drop an object whose age relative to the most recent object on the track exceeds a locally configured or application-supplied threshold, resetting the corresponding stream per {{moqt}}. -This decision is identical at every hop because it depends only on values embedded in the objects, not on arrival time. +A relay serving a live subscription MAY drop a group once a newer group supersedes it and either of these bounds reaches the subscriber's max age ({{max-age}}): + +- Its reach, bounded by the first timestamp of its immediate successor, trails the live media edge by at least the max age. +- Its immediate successor arrived at this hop at least the max age ago, measured on the hop's clock. + +The newest group has no successor and MUST NOT expire under either bound, including when the max age is zero. +Media time permits dropping an old backlog immediately; the hop's clock also bounds untimed groups and a stalled media timeline. A relay MUST NOT use timestamps to reorder delivery beyond what {{moqt}} already permits; this property informs *dropping*, not transmission order. +# Subscriber Max Age {#max-age} + +An endpoint declares that it understands subscriber max age with this Setup Option ({{moqt}} Section 10.3): + +~~~ +MAX_AGE Setup Option { + Type (vi64) = 0x915C4 + Value (vi64) = 1 +} +~~~ + +The value is reserved and MUST be ignored on receipt. +An endpoint MUST NOT send the MAX_AGE parameter to a peer that did not declare this option. +The Timestamp and Timescale properties themselves still require no negotiation. + +~~~ +MAX_AGE Parameter { + Type (vi64) = 0x915C6 + Value (vi64) ; milliseconds +} +~~~ + +MAX_AGE is a subscriber parameter on SUBSCRIBE and REQUEST_UPDATE (SUBSCRIBE_UPDATE on older drafts). +Its value MUST be at most `2^53 - 1` milliseconds; a larger value is malformed. +Zero expires superseded groups immediately, but never the newest group. +On SUBSCRIBE, absence leaves the publisher's existing delivery policy unchanged. +On an update, absence leaves the subscription's budget unchanged; presence replaces it. +A publisher receiving MAX_AGE MUST apply the budget as specified in {{age}} and MUST NOT translate it into OBJECT_DELIVERY_TIMEOUT or SUBGROUP_DELIVERY_TIMEOUT. +Those timeouts can expire the newest group and do not express this policy. + +A relay combining subscribers MUST request at least the most tolerant subscriber's max age upstream, bounded by the publisher's retention policy. +One subscriber's shorter budget MUST NOT tighten another subscriber's budget. + + # Security Considerations Timestamps expose the media timeline to relays, which is the point of the extension, but a relay still treats payloads as opaque and gains no access to media content. @@ -143,7 +182,15 @@ Because age-based dropping only affects which objects a live subscription receiv # IANA Considerations -This document requests no registrations. +This document requests these registrations in the "MOQ Setup Options" and "MOQ Message Parameters" registries, respectively: + +| Value | Name | Reference | +|:------|:-----|:----------| +| 0x915C4 | MAX_AGE | This Document | + +| Value | Name | Messages | Reference | +|:------|:-----|:---------|:----------| +| 0x915C6 | MAX_AGE | SUBSCRIBE, REQUEST_UPDATE | This Document | Both properties it uses are already registered by {{loc}} in the "MOQ Properties" registry ({{moqt}} Section 15.8), and this document changes neither their code points nor their value encodings: @@ -165,6 +212,7 @@ An earlier version of this document requested its own code points (`0x915C0` and ## moq-timestamp-02 {:numbered="false"} +- Add a negotiated subscriber MAX_AGE parameter for SUBSCRIBE and REQUEST_UPDATE, preserving the newest group and aggregating the most tolerant budget. - Read TIMESTAMP only from Type `0x10`, independently of the moq-transport version. - Replaced the wall-clock arrival-time fallback with untimed tracks: a track without TIMESCALE has no media time, and a relay forwards it untimed. A relay MAY drop one of its groups once a later group arrived longer ago than the threshold. - On a track with TIMESCALE, a publisher MUST attach TIMESTAMP to every object with a payload, and a receiver treats one without it as malformed. diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index eb9fe2240e..caabd809fd 100644 --- a/js/net/src/connection/accept.ts +++ b/js/net/src/connection/accept.ts @@ -118,26 +118,15 @@ async function acceptAlpn( version: Ietf.IetfVersion, wiring: SessionProps, ): Promise { - const { control, early, solicit, hidden, activeCount, cluster, auth, epoch } = await exchangeSetup( - transport, - version, - "moq-lite-js", - wiring, - ); + // Spread whole, so every negotiated flag reaches the connection. + const setup = await exchangeSetup(transport, version, "moq-lite-js", wiring); return new Ietf.Connection({ ...wiring, client: false, url, quic: transport, - control, - early, - solicit, - hidden, - activeCount, - cluster, - auth, - epoch, + ...setup, // v17+ uses NativeSession which manages its own request IDs; maxRequestId is unused. maxRequestId: 0n, version, @@ -173,6 +162,7 @@ async function acceptSetup( params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Extensions.intoSetup(params, wiring.extensions, version); Ietf.hiddenIntoSetup(params); + Ietf.maxAgeIntoSetup(params); Ietf.activeCountIntoSetup(params, version); const server = new Ietf.ServerSetup({ version, parameters: params }); @@ -190,6 +180,7 @@ async function acceptSetup( version, solicit: Ietf.solicitFromSetup(client.parameters), hidden: Ietf.hiddenFromSetup(client.parameters), + maxAge: Ietf.maxAgeFromSetup(client.parameters), activeCount: Ietf.activeCountFromSetup(client.parameters, version), }); } @@ -235,6 +226,7 @@ async function acceptNegotiated( params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Extensions.intoSetup(params, wiring.extensions, setupVersion); Ietf.hiddenIntoSetup(params); + Ietf.maxAgeIntoSetup(params); Ietf.activeCountIntoSetup(params, setupVersion); const server = new Ietf.ServerSetup({ version: selectedVersion, parameters: params }); @@ -260,6 +252,7 @@ async function acceptNegotiated( version: selectedVersion as Ietf.IetfVersion, solicit: Ietf.solicitFromSetup(client.parameters), hidden: Ietf.hiddenFromSetup(client.parameters), + maxAge: Ietf.maxAgeFromSetup(client.parameters), activeCount: Ietf.activeCountFromSetup(client.parameters, selectedVersion as Ietf.IetfVersion), }); } else { diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index b3924d532b..7d537f1002 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -320,6 +320,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps): params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Extensions.intoSetup(params, wiring.extensions, setupVersion); Ietf.hiddenIntoSetup(params); + Ietf.maxAgeIntoSetup(params); Ietf.activeCountIntoSetup(params, setupVersion); const client = new Ietf.ClientSetup({ @@ -360,6 +361,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps): version: server.version as Ietf.IetfVersion, solicit: Ietf.solicitFromSetup(server.parameters), hidden: Ietf.hiddenFromSetup(server.parameters), + maxAge: Ietf.maxAgeFromSetup(server.parameters), activeCount: Ietf.activeCountFromSetup(server.parameters, server.version as Ietf.IetfVersion), }); } else { @@ -377,26 +379,15 @@ async function handshakeAlpn( version: Ietf.IetfVersion, wiring: SessionProps, ): Promise { - const { control, early, solicit, hidden, activeCount, cluster, auth, epoch } = await exchangeSetup( - session, - version, - "moq-lite-js", - wiring, - ); + // Spread whole, so every negotiated flag reaches the connection. + const setup = await exchangeSetup(session, version, "moq-lite-js", wiring); return new Ietf.Connection({ ...wiring, client: true, url, quic: session, - control, - early, - solicit, - hidden, - activeCount, - cluster, - auth, - epoch, + ...setup, // v17+ uses NativeSession which manages its own request IDs; maxRequestId is unused. maxRequestId: 0n, version, diff --git a/js/net/src/connection/handshake.ts b/js/net/src/connection/handshake.ts index 08a1f6628b..1107f2fb91 100644 --- a/js/net/src/connection/handshake.ts +++ b/js/net/src/connection/handshake.ts @@ -29,6 +29,7 @@ export async function exchangeSetup( early: Reader[]; solicit: boolean | undefined; hidden: boolean; + maxAge: boolean; activeCount: boolean; cluster: Ietf.Cluster.Hops; auth: boolean; @@ -39,6 +40,7 @@ export async function exchangeSetup( params.setBytes(Ietf.SetupOption.Implementation, encoder.encode(implementation)); Extensions.intoSetup(params, extensions, version); Ietf.hiddenIntoSetup(params); + Ietf.maxAgeIntoSetup(params); Ietf.activeCountIntoSetup(params, version); // One id per session, like the moq-lite connection: nothing in this process forwards @@ -67,6 +69,7 @@ export async function exchangeSetup( early: received.early, solicit: received.solicit, hidden: received.hidden, + maxAge: received.maxAge, activeCount: received.activeCount, cluster: { self, peer: received.cluster }, // Negotiated only when both sides offer it. @@ -99,6 +102,7 @@ async function receiveSetup( early: Reader[]; solicit: boolean | undefined; hidden: boolean; + maxAge: boolean; activeCount: boolean; cluster: Hop | undefined; auth: boolean; @@ -154,6 +158,7 @@ async function receiveSetup( early, solicit: Ietf.solicitFromSetup(setup.parameters), hidden: Ietf.hiddenFromSetup(setup.parameters), + maxAge: Ietf.maxAgeFromSetup(setup.parameters), activeCount: Ietf.activeCountFromSetup(setup.parameters, version), cluster: Ietf.Cluster.fromSetup(setup.parameters, version), auth: Ietf.Auth.fromSetup(setup.parameters, version) === true, diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 3600400878..4cbb5e6000 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -126,6 +126,7 @@ export class Connection implements Established { publish, solicit, hidden = false, + maxAge = false, activeCount = false, cluster, auth = false, @@ -150,6 +151,8 @@ export class Connection implements Established { solicit?: boolean; /** Whether the peer understands the HIDDEN parameter (MoQ Hidden). */ hidden?: boolean; + /** Whether the peer accepts subscriber max age. */ + maxAge?: boolean; /** Whether both SETUPs declared MoQ Active Count. */ activeCount?: boolean; /** @@ -231,6 +234,7 @@ export class Connection implements Established { this.#cluster = cluster; this.#epoch = epoch; this.#subscriber = new Subscriber({ + maxAge, session: this.#session, quic, cluster, diff --git a/js/net/src/ietf/index.ts b/js/net/src/ietf/index.ts index 14ad7738bd..653d77e10e 100644 --- a/js/net/src/ietf/index.ts +++ b/js/net/src/ietf/index.ts @@ -8,6 +8,7 @@ export * as Epoch from "./epoch.ts"; export * from "./fetch.ts"; export * from "./goaway.ts"; export * from "./hidden.ts"; +export * from "./max_age.ts"; export * from "./object.ts"; export * from "./parameters.ts"; export * from "./publish.ts"; diff --git a/js/net/src/ietf/max-age.test.ts b/js/net/src/ietf/max-age.test.ts index e0fa269c2e..0d16e303d4 100644 --- a/js/net/src/ietf/max-age.test.ts +++ b/js/net/src/ietf/max-age.test.ts @@ -1,6 +1,9 @@ import { expect, test } from "bun:test"; +import * as Path from "../path.ts"; import { Reader, Writer } from "../stream.ts"; -import { SubscribeOk } from "./subscribe.ts"; +import { maxAgeFromSetup, maxAgeIntoSetup } from "./max_age.ts"; +import { Parameters, SetupOptions } from "./parameters.ts"; +import { Subscribe, SubscribeOk, SubscribeUpdate } from "./subscribe.ts"; import { type IetfVersion, Version } from "./version.ts"; async function bytes(version: IetfVersion, encode: (writer: Writer) => Promise) { @@ -59,3 +62,42 @@ for (const version of Object.values(Version).filter( } }); } + +for (const version of Object.values(Version).filter((v): v is IetfVersion => v !== Version.DRAFT_07)) { + test(`subscriber max age round trips on ${version}`, async () => { + for (const maxAge of [undefined, 0, 250, Number.MAX_SAFE_INTEGER]) { + const msg = new Subscribe({ + requestId: 0n, + trackNamespace: Path.from("live"), + trackName: "catalog", + subscriberPriority: 128, + maxAge, + }); + const encoded = await bytes(version, (w) => msg.encode(w, version)); + expect((await Subscribe.decode(new Reader(undefined, encoded, version), version)).maxAge).toBe(maxAge); + const parameters = new Parameters(); + parameters.maxAge = maxAge; + const legacy = version <= Version.DRAFT_16; + const update = new SubscribeUpdate({ requestId: 2n, ownRequestId: legacy ? 4n : undefined, parameters }); + const encodedUpdate = await bytes(version, (w) => update.encode(w, version)); + expect( + (await SubscribeUpdate.decode(new Reader(undefined, encodedUpdate, version), version)).parameters + .maxAge, + ).toBe(maxAge); + } + }); +} + +test("subscriber max age has its own Setup Option and refuses unsafe budgets", () => { + const params = new SetupOptions(); + expect(maxAgeFromSetup(params)).toBeFalse(); + maxAgeIntoSetup(params); + expect(maxAgeFromSetup(params)).toBeTrue(); + const budget = new Parameters(); + for (const value of [-1, 0.5, Number.MAX_SAFE_INTEGER + 1]) + expect(() => { + budget.maxAge = value; + }).toThrow(); + budget.vars.set(0x915c6n, 1n << 53n); + expect(() => budget.maxAge).toThrow(); +}); diff --git a/js/net/src/ietf/max_age.ts b/js/net/src/ietf/max_age.ts new file mode 100644 index 0000000000..5601c0a67d --- /dev/null +++ b/js/net/src/ietf/max_age.ts @@ -0,0 +1,17 @@ +/** Subscriber age budgets from the MoQ Timestamp extension. @module @internal */ +import { SetupOption, type SetupOptions } from "./parameters.ts"; + +/** The negotiated max-age Setup Option. @internal */ +export const MAX_AGE_OPTION = SetupOption.MaxAge; +/** The subscriber max-age message parameter. @internal */ +export const MAX_AGE_PARAMETER = 0x915c6n; + +/** Whether the peer accepts subscriber max age. @internal */ +export function maxAgeFromSetup(params: SetupOptions): boolean { + return params.getVarint(MAX_AGE_OPTION) !== undefined; +} + +/** Declare support for subscriber max age. @internal */ +export function maxAgeIntoSetup(params: SetupOptions): void { + params.setVarint(MAX_AGE_OPTION, 1n); +} diff --git a/js/net/src/ietf/parameters.ts b/js/net/src/ietf/parameters.ts index 5ace084454..293e6ed1f4 100644 --- a/js/net/src/ietf/parameters.ts +++ b/js/net/src/ietf/parameters.ts @@ -24,6 +24,8 @@ export const SetupOption = { Solicit: 0x40b5an, /** HIDDEN, from the MoQ Hidden extension. See `hidden.ts`. */ Hidden: 0x40b5cn, + /** Subscriber max age, from the MoQ Timestamp extension. */ + MaxAge: 0x915c4n, /** AUTH, from the MoQ Auth extension. See `auth.ts`. */ Auth: 0x40b60n, /** ACTIVE_COUNT, from the MoQ Active Count extension. See `active_count.ts`. */ @@ -227,6 +229,7 @@ const MSG_PARAM_INCLUDE_PROPERTIES = 0x35n; const MSG_PARAM_ROUTE_COST = 0x40b58n; /// HIDDEN, from the MoQ Hidden extension. See `hidden.ts`. const MSG_PARAM_HIDDEN = 0x40b5en; +const MSG_PARAM_MAX_AGE = 0x915c6n; // Bytes parameter IDs (odd) /// AUTHORIZATION TOKEN. Decoded, but grants nothing: the session's grant is what authorizes a request. @@ -293,6 +296,7 @@ function allows(message: ControlMessage, version: IetfVersion, id: bigint): bool switch (message) { case "subscribe": return ( + id === MSG_PARAM_MAX_AGE || objectTimeout || auth || (id === MSG_PARAM_MAX_CACHE_DURATION && from17) || @@ -314,6 +318,7 @@ function allows(message: ControlMessage, version: IetfVersion, id: bigint): bool case "subscribe-update": case "request-update": return ( + id === MSG_PARAM_MAX_AGE || objectTimeout || auth || subgroupTimeout || @@ -398,6 +403,7 @@ function getMessageParamKind(id: bigint): MessageParamKind { case MSG_PARAM_EXPIRES: case MSG_PARAM_ROUTE_COST: case MSG_PARAM_HIDDEN: + case MSG_PARAM_MAX_AGE: case MSG_PARAM_ACTIVE_COUNT: return "varint"; case MSG_PARAM_PUBLISHER_PRIORITY: @@ -550,6 +556,21 @@ export class Parameters { this.vars.set(MSG_PARAM_MAX_CACHE_DURATION, v); } + /** Subscriber max age in milliseconds (MoQ Timestamp). */ + get maxAge(): number | undefined { + const value = this.vars.get(MSG_PARAM_MAX_AGE); + if (value === undefined) return undefined; + if (value > BigInt(Number.MAX_SAFE_INTEGER)) throw new RangeError("max age exceeds safe milliseconds"); + return Number(value); + } + set maxAge(value: number | undefined) { + if (value === undefined) this.vars.delete(MSG_PARAM_MAX_AGE); + else { + if (!Number.isSafeInteger(value) || value < 0) throw new RangeError("invalid max age"); + this.vars.set(MSG_PARAM_MAX_AGE, BigInt(value)); + } + } + /** HIDDEN (MoQ Hidden): also advertise hidden namespaces. Absent and 0 both mean no. */ get hidden(): boolean { const v = this.vars.get(MSG_PARAM_HIDDEN); diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 35e36a09d0..f300507ade 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -3584,3 +3584,37 @@ test.each([0, Number.MAX_SAFE_INTEGER + 2])( origin.close(); }, ); + +test("subscriber max age and REQUEST_UPDATE preserve omitted preferences", async () => { + const fx = fixture(Version.DRAFT_19); + const track = fx.broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const { client } = await runSubscribe( + fx, + new Subscribe({ + requestId: 0n, + trackNamespace: Path.from("test"), + trackName: "video", + subscriberPriority: 128, + maxAge: 250, + }), + ); + try { + expect(track.subscription.peek()?.maxDelay).toBe(Milli(250)); + const parameters = new Parameters(); + parameters.maxAge = 900; + await client.writer.u53(SubscribeUpdate.id); + await new SubscribeUpdate({ requestId: 2n, parameters }).encode(client.writer, fx.version); + expect(await client.reader.u53()).toBe(RequestOk.id); + await RequestOk.decode(client.reader, fx.version); + expect(track.subscription.peek()?.maxDelay).toBe(Milli(900)); + await client.writer.u53(SubscribeUpdate.id); + await new SubscribeUpdate({ requestId: 4n }).encode(client.writer, fx.version); + expect(await client.reader.u53()).toBe(RequestOk.id); + await RequestOk.decode(client.reader, fx.version); + expect(track.subscription.peek()?.maxDelay).toBe(Milli(900)); + } finally { + client.close(); + track.close(); + fx.close(); + } +}); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index a22767d964..2ee0529792 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -41,6 +41,7 @@ import { encodeDatagram } from "./datagram.ts"; import { type RequestKind, requestReason, toRequestCode } from "./error.ts"; import { type Fetch, FetchError, FetchHeader } from "./fetch.ts"; import * as Filter from "./filter.ts"; +import { MAX_AGE_PARAMETER } from "./max_age.ts"; import * as Message from "./message.ts"; import * as Namespace from "./namespace.ts"; import { FetchFrame, Frame, Group as GroupMessage } from "./object.ts"; @@ -498,11 +499,8 @@ export class Publisher { const track = wireOf(broadcast).subscribe(msg.trackName, { priority, order, - // moq-transport has no subscriber latency parameter. Keep everything the - // producer retained and let the receiving subscriber enforce its own budget. - // Keep the sentinel encodable if this demand crosses a Lite hop before the - // producer's retention bound is known. - maxDelay: Milli(Varint.MAX_U53), + // An absent budget preserves everything retained by the producer. + maxDelay: Milli(msg.maxAge ?? Varint.MAX_U53), groups: fixed.start ? rangeGroups(fixed) : undefined, }); @@ -555,7 +553,7 @@ export class Publisher { track.update({ priority, order, - maxDelay: Milli(Varint.MAX_U53), + maxDelay: Milli(msg.maxAge ?? Varint.MAX_U53), groups: rangeGroups(range), }); const startGroup = range.start ? Number(range.start.group) : hooks.latestSequence(track); @@ -681,13 +679,20 @@ export class Publisher { params.bytes.size > 0 || params.rangeFilters || [...params.vars.keys()].some( - (key) => key !== 0x02n && key !== 0x06n && key !== 0x10n && key !== 0x20n, + (key) => + key !== MAX_AGE_PARAMETER && + key !== 0x02n && + key !== 0x06n && + key !== 0x10n && + key !== 0x20n, ); if (unsupported) throw await refuse("not_supported", "REQUEST_UPDATE parameters not supported", ownRequestId); // update() replaces every option, so carry the rest over as Rust does. if (params.subscriberPriority !== undefined) track.update({ ...track.subscription.peek(), priority: fromWire(params.subscriberPriority) }); + if (params.maxAge !== undefined) + track.update({ ...track.subscription.peek(), maxDelay: Milli(params.maxAge) }); if (version === Version.DRAFT_14) continue; await stream.writer.u53(RequestOk.id); await new RequestOk({ requestId: ownRequestId }).encode(stream.writer, version); diff --git a/js/net/src/ietf/subscribe.ts b/js/net/src/ietf/subscribe.ts index d01807ff9d..d00a83d75f 100644 --- a/js/net/src/ietf/subscribe.ts +++ b/js/net/src/ietf/subscribe.ts @@ -64,6 +64,8 @@ export class Subscribe { trackNamespace: Path.Valid; trackName: string; subscriberPriority: number; + /** Subscriber max age in milliseconds (MoQ Timestamp), if negotiated. */ + maxAge?: number; /** The order the publisher is asked to deliver groups in (GROUP_ORDER). Defaults to descending. */ groupOrder: Order; @@ -97,6 +99,7 @@ export class Subscribe { trackNamespace, trackName, subscriberPriority, + maxAge, groupOrder, filter, fill, @@ -110,6 +113,7 @@ export class Subscribe { trackNamespace: Path.Valid; trackName: string; subscriberPriority: number; + maxAge?: number; groupOrder?: Order; filter?: Filter.Filter; fill?: Filter.Fill; @@ -123,6 +127,7 @@ export class Subscribe { this.trackNamespace = trackNamespace; this.trackName = trackName; this.subscriberPriority = subscriberPriority; + this.maxAge = maxAge; this.groupOrder = groupOrder ?? "descending"; this.filter = filter ?? { kind: "unfiltered" }; this.fill = fill; @@ -146,11 +151,14 @@ export class Subscribe { await w.u8(encodeGroupOrder(this.groupOrder)); await w.bool(this.forward); await w.write(Filter.encode(this.filter, version)); - await w.u53(0); // no parameters + const params = new Parameters(); + params.maxAge = this.maxAge; + await params.encode(w, version); } else { // v15+: fields moved into parameters const params = new Parameters(); params.subscriberPriority = this.subscriberPriority; + params.maxAge = this.maxAge; params.groupOrder = encodeGroupOrder(this.groupOrder); params.forward = this.forward; params.subscriptionFilter = this.filter; @@ -197,10 +205,11 @@ export class Subscribe { const forward = await r.bool(); const filter = await Filter.decodeInline(r); - await Parameters.decode(r, version); // ignore parameters + const params = await Parameters.decode(r, version, "subscribe"); return new Subscribe({ requestId, + maxAge: params.maxAge, trackNamespace, trackName, subscriberPriority, @@ -235,6 +244,7 @@ export class Subscribe { const fill = rawFill !== undefined ? Filter.decodeFill(rawFill, version) : undefined; return new Subscribe({ + maxAge: params.maxAge, requestId, trackNamespace, trackName, diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index 16cf47f245..b42b699eb5 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -2273,6 +2273,36 @@ test.each(["done", "fin"] as const)("a counted set cut short by %s closes the se } }); +for (const negotiated of [false, true]) { + test(`subscriber max age is sent only to a declaring peer: ${negotiated}`, async () => { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const subscriber = new Subscriber({ + session: new NativeSession(pair.server, VERSION, true), + maxAge: negotiated, + }); + const track = subscriber + .consume(Path.from("room")) + .track("video") + .subscribe({ maxDelay: Milli(250) }); + const peer = await nextStream(pair.client); + if (!peer) throw new Error("no subscribe stream"); + expect(await peer.reader.u53()).toBe(Subscribe.id); + const request = await Subscribe.decode(peer.reader, VERSION); + expect(request.maxAge).toBe(negotiated ? 250 : undefined); + await peer.writer.u53(SubscribeOk.id); + await new SubscribeOk({ requestId: request.requestId, trackAlias: ALIAS }).encode(peer.writer, VERSION); + await track.info(); + // The aggregate can change before SUBSCRIBE_OK: the first update must catch it. + const tolerant = track.fork({ maxDelay: Milli(900), priority: 9 }); + expect(await peer.reader.u53()).toBe(SubscribeUpdate.id); + const update = await SubscribeUpdate.decode(peer.reader, VERSION); + expect(update.parameters.maxAge).toBe(negotiated ? 900 : undefined); + expect(update.parameters.subscriberPriority).toBe(246); + tolerant.close(); + track.close(); + peer.close(); + }); +} /** Advertise `path` inline under `epoch`, on a session that negotiated MoQ Epoch alone. */ async function inlineEpoch(stream: Stream, path: Path.Valid, epoch: Epoch.Valid): Promise { await stream.writer.u53(SubscribeNamespaceEntry.id); diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 506828064c..16f1c2262a 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -191,6 +191,7 @@ export class Subscriber { // Whether the peer understands the HIDDEN parameter (MoQ Hidden). #hidden: boolean; + #maxAge: boolean; // Whether a SUBSCRIBE_NAMESPACE's REQUEST_OK counts the NAMESPACEs that follow it (MoQ Active Count). #activeCount: boolean; @@ -221,6 +222,7 @@ export class Subscriber { cluster, epoch = false, hidden = false, + maxAge = false, activeCount = false, solicit, goaway, @@ -237,6 +239,8 @@ export class Subscriber { epoch?: boolean; /** Whether the peer understands the HIDDEN parameter (MoQ Hidden). */ hidden?: boolean; + /** Whether the peer accepts subscriber max age. */ + maxAge?: boolean; /** Whether both SETUPs declared MoQ Active Count. */ activeCount?: boolean; /** What the peer's SETUP declared about being solicited (MoQ Solicit). */ @@ -254,6 +258,7 @@ export class Subscriber { this.#cluster = cluster; this.#epoch = epoch; this.#hidden = hidden; + this.#maxAge = maxAge; this.#activeCount = activeCount; this.#solicit = solicit; this.#goaway = goaway; @@ -805,6 +810,7 @@ export class Subscriber { let trackAlias: bigint; // The SUBSCRIBER_PRIORITY the request went out with. let priority: number; + let maxAge: number | undefined; const abandoned = new Error("subscribe abandoned before it was accepted"); try { // Returning demand keeps the same setup and its original deadline. @@ -815,6 +821,7 @@ export class Subscriber { stream = result.stream; trackAlias = result.alias; priority = result.priority; + maxAge = result.maxAge; break; } if (demand.closed.peek() === undefined && demand.wanted.peek() !== undefined) continue; @@ -899,7 +906,7 @@ export class Subscriber { producer.closed.then(() => localEnded), revoked, ]); - const updates = this.#runUpdates(stream, requestId, producer, priority, unanswered, done); + const updates = this.#runUpdates(stream, requestId, producer, priority, maxAge, unanswered, done); // Serve until a terminal condition fires or the last local subscriber leaves. The unused // wake is level-triggered: re-check demand so a subscriber that returns before we tear @@ -1009,6 +1016,7 @@ export class Subscriber { requestId: bigint, producer: track.Producer, sent: number, + sentMaxAge: number | undefined, unanswered: Signal, done: Promise, ): Promise { @@ -1022,8 +1030,10 @@ export class Subscriber { void stopped.then(() => stop.abort()); for (;;) { - const wanted = toWire(producer.subscription.peek()?.priority ?? 0); - if (wanted === sent || unanswered.peek() > 0) { + const subscription = producer.subscription.peek(); + const wanted = toWire(subscription?.priority ?? 0); + const wantedMaxAge = this.#maxAge ? (subscription?.maxDelay ?? Number.MAX_SAFE_INTEGER) : undefined; + if ((wanted === sent && wantedMaxAge === sentMaxAge) || unanswered.peek() > 0) { const disposes: Dispose[] = []; const wake = new Promise((resolve) => { disposes.push( @@ -1047,6 +1057,7 @@ export class Subscriber { const parameters = new Parameters(); parameters.subscriberPriority = wanted; + parameters.maxAge = wantedMaxAge; const update = legacy ? new SubscribeUpdate({ requestId, ownRequestId: own, parameters }) : new SubscribeUpdate({ requestId: own, parameters }); @@ -1060,6 +1071,7 @@ export class Subscriber { return; } sent = wanted; + sentMaxAge = wantedMaxAge; console.debug(`subscribe update: id=${requestId} priority=${wanted}`); } } @@ -1097,7 +1109,7 @@ export class Subscriber { requestId: bigint, deadline: Deadline, epoch: Epoch.Valid | undefined, - ): Promise<{ stream: Stream; alias: bigint; priority: number }> { + ): Promise<{ stream: Stream; alias: bigint; priority: number; maxAge?: number }> { const version = this.#session.version; const legacy = version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16; @@ -1122,6 +1134,7 @@ export class Subscriber { trackName: request.name, // Read now, since a subscriber can join while the stream opened. Updates follow it. subscriberPriority: toWire(request.priority), + maxAge: this.#maxAge ? (request.subscription.maxDelay ?? 0) : undefined, // Fixed for the subscription's life: no REQUEST_UPDATE parameter carries it. groupOrder: request.subscription.order, // No fill is requested: the fill's fetch stream and the live subscription split @@ -1204,7 +1217,7 @@ export class Subscriber { throw err; } state.registeredAlias = ok.trackAlias; - return { stream: state.stream, alias: ok.trackAlias, priority: msg.subscriberPriority }; + return { stream: state.stream, alias: ok.trackAlias, priority: msg.subscriberPriority, maxAge: msg.maxAge }; } /** diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 1a344a5d3d..213c5667df 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -234,6 +234,43 @@ test("integration: lite subscription options and updates reach the publisher", a server.abort(); }); +// Both handshake paths: a SETUP-negotiated draft (16) and ALPN-selected ones (17+). +test.each([Ietf.ALPN.DRAFT_16, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])( + "integration: %s carries the subscriber max age to the publisher", + async (alpn) => { + const pair = createMockTransportPair(alpn); + const origin = new OriginProducer(); + const [client, server] = await Promise.all([ + connect(url, { transport: pair.client }), + accept(pair.server, url, { publish: origin.consume() }), + ]); + + const broadcast = publish(origin, Path.from("test")); + const accepted = (async () => { + const request = await wireOf(broadcast).requested(); + if (!request) throw new Error("no request"); + return request.accept({ timescale: Timescale.MILLI }); + })(); + + const remote = wireOf(client).consume(Path.from("test")); + const subscriber = remote.track("video").subscribe({ maxDelay: Milli(250) }); + const producer = await accepted; + // Without the negotiated parameter the publisher keeps its whole backlog instead. + expect(producer.subscription.peek()?.maxDelay).toBe(Milli(250)); + + const updated = producer.subscription.changed(); + subscriber.update({ maxDelay: Milli(500) }); + expect((await updated)?.maxDelay).toBe(Milli(500)); + + subscriber.close(); + remote.close(); + broadcast.close(); + origin.close(); + client.abort(); + server.abort(); + }, +); + test("integration: lite carries a fractional maxDelay as a whole millisecond", async () => { const pair = createMockTransportPair(Lite.ALPN_05); const origin = new OriginProducer(); diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 2596354e2c..77ac0a8c61 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -400,6 +400,7 @@ impl Client { ietf::solicit::into_setup(&mut parameters, ietf_encoding); } ietf::hidden::into_setup(&mut parameters, ietf_encoding); + ietf::max_age::into_setup(&mut parameters, ietf_encoding); ietf::active_count::into_setup(&mut parameters, ietf_encoding); let parameters = parameters.encode_bytes(ietf_encoding)?; @@ -457,6 +458,7 @@ impl Client { let peer_declared = ietf::peer::Peer { solicit: ietf::solicit::from_setup(¶meters, v)?, hidden: ietf::hidden::from_setup(¶meters, v), + max_age: ietf::max_age::from_setup(¶meters, v), active_count: ietf::active_count::from_setup(¶meters, v), ..Default::default() }; diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index a2a48927a4..442fead6fe 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -634,6 +634,7 @@ impl Default for Messages { frame_start: 0, }, ietf_subscribe: ietf::Subscribe { + max_age: None, new_group: None, request_id: ietf::RequestId(2), track_namespace: Path::new("room/alice"), diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index 30aa757bdd..5e1ad51183 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -1419,6 +1419,7 @@ mod tests { for version in [Version::Draft14, Version::Draft15, Version::Draft16] { let update = ietf::SubscribeUpdate { + max_age: None, new_group: None, request_id: update_id, subscription_request_id: Some(target_id), @@ -1956,6 +1957,7 @@ mod tests { async fn an_update_past_the_window_closes_the_session() { const VERSION: Version = Version::Draft14; let update = ietf::SubscribeUpdate { + max_age: None, new_group: None, request_id: RequestId(4), subscription_request_id: Some(RequestId(0)), diff --git a/rs/moq-net/src/ietf/max_age.rs b/rs/moq-net/src/ietf/max_age.rs new file mode 100644 index 0000000000..672c0850bb --- /dev/null +++ b/rs/moq-net/src/ietf/max_age.rs @@ -0,0 +1,44 @@ +//! Subscriber age budgets from the MoQ Timestamp extension. + +use super::{ParameterVarInt, Parameters, Version}; + +pub(super) const OPTION: u64 = 0x915C4; +pub(super) const PARAMETER: u64 = 0x915C6; +pub(super) const MAX_MILLIS: u64 = (1_u64 << 53) - 1; + +pub(crate) fn from_setup(params: &Parameters, _version: Version) -> bool { + params.get_varint(ParameterVarInt::MaxAge).is_some() +} + +pub(crate) fn into_setup(params: &mut Parameters, _version: Version) { + params.set_varint(ParameterVarInt::MaxAge, 1); +} + +pub(super) fn validate(value: Option) -> Result, crate::DecodeError> { + if value.is_some_and(|value| value > MAX_MILLIS) { + return Err(crate::DecodeError::InvalidValue); + } + Ok(value) +} + +pub(super) fn millis(delay: std::time::Duration) -> u64 { + delay.as_millis().min(MAX_MILLIS as u128) as u64 +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn negotiated_separately_and_bounded() { + for version in [Version::Draft14, Version::Draft16, Version::Draft22] { + let mut params = Parameters::default(); + assert!(!from_setup(¶ms, version)); + into_setup(&mut params, version); + assert!(from_setup(¶ms, version)); + } + assert_eq!(validate(None).unwrap(), None); + assert_eq!(validate(Some(0)).unwrap(), Some(0)); + assert!(validate(Some(MAX_MILLIS + 1)).is_err()); + assert_eq!(millis(std::time::Duration::MAX), MAX_MILLIS); + } +} diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index e0e35c0bf3..6f0998fc40 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -25,6 +25,7 @@ mod goaway; mod group; pub mod hidden; mod location; +pub(crate) mod max_age; pub mod message; mod namespace; mod object; diff --git a/rs/moq-net/src/ietf/parameters.rs b/rs/moq-net/src/ietf/parameters.rs index fcbfba381f..a662500e4b 100644 --- a/rs/moq-net/src/ietf/parameters.rs +++ b/rs/moq-net/src/ietf/parameters.rs @@ -31,6 +31,8 @@ pub enum ParameterVarInt { Auth = super::auth::AUTH, /// ACTIVE_COUNT, from the MoQ Active Count extension. ActiveCount = super::active_count::ACTIVE_COUNT, + /// Subscriber max age, from the MoQ Timestamp extension. + MaxAge = super::max_age::OPTION, /// EPOCH, from the MoQ Epoch extension. Epoch = super::epoch::EPOCH, #[num_enum(catch_all)] diff --git a/rs/moq-net/src/ietf/peer.rs b/rs/moq-net/src/ietf/peer.rs index f580642b37..05f7ec9fb5 100644 --- a/rs/moq-net/src/ietf/peer.rs +++ b/rs/moq-net/src/ietf/peer.rs @@ -20,6 +20,9 @@ pub(crate) struct Peer { /// SUBSCRIBE_NAMESPACE, so we may send it. pub hidden: bool, + /// Whether the peer understands subscriber max age (MoQ Timestamp). + pub max_age: bool, + /// MoQ Auth: whether both sides negotiated the Auth request streams. pub auth: bool, @@ -46,6 +49,14 @@ pub(crate) struct PeerSetup { } impl PeerSetup { + /// A fixture whose peer already sent its SETUP. + #[cfg(test)] + pub(crate) fn declared(peer: Peer) -> Self { + let setup = Self::default(); + setup.set(peer); + setup + } + /// A slot for a session that offered `offered`. pub fn new(offered: crate::setup::Extensions) -> Self { Self { @@ -103,6 +114,7 @@ mod tests { }, solicit: None, hidden: false, + max_age: false, auth: false, active_count: false, epoch: false, @@ -117,6 +129,7 @@ mod tests { }, solicit: Some(true), hidden: true, + max_age: false, auth: true, active_count: true, epoch: true, diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index dc743891e1..c61417cff5 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -26,10 +26,9 @@ use crate::{ use super::{Message, Version, cluster, error::request, peer, request_update}; /// Largest millisecond duration every implementation can carry losslessly. -const MAX_SAFE_DELAY_MS: u64 = (1_u64 << 53) - 1; +const MAX_SAFE_DELAY_MS: u64 = super::max_age::MAX_MILLIS; -/// Build the serving-side subscription for a peer whose wire protocol carries no -/// max delay preference. The receiver applies its own budget after the transfer. +/// Preserve retained groups when the subscriber supplies no age budget. fn serving_subscription(subscriber_priority: u8) -> Subscription { Subscription { priority: super::priority::from_wire(subscriber_priority), @@ -1221,6 +1220,9 @@ where // A live SUBSCRIBE's NEW_GROUP_REQUEST asks nothing more: subscribing is itself // a join, which the track's refresh ratio judges. let mut subscription = serving_subscription(msg.subscriber_priority).with_order(order); + if let Some(max_age) = msg.max_age { + subscription.max_delay = Duration::from_millis(max_age); + } // A range that resolves without the live edge has a fixed start, which is no // join: bound the subscription before subscribing so the ratio never judges it. // From the start of its group, so a relay's copy holds the group's head for @@ -1712,6 +1714,11 @@ where .set_priority(super::priority::from_wire(priority)) .map_err(failed)?; } + if let Some(max_age) = update.max_age { + let mut subscription = serve.track.subscription(); + subscription.max_delay = Duration::from_millis(max_age); + serve.track.update(subscription).map_err(failed)?; + } if let Some(value) = update.new_group { serve.request_new_group(value); } @@ -6777,6 +6784,7 @@ mod serve_tests { Version::Draft14 => { let mut body = Vec::new(); ietf::SubscribeUpdate { + max_age: None, new_group: None, request_id: RequestId(UPDATE_ID.into()), subscription_request_id: Some(RequestId(0)), @@ -6887,6 +6895,7 @@ mod serve_tests { fn new_group_update(version: Version, id: u64, value: u64) -> Vec { let mut body = Vec::new(); ietf::SubscribeUpdate { + max_age: None, new_group: Some(value), request_id: RequestId(id), subscription_request_id: matches!(version, Version::Draft15 | Version::Draft16).then_some(RequestId(0)), @@ -7077,6 +7086,7 @@ mod serve_tests { Version::Draft14 => { let mut body = Vec::new(); ietf::SubscribeUpdate { + max_age: None, new_group: None, request_id: RequestId(2), subscription_request_id: Some(RequestId(0)), @@ -7487,6 +7497,7 @@ mod serve_tests { fn subscribe(filter: Filter, fill: Option) -> ietf::Subscribe<'static> { ietf::Subscribe { + max_age: None, new_group: None, request_id: RequestId(REQUEST_ID), track_namespace: crate::Path::new("room"), @@ -11038,6 +11049,7 @@ mod serve_tests { _ => None, }; let msg = ietf::SubscribeUpdate { + max_age: None, new_group: None, request_id: RequestId(0x40), subscription_request_id, @@ -11350,6 +11362,7 @@ mod serve_tests { _ => None, }; let msg = ietf::SubscribeUpdate { + max_age: None, new_group: None, request_id: RequestId(rid), subscription_request_id, @@ -11376,6 +11389,7 @@ mod serve_tests { _ => None, }; let msg = ietf::SubscribeUpdate { + max_age: None, new_group: None, request_id: RequestId(rid), subscription_request_id, @@ -12569,8 +12583,7 @@ mod tests { slot } - /// moq-transport cannot carry the receiver's max delay budget, so the serving - /// subscription must preserve everything the producer still retains. + /// An absent max age preserves everything the producer still retains. #[test] fn serving_subscription_keeps_retained_backlog() { let producer = track::Producer::new( @@ -13874,6 +13887,7 @@ mod tests { }, solicit, hidden: false, + max_age: false, auth: false, active_count: false, epoch: false, @@ -14779,6 +14793,7 @@ mod tests { .run_subscribe_stream( stream, ietf::Subscribe { + max_age: None, new_group: None, request_id: RequestId(1), track_namespace: crate::Path::new("nothing/here"), @@ -15152,6 +15167,7 @@ mod range_tests { fn subscribe(filter: Filter) -> ietf::Subscribe<'static> { ietf::Subscribe { + max_age: None, new_group: None, request_id: RequestId(1), track_namespace: crate::Path::new("broadcast"), diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs index b2d7c4ee89..108755bba2 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -224,6 +224,7 @@ pub(super) struct Update { /// The first Request ID: the update's own on drafts 14-16, which their answer names. pub(super) request_id: super::RequestId, pub(super) priority: Option, + pub(super) max_age: Option, /// Why the update is refused, which fails the subscription: [`Error::Unsupported`], a /// [`DecodeError::MalformedAuthToken`] for tokens that repeat, or a /// [`DecodeError::MultipleAuthTokens`] for more than one token. @@ -246,7 +247,7 @@ impl Update { /// Decode the message body, after its type and length. fn decode_body(data: &mut Decoder<'_>, version: Version) -> Result { - use super::{Fill, Filter, Location, Opaque, Parameters, RequestId}; + use super::{Fill, Filter, Location, Opaque, RequestId}; let request_id = RequestId::decode(data, version)?; match version { Version::Draft14 => { @@ -259,7 +260,10 @@ impl Update { let _end_group = data.varint()?; let priority = data.u8()?; let forward = data.bool()?; - let tokens = Parameters::skip_request(data)?; + decode_params!(data, version, + 0x03 => tokens: Vec, + super::max_age::PARAMETER => max_age: Option, + ); if !data.is_empty() { return Err(DecodeError::InvalidValue); } @@ -267,6 +271,7 @@ impl Update { return Ok(Self { request_id, priority: Some(priority), + max_age: super::max_age::validate(max_age)?, refused, authorization_token, new_group: None, @@ -295,6 +300,7 @@ impl Update { 0x28 => property_filter: Vec where super::subscribe::has_range_filters(version), 0x29 => track_filter: Vec where super::subscribe::has_range_filters(version), 0x32 => new_group: Option where super::subscribe::has_new_group_request(version), + super::max_age::PARAMETER => max_age: Option, ); if !data.is_empty() { return Err(DecodeError::InvalidValue); @@ -314,6 +320,7 @@ impl Update { Ok(Self { request_id, priority, + max_age: super::max_age::validate(max_age)?, refused, authorization_token, new_group, diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index abbae7f402..3f5f681511 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -10,7 +10,7 @@ use crate::{ use super::{ Control, Message, Publisher, Subscriber, Version, active_count, adapter::ControlStreamAdapter, - auth, cluster, epoch, hidden, peer, request_update, solicit, + auth, cluster, epoch, hidden, max_age, peer, request_update, solicit, subscriber::{is_protocol_violation, subscribe_prefixes}, }; @@ -766,6 +766,7 @@ fn peer_from_params(params: &ietf::Parameters, version: Version) -> Result( solicit::into_setup(&mut parameters, version); } hidden::into_setup(&mut parameters, version); + max_age::into_setup(&mut parameters, version); if extensions.auth { auth::into_setup(&mut parameters, version); } @@ -2579,6 +2581,7 @@ mod tests { let mut encoder = crate::coding::Encoder::new(&mut subscribe, VERSION.into()); encoder.varint(ietf::Subscribe::ID).unwrap(); ietf::Subscribe { + max_age: None, new_group: None, request_id: RequestId(1), track_namespace: crate::Path::new("room"), @@ -3000,6 +3003,7 @@ mod tests { /// `token`, framed as a control message. fn subscribe_with_token(version: Version, token: &[u8]) -> Vec { let msg = ietf::Subscribe { + max_age: None, new_group: None, request_id: RequestId(0), track_namespace: crate::Path::new("room"), diff --git a/rs/moq-net/src/ietf/subscribe.rs b/rs/moq-net/src/ietf/subscribe.rs index 5da339b17f..3967895258 100644 --- a/rs/moq-net/src/ietf/subscribe.rs +++ b/rs/moq-net/src/ietf/subscribe.rs @@ -46,6 +46,8 @@ pub struct Subscribe<'a> { /// NEW_GROUP_REQUEST (0x32, draft-15+): the subscriber's largest known Group ID plus /// one, or 0 when it knows none. Never encoded where the draft has no such parameter. pub new_group: Option, + /// Subscriber max age in milliseconds (MoQ Timestamp), if negotiated. + pub max_age: Option, /// The publisher instance the subscriber expects (MoQ Epoch, draft-17+). Only on a /// session that negotiated it; the session refuses one that arrives otherwise. pub epoch: Option, @@ -83,7 +85,10 @@ impl Message for Subscribe<'_> { let forward = r.bool()?; let filter = Filter::decode(r, version)?; - let tokens = Parameters::skip_request(r)?; + decode_params!(r, version, + 0x03 => tokens: Vec, + super::max_age::PARAMETER => max_age: Option, + ); let authorization_token = super::token::single(tokens)?; Ok(Self { @@ -100,6 +105,7 @@ impl Message for Subscribe<'_> { range_filters: false, authorization_token, new_group: None, + max_age: super::max_age::validate(max_age)?, epoch: None, }) } @@ -123,6 +129,7 @@ impl Message for Subscribe<'_> { 0x32 => new_group: Option where has_new_group_request(version), 0x35 => include_properties: Option where Filter::is_draft20(version), epoch::EPOCH_PARAM => epoch: Option where epoch::supported(version), + super::max_age::PARAMETER => max_age: Option, ); let authorization_token = super::token::single(authorization_token)?; @@ -161,6 +168,7 @@ impl Message for Subscribe<'_> { range_filters, authorization_token, new_group, + max_age: super::max_age::validate(max_age)?, epoch, }) } @@ -184,6 +192,7 @@ impl Message for Subscribe<'_> { self.filter.encode(w, version)?; encode_params!(w, version, 0x03 => self.authorization_token.clone(), + super::max_age::PARAMETER => self.max_age, ); } _ => { @@ -210,6 +219,7 @@ impl Message for Subscribe<'_> { 0x32 => new_group, 0x35 => include_properties, epoch::EPOCH_PARAM => epoch, + super::max_age::PARAMETER => self.max_age, ); } } @@ -422,6 +432,8 @@ pub struct SubscribeUpdate { pub authorization_token: Option, /// NEW_GROUP_REQUEST (0x32, draft-16+), as on [`Subscribe::new_group`]. pub new_group: Option, + /// Subscriber max age in milliseconds (MoQ Timestamp), if negotiated. + pub max_age: Option, } impl Message for SubscribeUpdate { @@ -440,6 +452,7 @@ impl Message for SubscribeUpdate { w.bool(self.forward.unwrap_or(true)); encode_params!(w, version, 0x03 => self.authorization_token.clone(), + super::max_age::PARAMETER => self.max_age, ); } Version::Draft15 | Version::Draft16 => { @@ -453,6 +466,7 @@ impl Message for SubscribeUpdate { 0x20 => self.subscriber_priority, 0x21 => self.filter, 0x32 => self.new_group.filter(|_| has_new_group_request(version)), + super::max_age::PARAMETER => self.max_age, ); } _ => { @@ -471,6 +485,7 @@ impl Message for SubscribeUpdate { 0x20 => self.subscriber_priority, 0x21 => self.filter, 0x32 => self.new_group, + super::max_age::PARAMETER => self.max_age, ); } } @@ -487,7 +502,10 @@ impl Message for SubscribeUpdate { let end_group = r.varint()?; let subscriber_priority = Some(r.u8()?); let forward = Some(r.bool()?); - let tokens = Parameters::skip_request(r)?; + decode_params!(r, version, + 0x03 => tokens: Vec, + super::max_age::PARAMETER => max_age: Option, + ); let authorization_token = super::token::single(tokens)?; Ok(Self { @@ -500,6 +518,7 @@ impl Message for SubscribeUpdate { filter: None, authorization_token, new_group: None, + max_age: super::max_age::validate(max_age)?, }) } Version::Draft15 | Version::Draft16 => { @@ -512,6 +531,7 @@ impl Message for SubscribeUpdate { 0x20 => subscriber_priority: Option, 0x21 => filter: Option, 0x32 => new_group: Option where has_new_group_request(version), + super::max_age::PARAMETER => max_age: Option, ); Ok(Self { @@ -524,6 +544,7 @@ impl Message for SubscribeUpdate { filter, authorization_token: super::token::single(authorization_token)?, new_group, + max_age: super::max_age::validate(max_age)?, }) } _ => { @@ -548,6 +569,7 @@ impl Message for SubscribeUpdate { 0x28 => _object_property_filter: Vec where has_range_filters(version), 0x29 => _track_property_filter: Vec where has_range_filters(version), 0x32 => new_group: Option, + super::max_age::PARAMETER => max_age: Option, ); Ok(Self { @@ -560,6 +582,7 @@ impl Message for SubscribeUpdate { filter, authorization_token: super::token::single(authorization_token)?, new_group, + max_age: super::max_age::validate(max_age)?, }) } } @@ -582,10 +605,63 @@ mod tests { crate::coding::decode_buf(&mut buf, version, M::decode_msg) } + #[test] + fn subscriber_max_age_round_trips() { + for version in [ + Version::Draft14, + Version::Draft15, + Version::Draft16, + Version::Draft17, + Version::Draft18, + Version::Draft19, + Version::Draft20, + Version::Draft21, + Version::Draft22, + ] { + for max_age in [None, Some(0), Some(250), Some(super::super::max_age::MAX_MILLIS)] { + let msg = Subscribe { + max_age, + request_id: RequestId(0), + track_namespace: Path::new("live"), + track_name: "catalog".into(), + subscriber_priority: 128, + group_order: GroupOrder::Descending, + filter: Filter::Unfiltered, + fill: None, + properties_wanted: true, + forward: true, + rendezvous: false, + range_filters: false, + authorization_token: None, + new_group: None, + epoch: None, + }; + let decoded: Subscribe = decode_message(&encode_message(&msg, version), version).unwrap(); + assert_eq!(decoded.max_age, max_age, "{version}"); + let update = SubscribeUpdate { + max_age, + request_id: RequestId(2), + subscription_request_id: matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) + .then_some(RequestId(0)), + start_location: Location { group: 0, object: 0 }, + end_group: 0, + subscriber_priority: None, + forward: None, + filter: None, + authorization_token: None, + new_group: None, + }; + let decoded: SubscribeUpdate = decode_message(&encode_message(&update, version), version).unwrap(); + assert_eq!(decoded.max_age, max_age, "{version}"); + } + } + } + #[test] fn test_subscribe_round_trip() { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(1), track_namespace: Path::new("test"), track_name: "video".into(), @@ -614,6 +690,7 @@ mod tests { fn test_subscribe_round_trip_v15() { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(1), track_namespace: Path::new("test"), track_name: "video".into(), @@ -659,6 +736,7 @@ mod tests { let expected = has_new_group_request(version).then_some(7); let subscribe = Subscribe { new_group: Some(7), + max_age: None, request_id: RequestId(1), track_namespace: Path::new("test"), track_name: "video".into(), @@ -678,6 +756,7 @@ mod tests { let update = SubscribeUpdate { new_group: Some(7), + max_age: None, request_id: RequestId(3), subscription_request_id: matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) .then_some(RequestId(1)), @@ -766,6 +845,7 @@ mod tests { fn rendezvous_timeout_is_never_sent() { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(1), track_namespace: Path::new("test"), track_name: "video".into(), @@ -801,6 +881,7 @@ mod tests { fn test_subscribe_nested_namespace() { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(100), track_namespace: Path::new("conference/room123"), track_name: "audio".into(), @@ -991,6 +1072,7 @@ mod tests { ] { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(1), track_namespace: crate::Path::new("broadcast"), track_name: "video".into(), @@ -1053,6 +1135,7 @@ mod tests { fn test_subscribe_update_v15_round_trip() { let msg = SubscribeUpdate { new_group: None, + max_age: None, request_id: RequestId(10), subscription_request_id: Some(RequestId(5)), start_location: Location { group: 0, object: 0 }, @@ -1076,6 +1159,7 @@ mod tests { fn test_subscribe_update_v14_round_trip() { let msg = SubscribeUpdate { new_group: None, + max_age: None, request_id: RequestId(10), subscription_request_id: Some(RequestId(5)), start_location: Location { group: 1, object: 2 }, @@ -1143,6 +1227,7 @@ mod tests { fn test_subscribe_v17_round_trip() { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(1), track_namespace: Path::new("test"), track_name: "video".into(), @@ -1187,6 +1272,7 @@ mod tests { fn test_subscribe_update_v17_round_trip() { let msg = SubscribeUpdate { new_group: None, + max_age: None, request_id: RequestId(10), subscription_request_id: None, start_location: Location { group: 0, object: 0 }, @@ -1210,6 +1296,7 @@ mod tests { fn test_subscribe_v18_round_trip() { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(1), track_namespace: Path::new("test"), track_name: "video".into(), @@ -1434,6 +1521,7 @@ mod tests { fn test_subscribe_update_v18_round_trip() { let msg = SubscribeUpdate { new_group: None, + max_age: None, request_id: RequestId(10), subscription_request_id: None, start_location: Location { group: 0, object: 0 }, @@ -1460,6 +1548,7 @@ mod tests { fn test_subscribe_update_v17_v18_size_differs() { let v17_msg = SubscribeUpdate { new_group: None, + max_age: None, request_id: RequestId(10), subscription_request_id: None, start_location: Location { group: 0, object: 0 }, @@ -1532,6 +1621,7 @@ mod tests { fn include_properties_is_a_uint8() { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(1), track_namespace: crate::Path::new("live"), track_name: "video".into(), @@ -1597,6 +1687,7 @@ mod tests { ] { let msg = Subscribe { new_group: None, + max_age: None, request_id: RequestId(1), track_namespace: crate::Path::new("live"), track_name: "video".into(), diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 28c85607b6..7d8a52fb5a 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2767,8 +2767,17 @@ where // Write Subscribe message. The aggregate is read now: a subscriber can join while // the request ID and stream were awaited. Updates follow its priority from here, but // no draft lets one change the order. + let max_age_supported = self.peer_setup.get().await.max_age; let aggregate = target.subscription(); let priority = aggregate.as_ref().map(|s| s.priority).unwrap_or(0); + let max_age = max_age_supported.then(|| { + super::max_age::millis( + aggregate + .as_ref() + .map(|s| s.max_delay) + .unwrap_or(Duration::from_millis(super::max_age::MAX_MILLIS)), + ) + }); let order = aggregate.map(|s| s.order).unwrap_or_default(); // Without an Established subscription, a relay includes the request it holds. let new_group = target @@ -2779,6 +2788,7 @@ where stream .writer .encode(&ietf::Subscribe { + max_age, new_group, request_id, track_namespace: broadcast_path.to_owned(), @@ -3084,6 +3094,8 @@ where // The group the track's consumers want, `None` once nobody does. let mut wanted = None; // The subscriber priority upstream holds, and the one the aggregate now asks for. + let mut sent_max_age = max_age; + let mut wanted_max_age = max_age; let mut sent_priority = priority; let mut priority = priority; // Draft-14's SUBSCRIBE_UPDATE repeats the bounds it leaves alone, and must not move the @@ -3137,6 +3149,9 @@ where subscribed = subscription.is_some(); if let Some(subscription) = subscription { priority = subscription.priority; + if max_age_supported { + wanted_max_age = Some(super::max_age::millis(subscription.max_delay)); + } } } loop { @@ -3169,9 +3184,10 @@ where let update = Update { start_location, priority: (priority != sent_priority).then_some(priority), + max_age: wanted_max_age.filter(|_| wanted_max_age != sent_max_age), new_group, }; - match update.priority.is_some() || update.new_group.is_some() { + match update.priority.is_some() || update.max_age.is_some() || update.new_group.is_some() { true => Poll::Ready(End::Update(update)), false => Poll::Pending, } @@ -3180,12 +3196,15 @@ where match end { End::Update(update) => { - let (asked, new_group) = (update.priority, update.new_group); + let (asked, new_group, asked_age) = (update.priority, update.new_group, update.max_age); if let Err(err) = self.write_update(&mut stream.writer, request_id, update).await { tracing::debug!(%err, "failed to update the subscription"); let _ = track.clone().abort(err); break false; } + if asked_age.is_some() { + sent_max_age = asked_age; + } if let Some(asked) = asked { sent_priority = asked; } @@ -3453,6 +3472,7 @@ where writer.varint(ietf::SubscribeUpdate::ID).await?; writer .encode(&ietf::SubscribeUpdate { + max_age: update.max_age, request_id, subscription_request_id: matches!(self.version, Version::Draft14 | Version::Draft15 | Version::Draft16) .then_some(subscription), @@ -6705,7 +6725,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, VERSION, @@ -6763,7 +6783,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, VERSION, @@ -6826,7 +6846,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, VERSION, @@ -6886,7 +6906,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, VERSION, @@ -7017,7 +7037,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), control, None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, VERSION, @@ -7091,7 +7111,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, VERSION, @@ -7212,7 +7232,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, version, @@ -7320,7 +7340,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, version, @@ -7426,7 +7446,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, version, @@ -7504,7 +7524,7 @@ mod tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, version, @@ -11177,6 +11197,8 @@ struct Update { start_location: ietf::Location, /// The aggregate subscriber priority, in the model's higher-first terms. priority: Option, + /// Subscriber age budget, in milliseconds. + max_age: Option, /// NEW_GROUP_REQUEST. new_group: Option, } @@ -11981,6 +12003,7 @@ mod stitch_tests { ietf::Fetch::ID } else { ietf::Subscribe { + max_age: None, new_group: None, request_id: RequestId(3), track_namespace: Path::new("room"), @@ -12518,6 +12541,7 @@ mod stitch_tests { version, ); let subscribe = ietf::Subscribe { + max_age: None, new_group: None, request_id: RequestId(3), track_namespace: Path::new("room"), @@ -14267,7 +14291,7 @@ mod joining_fetch_tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), control, None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, version, @@ -15202,7 +15226,7 @@ mod new_group_tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, VERSION, @@ -15503,7 +15527,7 @@ mod track_property_tests { crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), Control::new(None, false), None, - peer::PeerSetup::default(), + peer::PeerSetup::declared(peer::Peer::default()), crate::Hop::new(1).unwrap(), None, version, diff --git a/rs/moq-net/src/ietf/version.rs b/rs/moq-net/src/ietf/version.rs index 2fdf1a2253..0e29f18454 100644 --- a/rs/moq-net/src/ietf/version.rs +++ b/rs/moq-net/src/ietf/version.rs @@ -89,6 +89,7 @@ mod tests { }; let subscribe = Subscribe { + max_age: None, new_group: None, request_id: RequestId(1), track_namespace: Path::new("broadcast"), @@ -168,6 +169,7 @@ mod tests { #[test] fn draft22_changes_only_the_location_filter() { let subscribe = Subscribe { + max_age: None, new_group: None, request_id: RequestId(1), track_namespace: Path::new("broadcast"), diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 9c6c1e7d54..8d46cfbb4d 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3331,7 +3331,13 @@ async fn serve_front(task: FrontTask) -> Vec { && let Some(accepted) = &mut io.accepted { while let Poll::Ready(Ok(_)) = accepted.poll_subscription_changed(waiter) {} - if accepted.subscription().is_none() { + if let Some(subscription) = accepted.subscription() { + // The query's hold must follow demand too, or its initial budget + // keeps a later, shorter aggregate from reaching the publisher. + if let Some(held) = &mut io.held { + let _ = held.update(subscription); + } + } else { io.held = None; } } diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index e6d8675d22..d338b256de 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -442,6 +442,7 @@ impl Server { let peer_declared = ietf::peer::Peer { solicit: ietf::solicit::from_setup(¶ms, v)?, hidden: ietf::hidden::from_setup(¶ms, v), + max_age: ietf::max_age::from_setup(¶ms, v), active_count: ietf::active_count::from_setup(¶ms, v), ..Default::default() }; @@ -715,6 +716,7 @@ where ietf::solicit::into_setup(&mut parameters, v); } ietf::hidden::into_setup(&mut parameters, v); + ietf::max_age::into_setup(&mut parameters, v); ietf::active_count::into_setup(&mut parameters, v); parameters.encode_bytes(v)? } diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index 7f2f87f934..bddf638a90 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -50,6 +50,7 @@ mod source_errors; mod subscribe_priority; mod subscribe_ranges; mod subscribe_tracks; +mod subscriber_max_age; mod subscription_end_integrity; mod support; mod synced; diff --git a/rs/moq-net/tests/integration/partial_group.rs b/rs/moq-net/tests/integration/partial_group.rs index 7e53d26bc6..54c35c6f7e 100644 --- a/rs/moq-net/tests/integration/partial_group.rs +++ b/rs/moq-net/tests/integration/partial_group.rs @@ -16,8 +16,20 @@ const TIMEOUT: Duration = Duration::from_secs(10); const BUDGET: Duration = Duration::from_millis(500); -/// moq-transport carries no max delay, so its publisher never gives up on a group. -const VERSIONS: &[&str] = &["moq-lite-06", "moq-lite-07-wip"]; +/// IETF peers negotiate the same superseded-group budget as Lite peers. +const VERSIONS: &[&str] = &[ + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-14", + "moq-transport-15", + "moq-transport-16", + "moq-transport-17", + "moq-transport-18", + "moq-transport-19", + "moq-transport-20", + "moq-transport-21", + "moq-transport-22", +]; 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())); diff --git a/rs/moq-net/tests/integration/subscriber_max_age.rs b/rs/moq-net/tests/integration/subscriber_max_age.rs new file mode 100644 index 0000000000..249d6a3d5c --- /dev/null +++ b/rs/moq-net/tests/integration/subscriber_max_age.rs @@ -0,0 +1,76 @@ +#![cfg(feature = "async")] + +//! IETF relays forward their most tolerant subscriber's negotiated age budget. + +use std::time::Duration; + +use moq_net::{Hop, Timestamp, Version, origin, track}; + +use crate::support::harness::{peer, spawn}; + +fn origin(hop: u64) -> origin::Producer { + let (origin, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + spawn(driver); + origin +} + +async fn budget(track: &mut track::Producer, millis: u64, version: Version) { + moq_clock::timeout(Duration::from_secs(5), async { + loop { + if track + .subscription() + .is_some_and(|subscription| subscription.max_delay == Duration::from_millis(millis)) + { + return; + } + track.subscription_changed().await.unwrap(); + } + }) + .await + .unwrap_or_else(|_| panic!("{version}: wanted {millis}ms, publisher has {:?}", track.subscription())); +} + +#[moq_clock::test] +async fn relay_forwards_the_most_tolerant_subscriber_max_age() { + for draft in 14..=22 { + let version: Version = format!("moq-transport-{draft}").parse().unwrap(); + let publisher = origin(1); + let relay = origin(2); + let viewer = origin(3); + let _upstream = peer(version, &publisher, &relay).await; + let _downstream = peer(version, &relay, &viewer).await; + let broadcast = publisher.create_broadcast("live").unwrap(); + let mut source = broadcast + .create_track( + "catalog", + track::Info::default().with_timescale(moq_net::Timescale::MILLI), + ) + .unwrap(); + let mut group = source.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"catalog").unwrap(); + group.finish().unwrap(); + broadcast.announce(Default::default()).unwrap(); + let consumer = viewer.consume(); + consumer.routed("live").await.unwrap(); + let remote = consumer.request_broadcast("live", None).await.unwrap(); + let cache = remote.track("catalog").unwrap(); + let prefs = |millis| track::Subscription::default().with_max_delay(Duration::from_millis(millis)); + let mut short = cache.subscribe(prefs(250)).await.unwrap(); + budget(&mut source, 250, version).await; + let tolerant = cache.subscribe(prefs(2000)).await.unwrap(); + budget(&mut source, 2000, version).await; + short.update(prefs(10)).unwrap(); + // Drive the logical cursor so its serving copy observes the new preferences. + let mut snapshot = short.recv_group().await.unwrap().unwrap(); + // Dropping the tolerant subscriber exposes the updated shorter budget, across both hops. + drop(tolerant); + budget(&mut source, 10, version).await; + // A quiet catalog survives far longer than the budget because it has no successor. + moq_clock::sleep(Duration::from_secs(1)).await; + assert_eq!(snapshot.sequence, 0, "{version}"); + assert_eq!( + snapshot.read_frame().await.unwrap().unwrap().payload, + b"catalog".as_slice() + ); + } +} diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index db2eaa1ab9..69ac35ea3d 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -1372,3 +1372,136 @@ async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq moq_net::announce::Event::End(route) => Some((route, false)), } } + +/// A finite deep queue, below the media bitrate, must shed superseded groups while +/// preserving the newest catalog snapshot. Random loss is disabled in this acceptance test. +#[tokio::test] +async fn ietf_subscriber_max_age_skips_bufferbloat() { + let version: moq_net::Version = "moq-transport-22".parse().unwrap(); + let mut config = relay_config(); + config.listen.version = vec![version]; + let relay = RelayHost::start(config).await; + let path = Path::through( + &relay, + moq_shaper::Profile { + rate: Some(moq_shaper::Rate { + bits_per_second: 64_000, + burst: 1500, + queue: Duration::from_millis(500), + }), + ..Default::default() + }, + 19, + ) + .await; + let publisher = moq_tokio::origin::spawn(); + let broadcast = publisher.create_broadcast("live").unwrap(); + let mut track = broadcast.create_track(TRACK, timed()).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let mut dial = dial(); + dial.version = vec![version]; + let publish_url = relay.url(); + let publish_session = dial + .clone() + .init(quic()) + .unwrap() + .with_publisher(&publisher) + .with_reconnect(false) + .connect(publish_url) + .established() + .await + .unwrap(); + let subscriber = moq_tokio::origin::spawn(); + let consumed = subscriber.consume(); + let subscribe_session = dial + .init(quic()) + .unwrap() + .with_subscriber(subscriber) + .with_reconnect(false) + .connect(path.url.clone()) + .established() + .await + .unwrap(); + write_group(&mut track, b"warmup"); + let remote = tokio::time::timeout(TIMEOUT, consumed.routed_broadcast("live")) + .await + .unwrap() + .unwrap(); + let mut groups = remote + .track(TRACK) + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_millis(100))) + .await + .unwrap(); + let mut warmup = groups.recv_group().await.unwrap().unwrap(); + assert_eq!( + warmup.read_frame().await.unwrap().unwrap().payload, + b"warmup".as_slice() + ); + tokio::time::timeout(TIMEOUT, async { + loop { + if track + .subscription() + .is_some_and(|subscription| subscription.max_delay == Duration::from_millis(100)) + { + break; + } + track.subscription_changed().await.unwrap(); + } + }) + .await + .expect("subscriber max age did not reach the publisher"); + + // The initial congestion window exceeds this queue's 5.5KB allowance, so this + // gate proves overflow without depending on how the OS schedules a packet burst. + let burst = path + .shaper + .as_ref() + .unwrap() + .burst(moq_shaper::ingress::Direction::Down) + .unwrap(); + let payload = vec![42u8; 32_000]; + write_group(&mut track, &payload); + tokio::time::timeout(TIMEOUT, burst.completed()) + .await + .expect("rate-limit queue never filled") + .unwrap(); + assert!(path.overflowed() > 0, "the finite queue did not drop overflow"); + // 5.12Mb/s of media behind a 64kb/s link, for a second. + for sequence in 2..22 { + let mut group = track.append_group().unwrap(); + group + .write_frame(moq_net::Timestamp::from_millis(sequence * 50).unwrap(), payload.clone()) + .unwrap(); + group.finish().unwrap(); + tokio::time::sleep(Duration::from_millis(50)).await; + } + let mut newest = track.append_group().unwrap(); + newest + .write_frame(moq_net::Timestamp::from_millis(1100).unwrap(), b"snapshot") + .unwrap(); + newest.finish().unwrap(); + // No successor means this snapshot must survive even after its budget elapsed. + tokio::time::sleep(Duration::from_millis(500)).await; + let sequence = tokio::time::timeout(TIMEOUT, async { + loop { + let mut group = groups.recv_group().await.unwrap().expect("subscription ended"); + if let Ok(Some(frame)) = group.read_frame().await + && frame.payload == b"snapshot".as_slice() + { + break group.sequence; + } + } + }) + .await + .expect("subscriber did not skip the backlog to the live edge"); + assert_eq!(sequence, 22); + println!( + "max age: newest group {sequence}, {} queue-overflow drops", + path.overflowed() + ); + drop(groups); + drop(subscribe_session); + drop(publish_session); + path.verify(); +} diff --git a/test/drill/README.md b/test/drill/README.md index 6a896b45a2..83b62348e4 100644 --- a/test/drill/README.md +++ b/test/drill/README.md @@ -32,6 +32,7 @@ checks that what the fault owned was released. | `bursts_cross_a_cluster` | Bursts of one-frame groups cross from an origin relay to a clustered edge relay; the subscriber reads newest-only and FETCHes every group it misses | how many groups missed the live subscription, which live groups were reset, the completed ingress burst and overflow count, and the slowest group's latency | every group arrives, live or by FETCH, within 10s of being written; a lost group fails by name (FETCH failed, FETCH unanswered, live group stalled), and the newest group comes live | | `bursts_cross_a_flapping_peer` | The same bursts, with the peer link cut as the second burst is written and restored once the edge withdraws the route | how long the link was down before the withdrawal, and the reads and FETCHes the flap failed | the route comes back and every group still arrives within 10s of being written, or of the route's return when it arrives after it; only a failure the flap caused is retried, on the restored route, so a hang on the dead one past the relays' idle timeout or a withdrawal with the link up still fails | | `bursts_cross_a_transport_peer` | The same bursts, with the relays peering over `moq-transport-19`, so the edge recovers gaps with IETF FETCHes | as `bursts_cross_a_cluster` | as `bursts_cross_a_cluster`: a group the origin still holds is served, never refused | +| `ietf_subscriber_max_age_skips_bufferbloat` | moq-transport-22 media at 5.12Mb/s crosses a 64kb/s link behind a finite 500ms queue, with no random loss | the propagated 100ms budget and forced queue overflow | superseded groups are skipped and the newest catalog snapshot survives its budget | | `no_publisher_never_delivers` | none: the negative control | - | with nothing publishing, nothing is announced and no broadcast resolves | A moq publisher is never blocked by a slow reader: `write_frame` is synchronous diff --git a/test/drill/mutations/subscriber-max-age-ignored.patch b/test/drill/mutations/subscriber-max-age-ignored.patch new file mode 100644 index 0000000000..40d5e93dbe --- /dev/null +++ b/test/drill/mutations/subscriber-max-age-ignored.patch @@ -0,0 +1,16 @@ +# drill: ietf_subscriber_max_age_skips_bufferbloat +# expect: subscriber max age did not reach the publisher +# +# Preserve negotiation and decoding, but ignore the requested budget when serving. +--- a/rs/moq-net/src/ietf/publisher.rs ++++ b/rs/moq-net/src/ietf/publisher.rs +@@ -1220,9 +1220,6 @@ + // A live SUBSCRIBE's NEW_GROUP_REQUEST asks nothing more: subscribing is itself + // a join, which the track's refresh ratio judges. + let mut subscription = serving_subscription(msg.subscriber_priority).with_order(order); +- if let Some(max_age) = msg.max_age { +- subscription.max_delay = Duration::from_millis(max_age); +- } + // A range that resolves without the live edge has a fixed start, which is no + // join: bound the subscription before subscribing so the ratio never judges it. + // From the start of its group, so a relay's copy holds the group's head for