diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 6df35b1a810..7ea2af6ea11 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -775,7 +775,7 @@ class MoqSubscription { final MoqOrder? order; MoqSubscription({ this.priority = 0, - this.maxDelay = null, + this.maxDelay, this.ranges = null, this.order = null, }); diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 4a40be8e049..75fac1d3423 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -51,7 +51,7 @@ publication.announce(); - **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. The application owns sequence numbers: the broadcast keeps nothing of an ended track, so a track that replaces it, created or requested, starts at 0. A name always means the same content, so the replacement continues past its predecessor's groups by writing its first group with an explicit sequence. - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Omitting it means untimed, as in Rust. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` producers. - **Subscriber staleness is `maxDelay`.** It counts from a group's successor, by media time and by wall clock, whichever reaches it first. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`. -- **Subscription defaults.** `Track.Subscription.default()` returns the default subscription, whose zero `maxDelay` skips a superseded group at once; calling `subscribe()` with no options means the same. Spread it to change a field (`{ ...Track.Subscription.default(), priority: 2 }`) rather than passing a bare options object, whose omitted `maxDelay` sets no subscriber limit. +- **Subscription defaults.** `Track.Subscription.default()` returns the default subscription, whose zero `maxDelay` skips a superseded group at once; calling `subscribe()` with no options means the same. An omitted `maxDelay` means the same zero; `null` sets no subscriber limit. - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts). - **Authorization is in band.** On moq-lite 07 (opt in with `versions: ["moq-lite-07-wip", "moq-lite-06"]`, listing every fallback you still want) and on moq-transport draft-18+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription or track listing the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. On moq-transport, `auth.add(token)` keeps the token even without the extension, and a broadcast announced with the `Auth.Credential` it returns (`publication.announce({ auth })`) carries it on that announcement's `PUBLISH_NAMESPACE`; announcing again with another renews the request. A `Connection`'s `auth` handle (a getter on a pooled connection, following its URL) outlives its sessions: a token added there is presented on every reconnect and keeps riding the announcements that name it, with no re-add or re-announce. A one-shot session's `auth` ends with it, unless `Connection.connect` was given a `new Auth.Handle()` as `auth`. - **A graceful close waits.** `await connection.close()` withdraws announcements and drains for up to one second. Application delivery requires explicit `moq-lite-07-wip`; older Lite versions wait only for transport ACKs. IETF media streams are not drained. `abort()` ends immediately. diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 5274d995abf..d6d2b9d35da 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -112,7 +112,7 @@ shows which module holds each. - **Tokens.** Pass `ClientConfig(tokens = listOf(...))` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. `moq.tokens().set(listOf(...))` replaces them before they expire: the next dial presents them, a live AUTH session at once, and after an authorization refusal the session redials only if they changed. Suspending `tokens().answered()` returns once the relay accepts them and throws its refusal with the relay's code. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode. -- **Subscription defaults.** `Subscription.default()` returns moq-net's defaults, whose zero `maxDelay` skips a superseded group at once; passing no subscription means the same. `copy` it rather than starting from `Subscription()`, whose null `maxDelay` sets no subscriber limit. +- **Subscription defaults.** `Subscription.default()` returns moq-net's defaults, whose zero `maxDelay` skips a superseded group at once; passing no subscription means the same. `copy` it to change a field; building a `Subscription(...)` must name `maxDelay`, where null sets no subscriber limit. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. ## Runtime diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 6e089b5d209..cbd2f82acf0 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -128,7 +128,7 @@ shows which module holds each. - **Tokens.** Pass `tokens=[...]` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. `session.tokens().set([...])` replaces a client session's tokens before they expire: the next dial presents them, a live AUTH session at once, and after an authorization refusal the session redials only if they changed. `await tokens.answered()` returns once the relay accepts them and raises its refusal with the relay's code. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode. -- **Subscription defaults.** `moq.Subscription.default()` returns moq-net's defaults, whose zero `max_delay` skips a superseded group at once; passing no subscription means the same. `dataclasses.replace` it rather than starting from `moq.Subscription()`, whose `None` `max_delay` sets no subscriber limit. +- **Subscription defaults.** `moq.Subscription.default()` returns moq-net's defaults, whose zero `max_delay` skips a superseded group at once; passing no subscription means the same. A bare `moq.Subscription()` has the same defaults; an explicit `max_delay=None` sets no subscriber limit. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. ## Runtime diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 11da386d742..21b9195faf8 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -111,7 +111,7 @@ shows which module holds each. - **Tokens.** Pass `Client(tokens: [...])` instead of putting the credential in the URL. The first is the connection credential, sent in SETUP and, while an offered version can't carry it there (lite before 07), as the URL's `?jwt=`; each other token widens the grant over AUTH. `try session.tokens()?.set([...])` replaces them before they expire: the next dial presents them, a live AUTH session at once, and after an authorization refusal the session redials only if they changed. `try await tokens.answered()` returns once the relay accepts them and throws its refusal with the relay's code. A `?jwt=` already in the URL stays the connection credential, which is how to reach a relay older than this release. - **Errors.** Diagnostic cases use the `reason:` argument label; match their associated value to inspect it. Protocol errors carry `details`. - **Disabled renditions.** A rendition with `enabled: false` has no frames coming until its publisher re-enables it. Select an enabled rendition from the latest catalog before opening a decoder. Keep one already playing while every rendition of its kind is disabled: the convenience decoders accept a disabled configuration, and a re-enable, such as an unmute, then plays without a resubscribe. Among the video renditions you can decode, prefer the highest `preference`; `0` is the default and `-1` marks a compatibility transcode. -- **Subscription defaults.** `Subscription.default` holds moq-net's defaults, whose `maxDelay` of 0 skips a superseded group at once; passing no subscription means the same. Copy it and change a field rather than starting from `Subscription()`, whose `nil` `maxDelay` sets no subscriber limit. +- **Subscription defaults.** `Subscription.default` holds moq-net's defaults, whose `maxDelay` of 0 skips a superseded group at once; passing no subscription means the same. Copy it to change a field; building a `Subscription(...)` must name `maxDelay`, where `nil` sets no subscriber limit. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. ## Runtime diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index df2e31601ec..5f5894f0bc0 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -4,20 +4,9 @@ How to update your code for each breaking change, newest first. ## Unreleased -### An omitted `maxDelay` sets no subscriber limit ([#7077](https://github.com/moq-dev/moq/pull/7077)) +### `maxDelay: null` sets no subscriber limit ([#7403](https://github.com/moq-dev/moq/pull/7403)) -A `Track.Subscription` without `maxDelay` now reaches back as far as the publisher's `Track.Info.maxAge` keeps, as `None` does in Rust, instead of meaning zero. -`subscribe()` with no options still uses the default subscription, now spelled `Track.Subscription.default()`, whose zero `maxDelay` skips a superseded group at once. -A live reader that passed options without `maxDelay` spreads the default to keep its behavior: - -```ts -// Before -const sub = track.subscribe({ priority: 2 }); -// After -const sub = track.subscribe({ ...Track.Subscription.default(), priority: 2 }); -``` - -The same applies to `Subscriber.update(options)`, which replaces every field. +A `Track.Subscription` whose `maxDelay` is `null` reaches back as far as the publisher's `Track.Info.maxAge` keeps, as `None` does in Rust. An omitted or `undefined` `maxDelay` is still zero. ### Paths fit moq-transport's 4096-byte namespace ([#6927](https://github.com/moq-dev/moq/pull/6927)) diff --git a/js/net/api/index.api.md b/js/net/api/index.api.md index 651f65d387b..841422bf7bc 100644 --- a/js/net/api/index.api.md +++ b/js/net/api/index.api.md @@ -660,7 +660,7 @@ export namespace Track { }; export interface Subscription { groups?: Track.Groups; - maxDelay?: Milli; + maxDelay?: Milli | null; order?: Track.Order; priority?: number; } diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 7110321dd90..11fd75c881f 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -26,6 +26,7 @@ import { groupBounds, hiddenBelow, hooks, + maxDelayOf, scopeCaptures, scopeHead, scopeOverlaps, @@ -1087,7 +1088,7 @@ export class Subscriber { // keeps what it last asked for until one returns, as on lite and in Rust. const subscription = producer.subscription.peek(); const wanted = toWire(subscription?.priority ?? 0); - const wantedMaxAge = this.#maxAge ? subscription?.maxDelay : undefined; + const wantedMaxAge = this.#maxAge && subscription ? maxDelayOf(subscription) : undefined; if (!subscription || (wanted === sent && wantedMaxAge === sentMaxAge) || unanswered.peek() > 0) { const disposes: Dispose[] = []; const wake = new Promise((resolve) => { diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index e7c978a1151..0d4fcadaaf1 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -5,7 +5,7 @@ * * @module */ -import type { Instant, Milli } from "@moq/clock"; +import { type Instant, Milli } from "@moq/clock"; import type { Dispose, GetPromise, Getter } from "@moq/signals"; import type { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import type { Datagram } from "./datagram.ts"; @@ -18,8 +18,18 @@ import type { Request as OriginRequest } from "./origin.ts"; import * as Path from "./path.ts"; import type { Timestamp } from "./time.ts"; import type { Demand, Groups, Info, Position, Producer, Request, Subscriber, Subscription } from "./track.ts"; +import * as Varint from "./varint.ts"; import type { Advertised, Advertisements } from "./wire.ts"; +/** What a null maxDelay goes out as: the largest value every peer can read, as Rust sends for `None`. */ +export const NO_MAX_DELAY = Milli(Varint.MAX_U53); + +/** A subscription's budget as the wire carries it: null for no limit, omitted for zero. */ +export function maxDelayOf(subscription: Pick): Milli { + if (subscription.maxDelay === null) return NO_MAX_DELAY; + return subscription.maxDelay ?? Milli.zero; +} + /** Normalize public group bounds into an inclusive start and exclusive end. */ export function groupBounds(groups: Groups = {}): { start: number; end?: number } { const bound = (value: Groups["start"] | undefined, start: boolean): number | undefined => { diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 3d2f41fb474..6f47746e89d 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -22,7 +22,7 @@ import { } from "../error.ts"; import * as netGroup from "../group.ts"; import { Cost, type Hop, MAX_HOPS, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts"; -import { groupBounds, hiddenBelow, hooks, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; +import { groupBounds, hiddenBelow, hooks, maxDelayOf, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; import { grace } from "../ledger.ts"; import type { Lister } from "../listing.ts"; import * as Path from "../path.ts"; @@ -931,7 +931,7 @@ export class Subscriber { track: request.name, priority: subscription.priority ?? 0, order: subscription.order, - maxDelay: subscription.maxDelay, + maxDelay: maxDelayOf(subscription), startGroup: subscription.groups?.start === undefined ? undefined : bounds.start, endGroup: inclusiveGroupEnd(bounds.end), }); @@ -1650,7 +1650,7 @@ export class Subscriber { wantedGroup: refresh ?? sentWanted, priority: current.priority ?? 0, order: current.order, - maxDelay: current.maxDelay, + maxDelay: maxDelayOf(current), startGroup: current.groups?.start === undefined ? undefined : bounds.start, endGroup: inclusiveGroupEnd(bounds.end), }); diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 04835a54878..c4c4b54b022 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -826,7 +826,7 @@ test("zero latency takes the latest group when ages are equal", async () => { } }); -test("the default subscription skips superseded groups and a bare options object sets no limit", async () => { +test("an omitted maxDelay is the default zero, and null sets no limit", async () => { expect(Subscription.default()).toEqual({ priority: 0, maxDelay: Milli.zero, order: "descending" }); const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI, maxAge: Milli(5000) }); @@ -837,11 +837,15 @@ test("the default subscription skips superseded groups and a bare options object producer.writeFrame({ payload: enc.encode(`${timestamp}`), timestamp: Timestamp.fromMillis(timestamp) }); } - // No options means the default subscription, whose zero budget keeps only the newest group. + // Each spelling of the default keeps only the newest group. expect((await explicit.recvGroup())?.sequence).toBe(2); expect((await implicit.recvGroup())?.sequence).toBe(2); - // An omitted maxDelay leaves the publisher's retention as the only bound. - for (const sequence of [0, 1, 2]) expect((await bare.recvGroup())?.sequence).toBe(sequence); + expect((await bare.recvGroup())?.sequence).toBe(2); + expect(producer.subscription.peek()?.maxDelay).toBe(Milli.zero); + + // An explicit null leaves the publisher's retention as the only bound. + const unlimited = producer.subscribe({ maxDelay: null }); + for (const sequence of [0, 1, 2]) expect((await unlimited.recvGroup())?.sequence).toBe(sequence); expect(producer.subscription.peek()?.maxDelay).toBe(Milli(5000)); producer.finish(); producer.close(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 4c52b607ddd..ecac970601a 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -30,6 +30,7 @@ import { insertRun, type Land, MAX_DATAGRAMS, + NO_MAX_DELAY, type Recv, type RecvDatagram, type ResumeSource, @@ -41,7 +42,6 @@ import { } from "./internal.ts"; import { grace, Ledger } from "./ledger.ts"; import { Timescale, type Timestamp } from "./time.ts"; -import * as Varint from "./varint.ts"; import { type Broadcast as BroadcastWire, registerTrackConsumer } from "./wire.ts"; export type { Datagram } from "./datagram.ts"; @@ -227,14 +227,16 @@ export interface Subscription { /** * How long (milliseconds) a non-latest group stays worth delivering, by the same two * clocks as {@link Info.maxAge}: it is skipped once its successor arrived this long ago - * or starts this far behind the live edge. Omitted means no subscriber limit, so the - * publisher's {@link Info.maxAge} alone bounds it, which is what a reader reaching back for - * history wants. {@link Subscription.default} sets `0`, the live edge only, timed or not. + * or starts this far behind the live edge, and never past the publisher's + * {@link Info.maxAge}. Omitted (or `undefined`) means `0`, the live edge only, timed or + * not, as in {@link Subscription.default}. `null` sets no subscriber limit, so the + * publisher's {@link Info.maxAge} alone bounds it, which is what a reader reaching back + * for history wants; read in order, see {@link Ordered} for where that can block. * Rounded up to a whole millisecond, so a value derived from a measurement is never * shortened. A negative or non-finite value, or one past `Number.MAX_SAFE_INTEGER` * after rounding, is refused. */ - maxDelay?: Milli; + maxDelay?: Milli | null; /** * The groups this reader may receive, with an optional floor and cap. * @@ -281,10 +283,6 @@ export const Subscription: { default: () => ({ priority: DEFAULT_PRIORITY, maxDelay: Milli.zero, order: "descending" }), }; -// What an omitted maxDelay materializes to: the publisher's max age alone. The largest -// value every peer can read, as the Rust wire sends for `None`. -const NO_MAX_DELAY = Milli(Varint.MAX_U53); - // Materialize the defaults at the model boundary so every layer observes a complete // subscription rather than interpreting an omitted field differently. function subscriptionDefaults(subscription: Subscription = Subscription.default()): Subscription { @@ -292,7 +290,8 @@ function subscriptionDefaults(subscription: Subscription = Subscription.default( const bounds = groupBounds(subscription.groups ?? {}); return { priority: priorityValue(subscription.priority ?? DEFAULT_PRIORITY), - maxDelay: wireMillis("maxDelay", subscription.maxDelay ?? NO_MAX_DELAY), + maxDelay: + subscription.maxDelay === null ? NO_MAX_DELAY : wireMillis("maxDelay", subscription.maxDelay ?? Milli.zero), order: orderOf(subscription.order), groups: { start: subscription.groups?.start === undefined ? undefined : { included: bounds.start }, @@ -2977,6 +2976,14 @@ export class Subscriber { * inside the budget is still delivered whole, as a burst in order. The budget follows a * group already handed out too: a stalled group's pending frame read rejects once newer * content has pulled that far ahead. + * + * A history or VOD read sets `maxDelay: null` (no limit). This reader skips a missing + * group rather than waiting for it, but the Rust reader waits for one a subscription still + * owes, within the budget, or until it resolves without one. An unlimited or long budget + * only avoids blocking where the feed signals its gaps: moq-lite, whose publisher sends + * `SUBSCRIBE_DROP` for every group it won't deliver, or an ordered moq-transport FETCH. + * Over a moq-transport SUBSCRIBE, which signals no gaps, that wait can last until the + * track ends. */ export class Ordered { /** The track name. */ diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index de3a9f1986e..f5324b79443 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -32,7 +32,7 @@ Their handles `OriginDynamic`, `BroadcastDynamic`, and `TrackDynamic` are `Origi The records carry durations as `java.time.Duration` fields instead of `*Us` microsecond integers, so the duration extension properties are gone: read `backoff.initial`, `stats.rtt`, `grant.expires`, `info.maxAge`, `fill.maxDelay()`, and `consumer.delay()` directly. Convert with `toKotlinDuration()` and `toJavaDuration()` where you want `kotlin.time.Duration`. `AudioEncoderOutput.frameDuration` is nullable: pass null where you passed `frameDurationUs = 0u` for the codec's own frame. -`Subscription.maxDelay` defaults to null, which sets no subscriber limit: start from `Subscription.default()`, or pass `java.time.Duration.ZERO`, for the old default. +`Subscription.maxDelay` has no default: start from `Subscription.default()`, or pass `java.time.Duration.ZERO`, for the old default. Null sets no subscriber limit. `TrackConsumer.info()` no longer throws. ### Connection state diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index 82456e1e7fc..34e67c0b6e9 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -115,8 +115,8 @@ typealias Subscription = uniffi.moq.MoqSubscription /** * moq-net's defaults, whose `maxDelay` of zero skips a superseded group at once. A null - * subscription means the same; `copy` this one to change a field. A `Subscription()` leaves - * `maxDelay` null, which sets no subscriber limit. + * subscription means the same; `copy` this one to change a field. A `Subscription(...)` + * must name `maxDelay`, where null sets no subscriber limit. */ fun uniffi.moq.MoqSubscription.Companion.default(): Subscription = uniffi.moq.moqSubscriptionDefaults() diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index 122e13032a1..f96a36d52db 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -279,7 +279,7 @@ class SmokeTest { @Test fun `a default subscription skips superseded groups at once`() { assertEquals(java.time.Duration.ZERO, Subscription.default().maxDelay) - assertNull(Subscription().maxDelay) + assertNull(Subscription(maxDelay = null).maxDelay) } @Test diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index 23415be0c33..4e0326e3bf5 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -277,7 +277,6 @@ Pass `timestamp=` on every write to a timed track. ### Subscriber staleness is `max_delay` `Subscription.max_age_us` is `max_delay`, a `timedelta`, and the decoder outputs take `max_delay`, also a `timedelta`. -`Subscription.max_delay` defaults to `None`, which sets no subscriber limit: start from `Subscription.default()` for the old default of zero. `TrackInfo` keeps its `max_age`: it is the publisher's retention. ### Error payloads are named `reason` diff --git a/py/moq-rs/moq/_records.py b/py/moq-rs/moq/_records.py index 8b22335a972..1a4586a1776 100644 --- a/py/moq-rs/moq/_records.py +++ b/py/moq-rs/moq/_records.py @@ -117,12 +117,11 @@ class Subscription: to deliver, ``None`` for everything ``max_delay`` still allows. ``order`` is which pending group the publisher sends first, ``None`` for newest first. - Start from :meth:`default` for moq-net's defaults; a bare ``Subscription()`` sets no - max delay. + A bare ``Subscription()`` has moq-net's defaults, as :meth:`default` does. """ priority: int = 0 - max_delay: timedelta | None = None + max_delay: timedelta | None = timedelta(0) ranges: Sequence[MoqRange] | None = None order: MoqOrder | None = None diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 992a0387581..3efedb002bc 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -482,9 +482,10 @@ def test_track_info_keeps_native_default_priority(): def test_default_subscription_skips_superseded_groups_at_once(): - """Only the default carries moq-net's zero max delay; a bare record sets no limit.""" + """A bare record carries moq-net's zero max delay too; only an explicit None sets no limit.""" assert moq.Subscription.default()._ffi().max_delay == timedelta(0) - assert moq.Subscription()._ffi().max_delay is None + assert moq.Subscription()._ffi().max_delay == timedelta(0) + assert moq.Subscription(max_delay=None)._ffi().max_delay is None def test_string_sequences_refuse_a_bare_string(): diff --git a/quest/m2/README.md b/quest/m2/README.md index ea2bba57db2..77c2b2ecdc6 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -90,7 +90,6 @@ lite-07; bitrate enforcement is not part of this release. - [Shared group serving and ingest in JS](/quest/m2/shared-serve-js.md) - the same split in js/net, following the Rust seam - [Paired bench BASE](/quest/m2/bench-base-paired.md) - `just bench ` uses the nightly's paired comparison and the old one is deleted - [Read-only lookup](/quest/m2/read-only-lookup.md) - a live track lookup stops waking the front and every demand watcher -- [Ordered owed wait](/quest/m2/ordered-owed-wait.md) - an ordered reader waits within max_delay for a missing group only while the track still owes it - [lite-05/06 gap judgement](/quest/m2/lite-legacy-gap.md) - a resumed copy on lite-05/06 never reads a descending track's newest group as its start - [Refill evicted groups](/quest/m2/evicted-refill.md) - a group evicted after a reader judged it but before it took it is refilled, never delivered twice, and guarded against thrash - [Receive pool](/quest/m2/receive-pool.md) - one byte budget per worker covers every connection's receive credit and the cache, billing each frame once diff --git a/quest/m2/ordered-owed-wait.md b/quest/m2/ordered-owed-wait.md deleted file mode 100644 index 064a7530655..00000000000 --- a/quest/m2/ordered-owed-wait.md +++ /dev/null @@ -1,75 +0,0 @@ -# [S] Ordered reads wait only on owed groups - -## Goal - -`track::Ordered` waits within `max_delay` for a missing group only while that -sequence is still owed, and moves past it as soon as the track knows it was -dropped, evicted, or skipped. A reader that got every group behaves as today. -No new public API. - -## Plan - -Today a gap costs an ordered reader nothing: it seeks to the lowest cached -group in range, so a group reordered behind its successor is skipped even -when it would have arrived in budget. - -The shelved [#5475](https://github.com/moq-dev/moq/pull/5475) (branch -`codex/ordered-gap-delay`, kept for reference) waited on every missing -sequence until the budget elapsed. The model cannot tell "not yet" from -"never", so it hung or stalled readers; see its -[iteration comment](https://github.com/moq-dev/moq/pull/5475#issuecomment-6173264946). - -✅ Decided 2026-10-14 (maintainer): - -- Wait only on sequences the track still owes, read from the dispositions - that SUBSCRIBE_DROP and the track's arrival record - (`TrackState::arrivals`) keep. Not chosen: an opt-in - waiting reader (new public API no consumer needs, and opted-in readers hit - the same holes); fixing #5475 forward (discontinuity markers still stall, - and unlimited-budget readers still hang on a never-produced group). -- m2. The media path (`moq-mux`'s container consumer) already waits for - reordered groups within `max_delay`. The archive writer keeps its own - reader (`rs/moq-archive/src/producer/ordered.rs`, from - [moq-dev/moq#6160](https://github.com/moq-dev/moq/pull/6160)) that waits - within `max_delay` for a late group or datagram, since `Ordered` skips gaps. - -Open: what "owed" means on a locally produced track, which gets no drops or -dispositions. The motivating case was the archive's local timeline, read in -order with an unlimited budget; [moq-dev/moq#6262](https://github.com/moq-dev/moq/pull/6262) -removed that timeline, so no current reader motivates it. - -- Only a subscription owes sequences, so local tracks keep today's ordered - reads (recommended: no local reader needs to wait, and an unlimited-budget - local reader can never hang on a hole). -- A local producer's unreleased sequences also count as owed, until it - writes, skips, or aborts them. Every local skip must then be marked, or an - unlimited-budget local reader hangs. - -Watch for: - -- Untimed tracks: the budget can never expire a group, so a hole nothing - accounts for (a lost datagram, say) must not block forever. -- Deliberate skips: `Producer::discontinuity` and TS passthrough's flagged - jumps skip a sequence on purpose. A reader passes them at once, not after - `max_delay`. -- Eviction, expiry, and an aborted group whose slot was reclaimed leave no - record today; the wait needs the disposition to survive them. -- Readers with unlimited budgets: an evicted or never-produced group must not - hang them. On #5475 this failed in `moq-hls` history mode, which - [moq-dev/moq#6517](https://github.com/moq-dev/moq/pull/6517) removed. - -Reuse from #5475: its reorder regressions (exact deadline and wakeup, explicit -starts, capped reads, a changed budget), and its snapshot-reader change: -snapshot readers pick the latest group explicitly, since each group replaces -the value, instead of relying on `Ordered` skipping gaps. Add the iteration -comment's two cases: an untimed track with groups 0 and 2..10 keeps -delivering, and an aborted and reclaimed group 1 does not stall group 2 for -`max_delay`. - -Update `Ordered`'s doc, which says a gap costs nothing. - -Public API: behavior only. Wire: none. - -## Related - -- [Read backfill](/quest/m1/read-backfill.md) - container reads drop groups below the cursor diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index 8b64ade79f1..93ae354dd37 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -42,7 +42,8 @@ It defaults to empty in every generated binding; a Rust struct literal adds `tok Every duration is a UniFFI `Duration` instead of an integer microsecond field, and drops its `_us` suffix: each language receives its own duration type (`timedelta`, `TimeInterval`, `java.time.Duration`, `time.Duration`, Dart `Duration`, C++ `std::chrono` nanoseconds). The renamed fields are `MoqClientConfig.{timeout, race, resolution_delay}`, `MoqQuicConfig.{idle_timeout, keep_alive}`, `MoqWebSocketConfig.delay`, `MoqBackoff.{initial, max, timeout}`, `MoqConnectionStats.rtt`, `MoqGrant.expires`, `MoqTrackInfo.max_age`, `MoqSubscription.max_delay`, `MoqFetchGroupOptions.max_delay`, `MoqAudio.jitter`, `MoqAudioDecoderOutput.{max_delay, delay}`, and `MoqVideoDecoderOutput.max_delay`; the methods are `MoqFill::max_delay()` and `MoqAudioConsumer::delay()`. `MoqAudioEncoderOutput.frame_duration_us: u32` is `frame_duration: Option`: pass null where you passed 0 for the codec's own frame. -`MoqSubscription.max_delay` defaults to null, which sets no subscriber limit: a record that omits it no longer gets moq-net's 0. +`MoqSubscription.max_delay` has no default, since UniFFI can't default a duration to 0: Kotlin and Swift must name it, `0` for moq-net's default or null for no subscriber limit. +Dart, Go, and C can still omit it (Dart's optional parameter, Go's nil `*time.Duration`, a zeroed C struct), which means no subscriber limit; name it there too. Start from `moq_subscription_defaults()` where you relied on that default; a null subscription argument still means it. `MoqTrackConsumer::info()` no longer fails. A duration handed out saturates at `i64::MAX` nanoseconds (about 292 years), so C, C++, and Go hold every value. diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index acef58d88a1..eb0077e77bd 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -25,8 +25,9 @@ fn raw_frame(frame: moq_net::frame::Frame) -> Result { /// Subscriber-side delivery preferences, mirroring [`moq_net::track::Subscription`]. /// /// Start from [`moq_subscription_defaults`] for moq-net's defaults (priority 0, a `max_delay` -/// of 0, everything still fresh, newest first) and change the fields you care about. A record -/// built field by field leaves out `max_delay`, which means no subscriber limit. +/// of 0, everything still fresh, newest first) and change the fields you care about. Kotlin and +/// Swift make a record built field by field name `max_delay`; Dart, Go, and C can still omit it, +/// which means no subscriber limit. #[derive(Clone, uniffi::Record)] pub struct MoqSubscription { /// Delivery priority; higher values preempt lower ones under bandwidth contention. @@ -34,12 +35,12 @@ pub struct MoqSubscription { #[uniffi(default = 0)] pub priority: i8, /// Maximum age of a non-latest group before it is skipped. - /// `0` skips immediately; a larger value tolerates that much reordering; null, the - /// default, sets no subscriber limit and leaves the publisher's max age alone to bound it. + /// `0` skips immediately; a larger value tolerates that much reordering; null sets no + /// subscriber limit and leaves the publisher's max age alone to bound it. Required, + /// since uniffi can't default a duration to `0`. /// /// Enforced both by the publisher's cache (sent on the wire) and by any local /// buffering, such as [`MoqMediaTrackConsumer`]'s jitter buffer. It caps every range. - #[uniffi(default = None)] pub max_delay: Option, /// The spans of the track to deliver, or null for everything `max_delay` still /// considers fresh. An empty list asks for nothing. diff --git a/rs/moq-json/src/snapshot/consumer.rs b/rs/moq-json/src/snapshot/consumer.rs index 19a4f653ed2..72b20c52628 100644 --- a/rs/moq-json/src/snapshot/consumer.rs +++ b/rs/moq-json/src/snapshot/consumer.rs @@ -67,6 +67,8 @@ impl Consumer { } /// Skip to the newest buffered state and its timestamp, or wait for a new one. + /// + /// See [`Self::poll_latest`] for how a nonzero `max_delay` delays it. pub async fn latest(&mut self) -> Result>> where T: Unpin, @@ -78,6 +80,11 @@ impl Consumer { /// /// Frames inside the newest group are still decoded in order. What this skips is gone; /// a subsequent `next` resumes after it. An untimed frame keeps `at: None`. + /// + /// It reads groups in order, so on a received track with a nonzero `max_delay` it first + /// waits for a lower group the subscription still owes, up to that budget, and with no + /// limit until the gap resolves, even when a newer group is buffered. A reader that only + /// wants the newest state keeps the default zero budget. pub fn poll_latest(&mut self, waiter: &kio::Waiter) -> Poll>>> { self.poll_read(waiter, true) } diff --git a/rs/moq-net/src/model/subscription.rs b/rs/moq-net/src/model/subscription.rs index 2edda6dea91..c75b4945d55 100644 --- a/rs/moq-net/src/model/subscription.rs +++ b/rs/moq-net/src/model/subscription.rs @@ -28,7 +28,8 @@ pub struct Subscription { /// giving up. This never *adds* delay, since the bound is only reached once newer data /// is already that far ahead. `None` is the publisher's /// [`Info::max_age`](crate::track::Info::max_age) alone, what a reader reaching back - /// for history wants. + /// for history wants. Read in order, `None` waits at a missing group until it resolves; + /// see [`track::Ordered`](crate::track::Ordered) for where that can block. /// /// This is the `Subscriber Max Age` on the wire, and it is stored here verbatim so /// what was asked for stays readable. Encoded as milliseconds in a QUIC varint, so diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index fc110d7f368..3e8db6c49a0 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2314,24 +2314,54 @@ impl TrackState { /// Below the first arrival only when the feed was asked to start lower, as `floor` says. /// With `delivered`, only a group it lacks counts. fn may_land(&self, groups: ops::Range, floor: bool, grace: Duration, delivered: Option<&Delivered>) -> bool { + self.find_landable(groups, floor, grace, |open| { + delivered.is_none_or(|delivered| !delivered.covers(open)).then_some(()) + }) + .is_some() + } + + /// The first `f` answers of the runs of `groups` that may still land from the live + /// feed, lowest first, by the rule [`Self::may_land`] applies. + fn find_landable( + &self, + groups: ops::Range, + floor: bool, + grace: Duration, + mut f: impl FnMut(ops::Range) -> Option, + ) -> Option { let start = match floor { true => groups.start, - false => match self.arrivals.first() { - Some(first) => groups.start.max(first), - None => return false, - }, + false => groups.start.max(self.arrivals.first()?), }; let start = start.max(self.start_sequence.unwrap_or(0)); let end = groups.end.min(self.final_sequence().unwrap_or(u64::MAX)); let now = self.cache.pool().now_wall(); - self.arrivals.gaps(start..end).any(|(gap, opened)| { - opened + for (gap, opened) in self.arrivals.gaps(start..end) { + if opened .and_then(|opened| opened.checked_add(grace)) - .is_none_or(|folds| now < folds) - && self - .drops - .gaps(&gap) - .any(|open| delivered.is_none_or(|delivered| !delivered.covers(open))) + .is_some_and(|folds| now >= folds) + { + continue; + } + if let Some(found) = self.drops.gaps(&gap).find_map(&mut f) { + return Some(found); + } + } + None + } + + /// The lowest group in `groups` and `window` a sequence-order reader still waits for: + /// one only a subscription owes, so a received copy's live feed may still land it. Not a + /// group accounted for by its arrivals (since evicted, aborted, or a datagram), dropped, + /// outside the feed, or in a gap its grace stopped waiting for. A local track owes none. + fn owed(&self, groups: ops::Range, window: &Window) -> Option { + if !self.received || self.closed || !self.readable() { + return None; + } + let floor = groups.start.max(self.live_floor.unwrap_or(0)); + let asked = self.start_sequence.is_some(); + self.find_landable(floor..groups.end, asked, self.arrival_grace(), |open| { + window.clip(open).map(|run| run.start) }) } @@ -6528,6 +6558,32 @@ impl Cursor { }) } + /// Pending while a group in `groups` that the live feed still owes is within the drift + /// budget, so a read in sequence order waits for it; see `TrackState::owed`. A budget + /// with no deadline waits until the hole resolves: the group lands, or the track learns + /// it won't. + fn poll_owed(&self, groups: ops::Range, drift: &Drift, waiter: &kio::Waiter) -> Poll> { + if groups.is_empty() { + return Poll::Ready(Ok(())); + } + let window = &self.filter.window; + self.poll(waiter, |state| { + let Some(owed) = state.owed(groups.clone(), window) else { + return Poll::Ready(Ok(())); + }; + if state.poll_drifted(owed, &drift.edge.servable, drift.budget, None, waiter) { + return Poll::Ready(Ok(())); + } + // The ledger may stop owing it before the budget gives up on it. + if let Some(folds) = state.arrivals.folds_at(owed, state.arrival_grace()) { + state.cache.watch_wall(folds, waiter); + } + // Parked on the track too: the group landing, its drop, or a group landing + // between it and the candidate, which can convict it sooner. + Poll::Pending + }) + } + /// Whether the drift budget says to skip `group`, against a [`Drift`] already resolved /// for this poll. fn poll_stale(&self, group: &group::Consumer, drift: &Drift, waiter: &kio::Waiter) -> Poll> { @@ -6909,6 +6965,8 @@ impl Cursor { waiter: &kio::Waiter, ) -> Poll>> { let end = self.end_sequence; + // Where the gap below the candidate starts: groups stepped over as stale are cached. + let mut from = floor; let group = loop { let window = &self.filter.window; let polled = self.poll(waiter, |state| state.poll_next_in_range(floor, end, window)); @@ -6948,6 +7006,7 @@ impl Cursor { if !self.filter.window.is_pending() { drift = ready!(self.poll_drift(&self.servable, waiter))?; } + from = from.max(self.floor()); if group.sequence < self.floor() || !self.filter.window.contains(group.sequence) { floor = floor.max(self.floor()).max(group.sequence.saturating_add(1)); continue; @@ -6966,6 +7025,10 @@ impl Cursor { continue; } + // In order means a group the live feed still owes below this one comes first, + // until the budget gives up on it. Its landing, its drop, or that verdict wakes us. + ready!(self.poll_owed(from..group.sequence, &drift, waiter))?; + // A conviction the budget walked back: the group is handed over after all, // so it must never reach the stale count. self.seek_pending.remove(&group.sequence); @@ -7899,9 +7962,22 @@ impl Subscriber { /// reading a gap-free sequence needs. /// /// The same budget follows a group already handed out: once newer content pulls that far -/// ahead, a read still waiting on it ends with [`Error::Old`] and the cursor moves on. A -/// gap costs nothing either way, since the cursor seeks to the lowest cached group in -/// range rather than waiting for the sequence it would have read next. +/// ahead, a read still waiting on it ends with [`Error::Old`] and the cursor moves on. +/// +/// # Gaps +/// +/// A missing group is waited for only while the track still owes it: on a copy a +/// subscription feeds, a group that may still land, reordered behind its successor, is +/// read first, until the same budget gives up on it. Without a budget (`None`, which a +/// history or VOD read uses) it waits until the hole resolves. A group the track knows will never land costs nothing: +/// one the publisher dropped, one that arrived and was since evicted or aborted, a +/// datagram's sequence, or one below where the feed starts. A locally produced track owes +/// nothing, so there the cursor seeks to the lowest cached group in range. +/// +/// An unlimited or long budget only avoids blocking where the feed signals its gaps: +/// moq-lite, whose publisher sends `SUBSCRIBE_DROP` for every group it won't deliver, or an +/// ordered moq-transport FETCH. A moq-transport SUBSCRIBE signals no gaps, so there a group +/// lost in transit can hold the ordered reader at its hole until the track ends. /// /// The budget is updatable mid-stream through [`Self::control`]. [`Duration::ZERO`] (the /// default) keeps only what nothing newer has superseded, so a consumer that stalls and @@ -14522,6 +14598,250 @@ mod test { assert!(done.is_none(), "track should finish without waiting for gaps"); } + /// A received copy, as a session's subscription feeds it, on a pinned wall clock. + #[cfg(feature = "async")] + fn received_producer(info: impl Into>) -> Producer { + let producer = track_producer("test", info); + producer.received(); + elapse(&producer, Duration::ZERO); + producer + } + + /// Create and finish group `sequence`, with one frame at `millis` on a timed track. + #[cfg(feature = "async")] + fn put(producer: &Producer, sequence: u64, millis: Option) { + let mut group = producer.create_group(group::Info { sequence }).unwrap(); + if let Some(millis) = millis { + group + .write_frame(Timestamp::from_millis(millis).unwrap(), &b"x"[..]) + .unwrap(); + } + group.finish().unwrap(); + } + + /// A received group reordered behind its successor is still delivered in order: the + /// subscription owes it, so the reader waits for it within its budget. + #[test] + #[cfg(feature = "async")] + fn next_group_waits_for_an_owed_group() { + let producer = received_producer(None); + let mut consumer = producer + .subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))) + .ordered(); + put(&producer, 0, None); + put(&producer, 2, None); + assert_eq!(drain_ordered(&mut consumer), [0], "group 1 is still owed"); + put(&producer, 1, None); + assert_eq!(drain_ordered(&mut consumer), [1, 2]); + } + + /// A local track owes nothing: a gap costs nothing, as before. + #[test] + #[cfg(feature = "async")] + fn next_group_skips_a_local_gap() { + let producer = track_producer("test", None); + let mut consumer = producer + .subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))) + .ordered(); + put(&producer, 0, None); + put(&producer, 2, None); + assert_eq!(drain_ordered(&mut consumer), [0, 2]); + } + + /// An untimed track with groups 0 and 2..10 keeps delivering: the wall clock gives up + /// on group 1 a budget after its successor arrived, waking the parked reader exactly then, + /// and nothing after waits on it again. + #[test] + #[cfg(feature = "async")] + fn next_group_passes_an_owed_group_at_the_budget() { + let budget = Duration::from_millis(50); + let producer = received_producer(None); + let mut consumer = producer + .subscribe(Subscription::default().with_max_delay(budget)) + .ordered(); + put(&producer, 0, None); + put(&producer, 2, None); + assert_eq!(drain_ordered(&mut consumer), [0]); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone())))); + assert!(consumer.poll_next_group(&waiter).is_pending()); + elapse(&producer, budget - Duration::from_millis(1)); + assert!(!woken.load(Ordering::SeqCst), "woken before the budget"); + assert!(consumer.poll_next_group(&waiter).is_pending()); + elapse(&producer, Duration::from_millis(1)); + assert!(woken.load(Ordering::SeqCst), "the budget never woke the reader"); + assert_eq!(drain_ordered(&mut consumer), [2]); + for sequence in 3..10 { + put(&producer, sequence, None); + assert_eq!(drain_ordered(&mut consumer), [sequence]); + } + } + + /// On a timed track, media time gives up on an owed group too: once the edge is a full + /// budget past where its successor starts, and not a frame before. + #[test] + #[cfg(feature = "async")] + fn next_group_passes_an_owed_group_past_the_media_deadline() { + let producer = received_producer(Info::timed()); + let mut consumer = producer + .subscribe(Subscription::default().with_max_delay(Duration::from_secs(2))) + .ordered(); + put(&producer, 5, Some(5000)); + assert_eq!(drain_ordered(&mut consumer), [5]); + + let mut later = producer.create_group(group::Info { sequence: 7 }).unwrap(); + later + .write_frame(Timestamp::from_millis(7000).unwrap(), &b"start"[..]) + .unwrap(); + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone())))); + assert!(consumer.poll_next_group(&waiter).is_pending(), "group 6 is still owed"); + woken.store(false, Ordering::SeqCst); + later + .write_frame(Timestamp::from_millis(8999).unwrap(), &b"within"[..]) + .unwrap(); + assert!(consumer.poll_next_group(&waiter).is_pending()); + woken.store(false, Ordering::SeqCst); + later + .write_frame(Timestamp::from_millis(9000).unwrap(), &b"deadline"[..]) + .unwrap(); + assert!(woken.load(Ordering::SeqCst), "the deadline never woke the reader"); + assert_eq!(drain_ordered(&mut consumer), [7]); + } + + /// A group landing between the owed one and the candidate wakes the reader: it can + /// convict the owed group sooner, and is then read without waiting. + #[test] + #[cfg(feature = "async")] + fn next_group_rejudges_an_owed_group_when_a_lower_successor_lands() { + let producer = received_producer(Info::timed()); + let mut consumer = producer + .subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))) + .ordered(); + put(&producer, 0, Some(0)); + assert_eq!(drain_ordered(&mut consumer), [0]); + let mut edge = producer.create_group(group::Info { sequence: 4 }).unwrap(); + edge.write_frame(Timestamp::from_millis(4000).unwrap(), &b"edge"[..]) + .unwrap(); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone())))); + assert!(consumer.poll_next_group(&waiter).is_pending(), "group 1 is still owed"); + // Group 1 reaches only to group 2's start, a full budget behind the edge. + put(&producer, 2, Some(2000)); + assert!(woken.load(Ordering::SeqCst), "the lower successor went unjudged"); + assert_eq!(drain_ordered(&mut consumer), [2], "group 3 is still owed"); + } + + /// The arrival ledger giving up on a gap wakes the reader, even when the owed group's + /// successor now is a later arrival whose own budget runs longer. + #[test] + #[cfg(feature = "async")] + fn next_group_passes_an_owed_group_when_its_gap_folds() { + let budget = Duration::from_millis(50); + let producer = received_producer(None); + let mut consumer = producer + .subscribe(Subscription::default().with_max_delay(budget)) + .ordered(); + put(&producer, 0, None); + assert_eq!(drain_ordered(&mut consumer), [0]); + let opener = producer.create_group(group::Info { sequence: 2 }).unwrap(); + elapse(&producer, Duration::from_millis(40)); + put(&producer, 3, None); + opener.abort(Error::Cancel).unwrap(); + producer.modify().unwrap().reclaim_aborted(2); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone())))); + assert!(consumer.poll_next_group(&waiter).is_pending(), "group 1 is still owed"); + elapse(&producer, Duration::from_millis(9)); + assert!(!woken.load(Ordering::SeqCst), "woken before the gap folds"); + elapse(&producer, Duration::from_millis(1)); + assert!(woken.load(Ordering::SeqCst), "the fold never woke the reader"); + assert_eq!(drain_ordered(&mut consumer), [3]); + } + + /// What the track knows will never land is passed at once: a drop, a datagram, an + /// aborted group whose slot was reclaimed, or a group from below where the feed starts. + #[test] + #[cfg(feature = "async")] + fn next_group_passes_what_is_not_owed() { + let subscription = Subscription::default().with_max_delay(Duration::from_secs(30)); + let mut producer = received_producer(None); + let mut consumer = producer.subscribe(subscription.clone()).ordered(); + put(&producer, 0, None); + assert_eq!(producer.append_datagram(None, &b"d"[..]).unwrap(), 1); + put(&producer, 2, None); + assert_eq!( + drain_ordered(&mut consumer), + [0, 2], + "a datagram took group 1's sequence" + ); + + producer.drop_groups(3..4, StreamError::Old); + put(&producer, 4, None); + assert_eq!(drain_ordered(&mut consumer), [4], "group 3 was dropped"); + + let aborted = producer.create_group(group::Info { sequence: 5 }).unwrap(); + aborted.abort(Error::Cancel).unwrap(); + producer.modify().unwrap().reclaim_aborted(5); + assert!(!producer.state.read().lookup.contains_key(&5), "the slot was reclaimed"); + put(&producer, 6, None); + assert_eq!(drain_ordered(&mut consumer), [6], "group 5 was aborted and reclaimed"); + + // Below the first arrival, a feed with no declared start owes nothing. + let producer = received_producer(None); + let mut consumer = producer.subscribe(subscription.clone()).ordered(); + put(&producer, 1, None); + assert_eq!(drain_ordered(&mut consumer), [1]); + + // A declared start owes everything from it. + let mut producer = received_producer(None); + let mut consumer = producer.subscribe(subscription).ordered(); + producer.start_at(0).unwrap(); + put(&producer, 1, None); + assert!( + drain_ordered(&mut consumer).is_empty(), + "group 0 is owed from the start" + ); + put(&producer, 0, None); + assert_eq!(drain_ordered(&mut consumer), [0, 1]); + } + + /// The wait follows the reader's budget: a zero budget waits for nothing, a budget + /// narrowed mid-wait ends it, and an unbounded one waits until the hole resolves. + #[test] + #[cfg(feature = "async")] + fn next_group_waits_on_an_owed_group_within_the_budget() { + let producer = received_producer(None); + let mut consumer = producer.subscribe(Subscription::default()).ordered(); + put(&producer, 0, None); + assert_eq!(drain_ordered(&mut consumer), [0]); + put(&producer, 2, None); + assert_eq!(drain_ordered(&mut consumer), [2], "a zero budget waits for nothing"); + + let producer = received_producer(None); + let mut consumer = producer.subscribe(replay()).ordered(); + put(&producer, 0, None); + put(&producer, 2, None); + assert_eq!(drain_ordered(&mut consumer), [0]); + consumer.control().update(Subscription::default()).unwrap(); + assert_eq!(drain_ordered(&mut consumer), [2], "the narrowed budget gave up"); + + let producer = received_producer(None); + let mut consumer = producer + .subscribe(Subscription::default().with_max_delay(None)) + .ordered(); + put(&producer, 0, None); + put(&producer, 2, None); + assert_eq!(drain_ordered(&mut consumer), [0]); + elapse(&producer, Duration::from_secs(3600)); + assert!(drain_ordered(&mut consumer).is_empty(), "no deadline, however long"); + put(&producer, 1, None); + assert_eq!(drain_ordered(&mut consumer), [1, 2]); + } + #[test] #[cfg(feature = "async")] fn next_group_skips_late_arrivals() { diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 6a9756e1443..9f9fb9be298 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -31,7 +31,7 @@ Their handles `OriginDynamic`, `BroadcastDynamic`, and `TrackDynamic` are `Origi Every duration is a `TimeInterval` in seconds instead of a `UInt64` `...Us` field, and drops the suffix: `Client(timeout:race:resolutionDelay:)`, `Backoff(initial:max:timeout:)`, `QuicConfig(idleTimeout:keepAlive:)`, `WebSocketConfig(delay:)`, `ConnectionStats.rtt`, `Grant.expires`, `TrackInfo.maxAge`, `Subscription.maxDelay`, `Fill.maxDelay`, `Audio.Consumer.delay`, and the decoder outputs' `maxDelay` and `delay`. Divide a microsecond value by 1,000,000. A negative interval traps. `AudioEncoderOutput.frameDuration` is optional: pass `nil` where you passed `frameDurationUs: 0` for the codec's own frame. -`Subscription.maxDelay` defaults to `nil`, which sets no subscriber limit: start from `Subscription.default`, or pass `0`, for the old default. +`Subscription.maxDelay` has no default: start from `Subscription.default`, or pass `0`, for the old default. `nil` sets no subscriber limit. `TrackConsumer.info()` no longer throws. ### Connection state diff --git a/swift/Sources/Moq/Aliases.swift b/swift/Sources/Moq/Aliases.swift index 9787186ca16..79cc6830900 100644 --- a/swift/Sources/Moq/Aliases.swift +++ b/swift/Sources/Moq/Aliases.swift @@ -130,7 +130,7 @@ extension Subscription { public typealias Order = MoqFFI.MoqOrder /// moq-net's defaults, whose `maxDelay` of 0 skips a superseded group at once. A `nil` - /// subscription means the same; copy this one to change a field. A `Subscription()` - /// leaves `maxDelay` `nil`, which sets no subscriber limit. + /// subscription means the same; copy this one to change a field. A `Subscription(...)` + /// must name `maxDelay`, where `nil` sets no subscriber limit. public static var `default`: Subscription { moqSubscriptionDefaults() } } diff --git a/swift/Tests/MoqTests/SmokeTests.swift b/swift/Tests/MoqTests/SmokeTests.swift index d672f32f4d0..5a9c90c8117 100644 --- a/swift/Tests/MoqTests/SmokeTests.swift +++ b/swift/Tests/MoqTests/SmokeTests.swift @@ -11,7 +11,7 @@ private func nextRoute(_ announced: AnnounceConsumer) async throws -> AnnounceEv final class SmokeTests: XCTestCase { func testDefaultSubscriptionSkipsSupersededGroupsAtOnce() { XCTAssertEqual(Subscription.default.maxDelay, 0) - XCTAssertNil(Subscription().maxDelay) + XCTAssertNil(Subscription(maxDelay: nil).maxDelay) } func testContainerTimebaseBelongsToOneBroadcast() throws {