diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 809e8b5e944..5a218fd0416 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -20488,7 +20488,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqoriginconsumer_request_broadcast() != - 65225) { + 54627) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqorigindynamic_cancel() != 47453) { diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 24c56defda0..48f9f806421 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -25,7 +25,7 @@ maps everything else to "not supported" or a harmless equivalent. The [moq-lite page](/concept/moq-lite#what-moq-lite-leaves-out) lists what the subset drops. What a peer actually observes against this implementation: -- **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited, and we also ask for every prefix we may discover. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below): on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session. +- **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined, unless they answer our `SUBSCRIBE_TRACKS` (below); announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. Rust never holds two overlapping requests at a peer that did not declare solicit: a broader one replaces the narrower ones once they are closed. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We send `PUBLISH` only to answer a `SUBSCRIBE_TRACKS` (below): on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session. - **`SUBSCRIBE_TRACKS` lists, it never pushes.** On draft 18 and later, it lists every track of every broadcast under its prefix, later ones included. With `FORWARD=0`, each track gets a paused `PUBLISH` carrying its Track Properties, and an error `PUBLISH_DONE` once the track is no longer listed, never `TRACK_ENDED`, since the track may go on elsewhere. A `REQUEST_UPDATE` with `FORWARD=1` resumes one, served from the cache like a `SUBSCRIBE` with the update's filter and answered with `LARGEST_OBJECT`; `FORWARD=0` pauses it again. Rust opens each paused `PUBLISH` only as the peer's stream credit admits it, so a peer that grants none holds none; JavaScript caps them, and past the cap a track gets `PUBLISH_SKIPPED` (`PUBLISH_BLOCKED` on draft 18) instead. A listing that omits `FORWARD` or sets it to 1 asks for every track's objects: from draft 19 each track gets `PUBLISH_SKIPPED`, and the subscriber `SUBSCRIBE`s to what it wants; draft 18 has no such skip, so the request is refused with `NOT_SUPPORTED`. A prefix overlapping another listing gets `PREFIX_OVERLAP`, a `TRACK_PROPERTY_FILTER` gets `NOT_SUPPORTED`, and a `REQUEST_UPDATE` on the listing itself ends it. On draft 18, a `SUBSCRIBE` to a track we `PUBLISH`ed gets `DUPLICATE_SUBSCRIPTION`. Cancelling the listing stops new `PUBLISH`es, while those sent live on. A broadcast is listed only while our publish grant covers it, from scratch once it does. Rust sends it upstream for a listing (below); JavaScript does not. - **A Rust TRACKS listing sends `SUBSCRIBE_TRACKS` upstream, only on demand.** Listing a broadcast from a moq-transport peer on draft 18 and later sends `SUBSCRIBE_TRACKS` with `FORWARD=0` while someone lists it, never on connect, announce, or subscribe, so a peer that refuses, resets, or ignores it affects only that listing. Before draft 18 the listing is refused as unsupported. Each listed track's paused `PUBLISH` is its upstream copy, which a reader resumes rather than subscribing again. Only a `PUBLISH_DONE` with `TRACK_ENDED` removes a listed track, and it lists as finished; any other end, including the error our own answerer sends for a track that left unfinished, fails the listing, and readers list again. - **History is one group.** In Rust, Object ID gaps must be marked by Prior Object ID Gap properties for a group to be fetchable. A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On drafts 14 through 19, `FETCH_OK` End Object 0 covers the entire End Group, matching the request; Rust also accepts the legacy answer naming the next group with Object 0. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed, and so does a reader whose publisher refuses `TRACK_STATUS` as `NOT_SUPPORTED`. diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 600abb11c1e..2db638a1b5c 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -19,7 +19,7 @@ What you use it for, beyond what the concept page already describes: - **Requests can pin an epoch.** `consumer.request_broadcast(path, Some(epoch))` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs); `None` takes whichever route wins. The resolved consumer names its epoch in `info().epoch`, pinned or not. - **Rank several CDNs.** Dial each with its own subscribe handle, `origin.clone().with_preference(origin::Preference::new(rank))` (higher wins; 0 is neutral and a negative rank demotes), all on one origin. Requests use the preferred link that has a route and fail over to the next; see [multiple CDNs](/concept/moq-lite#multiple-cdns). - **Follow restarts.** `consumer.announced_broadcast(path, None)` hands out each publisher instance serving a path from `available()`: the first once announced, then each one that takes over or returns after an end, like JS `request(path, { announced: true })`. A broadcast already resolved stays on the old instance until dropped or its route goes. `consumer.follow(path)` yields the underlying start, update, restart, and end of the route a request resolves. -- **Discovery follows what you watch.** On moq-lite, a client session asks the relay for announcements only under the prefixes its subscribe origin's readers watch, with `announced()`, `routed_broadcast`, or `announced_broadcast`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A broadcast resolved through a discovered route keeps its path watched until dropped. `request_broadcast` watches nothing, so a path nobody watches is unroutable. A server session, or a client fed into an origin handle marked `with_link(origin::Link::Peer)` or `Link::Upstream`, asks for everything it may, so a relay discovers everything. An HTTP gateway (the moq-hls `Server`, the moq-rtc WHEP router) watches its whole scope while it lives, so the paths its callers name never open a request each. moq-transport sessions, and moq-lite-03/04 sessions (whose announcements mark no initial set), still ask for everything. +- **Discovery follows what you watch.** A client session asks the relay for announcements only under the prefixes its subscribe origin's readers watch, with `announced()`, `routed_broadcast`, or `announced_broadcast`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A broadcast resolved through a discovered route keeps its path watched until dropped. `request_broadcast` watches nothing, so a path nobody watches is unroutable. A server session, or a client fed into an origin handle marked `with_link(origin::Link::Peer)` or `Link::Upstream`, asks for everything it may, so a relay discovers everything. An HTTP gateway (the moq-hls `Server`, the moq-rtc WHEP router) watches its whole scope while it lives, so the paths its callers name never open a request each. moq-lite-03/04 sessions, whose announcements mark no initial set, still ask for everything. - **One track per name.** Concurrent subscriptions share one request and producer. The application owns sequence numbers: a producer that replaces an ended one gets `Error::Duplicate` from an append that would reuse a sequence the name already wrote, so it starts with `create_group` at an explicit sequence. Only a new broadcast starts over. Readers of an aborted track resume onto the replacement only when the broadcast announces an epoch; otherwise their subscription ends with the error. - **End a broadcast with `finish()`.** When the input ends, finish the tracks, then call `broadcast.finish()`. The returned handle keeps the broadcast announced and serving its finished tracks for `linger()`, their longest max-age, so a subscriber still catching up can drain them. Await `wait()` on a runtime, or drop the handle once your own clock passes `linger()`. Dropping it early is the interrupt, and `close()` on another handle ends it at once. - **Publish only while someone is watching.** `demand()` on a track, group, or broadcast says whether a subscriber is attached, which is how capture and transcode skip work nobody asked for. Holding a broadcast or track consumer is not demand: a track consumer requests nothing until it queries, subscribes, or fetches, and those count until dropped. A shared fetch stays up until its last reader leaves. diff --git a/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index 2db1b1352bc..e3133fc3745 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -111,7 +111,11 @@ interface StreamEntry { controller: ReadableStreamDefaultController; } -type NamespaceDirection = "incoming" | "outgoing"; +/** + * Which request a v14/v15 namespace-keyed message names: an announcement, by the side that + * made it, or the peer's namespace subscription (draft 14's UNSUBSCRIBE_NAMESPACE). + */ +type NamespaceDirection = "incoming" | "outgoing" | "subscribed"; /** Runs per chunk before it splits, so a split or removal shifts at most one chunk. */ const RUN_CHUNK = 256; @@ -252,6 +256,8 @@ export class ControlStreamAdapter implements Session { // may legitimately hold the same namespace in both maps at once. #incomingNamespaces = new Map(); #outgoingNamespaces = new Map(); + // The peer's draft-14 SUBSCRIBE_NAMESPACE prefixes, which its UNSUBSCRIBE_NAMESPACE names. + #subscribedNamespaces = new Map(); // requestId → namespace reverse lookup, one entry per request including a refused // duplicate, so #forget can clean up whichever request goes away. @@ -842,12 +848,13 @@ export class ControlStreamAdapter implements Session { } /** - * Decode a PublishNamespace body and register its namespace for the v14/v15 - * namespace-keyed withdrawals. Returns the request ID it carried. + * Decode a PublishNamespace body, or a draft-14 SubscribeNamespace one, and register its + * namespace for the v14/v15 namespace-keyed messages that end it. Returns the request + * ID it carried. * - * The one decode site for both directions: outgoing announcements go through - * #tryParseOutgoing and incoming ones through #classify. The direction selects the - * map that the peer's DONE or CANCEL will later consult. + * The one decode site for every direction: outgoing announcements go through + * #tryParseOutgoing and incoming requests through #classify. The direction selects the + * map that the peer's DONE, CANCEL, or UNSUBSCRIBE_NAMESPACE will later consult. */ async #registerNamespace(body: Uint8Array, direction: NamespaceDirection): Promise { const r = new Reader(undefined, body, this.version); @@ -866,7 +873,14 @@ export class ControlStreamAdapter implements Session { } #namespaceRequests(direction: NamespaceDirection): Map { - return direction === "incoming" ? this.#incomingNamespaces : this.#outgoingNamespaces; + switch (direction) { + case "incoming": + return this.#incomingNamespaces; + case "outgoing": + return this.#outgoingNamespaces; + case "subscribed": + return this.#subscribedNamespaces; + } } /** Create a WritableStream that buffers and writes complete messages to the control stream under mutex. */ @@ -960,7 +974,7 @@ export class ControlStreamAdapter implements Session { const namespace = await Namespace.decode(r); const namespaces = this.#namespaceRequests(direction); const requestId = namespaces.get(namespace); - if (requestId === undefined) console.warn(`adapter: no announcement for namespace: ${namespace}`); + if (requestId === undefined) console.warn(`adapter: no request for namespace: ${namespace}`); return requestId; }; @@ -994,7 +1008,11 @@ export class ControlStreamAdapter implements Session { if (this.version !== Version.DRAFT_14 && this.version !== Version.DRAFT_15) { throw new Error("unexpected SubscribeNamespace on control stream"); } - const requestId = await readRequestId(); + // Draft 14's UNSUBSCRIBE_NAMESPACE names the prefix rather than the request. + const requestId = + this.version === Version.DRAFT_14 + ? await this.#registerNamespace(body, "subscribed") + : await readRequestId(); return { route: Route.NewRequest, requestId }; } @@ -1119,11 +1137,14 @@ export class ControlStreamAdapter implements Session { return { route: Route.CloseStream, requestId }; } case 0x14: { - // UnsubscribeNamespace (v14/v15 only) - if (this.version !== Version.DRAFT_14 && this.version !== Version.DRAFT_15) { - throw new Error("unexpected UnsubscribeNamespace"); + // UnsubscribeNamespace (v14/v15 only): v14 names the prefix, v15 the request. + if (this.version === Version.DRAFT_15) { + const requestId = await readRequestId(); + return { route: Route.CloseStream, requestId }; } - const requestId = await readRequestId(); + if (this.version !== Version.DRAFT_14) throw new Error("unexpected UnsubscribeNamespace"); + const requestId = await readNamespaceRequestId("subscribed"); + if (requestId === undefined) return { route: Route.Ignore, requestId: 0n }; return { route: Route.CloseStream, requestId }; } @@ -1176,6 +1197,7 @@ export class ControlStreamAdapter implements Session { // Clear namespace mappings this.#incomingNamespaces.clear(); this.#outgoingNamespaces.clear(); + this.#subscribedNamespaces.clear(); this.#namespacesByRequestId.clear(); this.#subscribeNamespaces.clear(); diff --git a/js/net/src/ietf/ietf.test.ts b/js/net/src/ietf/ietf.test.ts index abb7d9239b2..febd52a8315 100644 --- a/js/net/src/ietf/ietf.test.ts +++ b/js/net/src/ietf/ietf.test.ts @@ -455,6 +455,20 @@ test("PublishNamespaceDone: round trip", async () => { expect(decoded.trackNamespace).toBe("old/stream" as Path.Valid); }); +test("UnsubscribeNamespace: names the prefix on draft 14 and the request on draft 15", async () => { + const msg = new SubscribeNamespace.UnsubscribeNamespace({ prefix: Path.from("cam/front"), requestId: 6n }); + + const v14 = await encodeVersioned(msg, Version.DRAFT_14); + const named = await decodeVersioned(v14, SubscribeNamespace.UnsubscribeNamespace.decode, Version.DRAFT_14); + expect(named.prefix).toBe("cam/front" as Path.Valid); + + const v15 = await encodeVersioned(msg, Version.DRAFT_15); + const numbered = await decodeVersioned(v15, SubscribeNamespace.UnsubscribeNamespace.decode, Version.DRAFT_15); + expect(numbered.requestId).toBe(6n); + + await expect((() => encodeVersioned(msg, Version.DRAFT_16))()).rejects.toThrow(/draft-14\/15 only/); +}); + test("PublishNamespaceCancel: round trip", async () => { const msg = new Announce.PublishNamespaceCancel({ trackNamespace: Path.from("canceled"), diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 9263e057a0c..94773cbc3c1 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -2652,7 +2652,7 @@ test("drafts 14 to 16 refuse a namespace update without ending the subscription" ask.writer.close(); } else { await peer.writer.u53(UnsubscribeNamespace.id); - await new UnsubscribeNamespace({ requestId: 0n }).encode(peer.writer, version); + await new UnsubscribeNamespace({ prefix: asked.namespace, requestId: 0n }).encode(peer.writer, version); } await running; origin.close(); diff --git a/js/net/src/ietf/subscribe_namespace.ts b/js/net/src/ietf/subscribe_namespace.ts index 341a565bbf4..d3a62cc91b9 100644 --- a/js/net/src/ietf/subscribe_namespace.ts +++ b/js/net/src/ietf/subscribe_namespace.ts @@ -253,30 +253,42 @@ export class SubscribeNamespaceError { } } +/** UNSUBSCRIBE_NAMESPACE (0x14), drafts 14 and 15 only: draft 14 names the prefix, draft 15 the request. */ export class UnsubscribeNamespace { static id = 0x14; + /** Draft 14: the Track Namespace Prefix of the SUBSCRIBE_NAMESPACE it cancels. */ + prefix: Path.Valid; + /** Draft 15: the Request ID of the SUBSCRIBE_NAMESPACE it cancels. */ requestId: bigint; - constructor({ requestId }: { requestId: bigint }) { + constructor({ prefix = "" as Path.Valid, requestId = 0n }: { prefix?: Path.Valid; requestId?: bigint }) { + this.prefix = prefix; this.requestId = requestId; } - async #encode(w: Writer): Promise { - await w.u62(this.requestId); + async #encode(w: Writer, version: IetfVersion): Promise { + if (version === Version.DRAFT_14) { + await Namespace.encode(w, this.prefix); + } else if (version === Version.DRAFT_15) { + await w.u62(this.requestId); + } else { + throw new Error(`UNSUBSCRIBE_NAMESPACE is draft-14/15 only, not ${version}`); + } } - async encode(w: Writer, _version: IetfVersion): Promise { - return Message.encode(w, this.#encode.bind(this)); + async encode(w: Writer, version: IetfVersion): Promise { + return Message.encode(w, (wr) => this.#encode(wr, version)); } - static async decode(r: Reader, _version: IetfVersion): Promise { - return Message.decode(r, UnsubscribeNamespace.#decode); + static async decode(r: Reader, version: IetfVersion): Promise { + return Message.decode(r, (rd) => UnsubscribeNamespace.#decode(rd, version)); } - static async #decode(r: Reader): Promise { - const requestId = await r.u62(); - return new UnsubscribeNamespace({ requestId }); + static async #decode(r: Reader, version: IetfVersion): Promise { + if (version === Version.DRAFT_14) return new UnsubscribeNamespace({ prefix: await Namespace.decode(r) }); + if (version === Version.DRAFT_15) return new UnsubscribeNamespace({ requestId: await r.u62() }); + throw new Error(`UNSUBSCRIBE_NAMESPACE is draft-14/15 only, not ${version}`); } } diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index acc2d40d13d..62147a9ccbd 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -563,7 +563,7 @@ export class Subscriber { // For v14/v15: send UnsubscribeNamespace before closing if (version === Version.DRAFT_14 || version === Version.DRAFT_15) { try { - const unsub = new UnsubscribeNamespace({ requestId }); + const unsub = new UnsubscribeNamespace({ prefix, requestId }); await Message.write(request.writer, UnsubscribeNamespace.id, unsub, version); } catch { // Stream might already be closed diff --git a/quest/m1/README.md b/quest/m1/README.md index c3bb97b25f2..9bf0d12633c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -155,7 +155,7 @@ Binding durations wait on Duration support in the generated C. - [API-diff check](/quest/m1/api-diff.md) - a recipe diffs every published package's API against its last release, run on PRs into release and against main before the cut - [API Extractor checks](/quest/m1/api-extractor-checks.md) - API Extractor fails a PR on a forgotten export, an unresolved declaration, or an undocumented export, replacing the hand-rolled `.d.ts` checkers - [Generated JS API reference](/quest/m1/js-api-reference.md) - the doc site renders an API reference for every `@moq/*` package from API Extractor's doc model -- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - moq-transport sessions ask for announcements only under watched prefixes, starting with UNSUBSCRIBE_NAMESPACE on d14/15 +- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - JS moq-transport sessions never hold overlapping SUBSCRIBE_NAMESPACE requests at a peer that refuses them - [Draft 14-16 session errors](/quest/m1/ietf-adapter-session-errors.md) - the moq-transport adapter keeps session-level error causes, so a terminal rejection is not redialed - [Interop coverage](/quest/m1/interop-coverage/README.md) - the riskiest moq-transport features moq-net exposes get cross-implementation tests over the even drafts 14 to 22 - [One Annex-B exporter](/quest/m1/annexb-export.md) - H.264 and H.265 Annex-B export share one private lifecycle, which also follows a rebound rendition diff --git a/quest/m1/ietf-lazy-discovery.md b/quest/m1/ietf-lazy-discovery.md index aa9f21c0653..aa86be17d8e 100644 --- a/quest/m1/ietf-lazy-discovery.md +++ b/quest/m1/ietf-lazy-discovery.md @@ -1,37 +1,41 @@ -# [L] IETF lazy discovery +# [M] IETF lazy discovery in JS ## Goal -A moq-transport session, like a lite one, asks for announcements only under -the prefixes its origin is watching, and withdraws the ask when nobody -watches any more, on every draft we support. +A JS moq-transport session never holds two overlapping SUBSCRIBE_NAMESPACE +requests at a peer that did not declare MoQ Solicit, so a peer that enforces +the drafts' PREFIX_OVERLAP rule, such as moxygen, never refuses a broader +watch and turns the session blind. ## Plan -[moq-dev/moq#6165](https://github.com/moq-dev/moq/pull/6165) made lite -discovery lazy and kept IETF eager: lazy SUBSCRIBE_NAMESPACE needs an -UNSUBSCRIBE_NAMESPACE path on drafts 14 and 15, and peers still push -unsolicited PUBLISH_NAMESPACE. - -In m1 because it owns a red interop cell (decided 2026-10-22): `moq` -against moxygen on draft-14 discovers nothing. moq-net never asks draft-14 -or 15 for the empty namespace, and moxygen announces nothing unasked. The -CLI does no scoping of its own: the subtree scope tried for this in -[moq-dev/moq#6255](https://github.com/moq-dev/moq/pull/6255) lost the -sibling broadcasts a catalog references, such as a transcode's `.` source. -Asking for the watched prefixes fixes the cell without that loss; verify -it with `MOQ_BIN=target/debug/moq just relays --only moxygen` from -`test/smoke`. - -- First step: Rust sends UNSUBSCRIBE_NAMESPACE on d14/15. It never does - today ([moq-dev/moq#6133](https://github.com/moq-dev/moq/pull/6133)). -- Then drive SUBSCRIBE_NAMESPACE from the origin's watched prefixes, as the - lite path does, in Rust and JS. -- Decide what an unsolicited PUBLISH_NAMESPACE outside any watched prefix - costs the receiver, and bound it. - -Public API: none expected. Wire: no new messages. - -## Related - -- [Keyed prefix polling](/quest/m2/keyed-prefix-polling.md) - the per-wake cost of many open prefixes +Rust landed in [moq-dev/moq#6530](https://github.com/moq-dev/moq/pull/6530): +lazy SUBSCRIBE_NAMESPACE on every draft, UNSUBSCRIBE_NAMESPACE on drafts +14/15 (named by prefix on 14, fixed in both languages), and exclusive +discovery at a peer that declared no Solicit (`Discovery::exclusive`, +`rs/moq-net/src/ietf/subscriber/namespaces.rs`). JS +`connection/forward.ts` already asks only about watched prefixes, but holds a +narrower request open under a broader one on every version. + +- At a peer that declared no Solicit, a broader watch closes the narrower + requests beneath it, and its SUBSCRIBE_NAMESPACE goes out only once each + has left the wire: UNSUBSCRIBE_NAMESPACE written on drafts 14/15, the peer's + half closed from draft 16. +- `forward.ts` needs the session to say when a request is off the wire; the + announced consumer reports only its local close today. + +Settled in [moq-dev/moq#6530](https://github.com/moq-dev/moq/pull/6530): + +- JS non-overlap is this quest, not that PR: it needs the session signal + above, and a refused JS watch only downgrades the session to blind answers. +- Overlap policy is keyed on MoQ Solicit. Requests are exclusive only at a + peer that declared no Solicit, since only foreign peers enforce + PREFIX_OVERLAP; our own peers keep overlapping hold semantics, because on + drafts 14/15 our publisher withdraws what an unsubscribed request + advertised, so exclusive-everywhere would restart routes mid-handover. +- A PUBLISH_NAMESPACE that outlives an UNSUBSCRIBE_NAMESPACE (drafts 14/15), + or arrives unsolicited, is kept until the peer withdraws it. It is bounded + by request credit or streams and the `announces` cap, and cancelling it + would lose it for good, since the peer does not repeat it. + +Public API: none expected. Wire: none. diff --git a/quest/m2/audit-interop.md b/quest/m2/audit-interop.md index 8b5594f7a1a..2f6a4150635 100644 --- a/quest/m2/audit-interop.md +++ b/quest/m2/audit-interop.md @@ -97,10 +97,11 @@ Still red from `main`, ours: refuses the new token, so every moq-go cell against our relay fails. - moq-dev-js `data-subgroup-basic` against libquicr never sees the track end, while moq-dev-rs passes. Not yet traced. -- moxygen draft-14 over raw QUIC waits on - [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md), and every - `js-web` publisher cell behind a joining FETCH (imquic, MOQtail draft-22, - aiomoqt, red5) on [JavaScript FETCH](/quest/m1/js-fetch.md). +- Every `js-web` publisher cell behind a joining FETCH (imquic, MOQtail + draft-22, aiomoqt, red5) waits on [JavaScript FETCH](/quest/m1/js-fetch.md). + moxygen draft-14 over raw QUIC discovered nothing, since the CLI asked + draft-14 for no namespace; it passed live once Rust asked for the watched + prefixes ([moq-dev/moq#6530](https://github.com/moq-dev/moq/pull/6530)). Theirs, reported or waiting on others: @@ -152,5 +153,4 @@ and moqtopus and xquic-draft-18 over WebTransport (raw QUIC only). - [Runner teardown](/quest/m3/interop-runner-approval.md) - the runner's moq.dev clients move to the withdrawing `close()` once it ships - [Moxygen minimal interop](/quest/m1/smoke-moxygen.md) - Moxygen's filter type, waiting upstream - [JavaScript FETCH](/quest/m1/js-fetch.md) - the JS publisher serves FETCH, which most red `js-web` publisher cells need -- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - asks for watched prefixes, which the red moxygen draft-14 cell needs - [Interop coverage](/quest/m1/interop-coverage/README.md) - the runner cases whose red cells this audit classifies diff --git a/quest/m2/keyed-prefix-polling.md b/quest/m2/keyed-prefix-polling.md index 45e1fca87d4..103bb02ec51 100644 --- a/quest/m2/keyed-prefix-polling.md +++ b/quest/m2/keyed-prefix-polling.md @@ -19,7 +19,3 @@ Decided 2026-10-21 with the maintainer: gated on a measurement. - Otherwise key polling by prefix and show the slope flatten. Public API: none. Wire: none. - -## Related - -- [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - more prefixes per session once IETF is lazy too diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 4d87b92c0e0..5aaf0f8fe79 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -2087,8 +2087,8 @@ pub extern "C" fn moq_origin_announced_broadcast_cancel(task: u32) -> i32 { /// Resolves against what is announced *now*, where [moq_origin_announced_broadcast] waits /// indefinitely: it returns an announced broadcast at once, and fails when none is reachable, /// including a broadcast created but not announced. It does NOT wait for a later -/// announcement. Serve on-demand paths with [moq_origin_dynamic]. A moq-lite client session -/// only asks its peer about paths something watches, through [moq_origin_announced] or +/// announcement. Serve on-demand paths with [moq_origin_dynamic]. A client session only +/// asks its peer about paths something watches, through [moq_origin_announced] or /// [moq_origin_announced_broadcast], so a path nothing watches is unreachable on it. /// /// `on_broadcast` is invoked with a positive broadcast handle once served, then exactly once more diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 66ece34585a..0516fb598a8 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -494,8 +494,8 @@ impl MoqOriginConsumer { /// /// Calling this straight after connecting therefore races the session's announcements /// and can report a live broadcast as unroutable. Await `announced_broadcast` first. A - /// moq-lite client only asks the relay about paths something watches, through `announced` - /// or `announced_broadcast`, so a path nothing watches stays unroutable. + /// client only asks the relay about paths something watches, through `announced` or + /// `announced_broadcast`, so a path nothing watches stays unroutable. pub async fn request_broadcast(&self, path: String) -> Result, MoqError> { let broadcast = self .driver diff --git a/rs/moq-net/src/discovery.rs b/rs/moq-net/src/discovery.rs index 7fa8163c41b..299603420c8 100644 --- a/rs/moq-net/src/discovery.rs +++ b/rs/moq-net/src/discovery.rs @@ -4,6 +4,9 @@ //! readers want discovered, as [`origin::Producer`] reports them, and stops asking //! [`LINGER`] after nothing at or beneath a prefix is wanted. A relay or a server asks //! about everything its scope may reach, once, for the session's lifetime. +//! +//! moq-lite lets requests overlap. moq-transport forbids it, and a peer that enforces +//! that refuses an overlapping request, so a session to one runs [`Discovery::exclusive`]. use std::{ collections::{BTreeSet, HashMap, HashSet, VecDeque}, @@ -37,11 +40,14 @@ pub(crate) enum Change { /// narrower want rides it rather than opening a request that would later have to take /// its routes over. A narrower request already open when a broader one opens, wanted or /// lingering, stays open until the broader one goes, so neither takes the other's routes -/// over before it has synced. Nothing here ever reopens a request that ended on its own: the next want -/// that nothing covers does. +/// over before it has synced. An [exclusive](Self::exclusive) session closes the narrower +/// requests instead. Nothing here ever reopens a request that ended on its own, nor one a +/// broader request replaced: the next want that nothing covers does. pub(crate) struct Discovery { /// `None` for a session that asks about everything once, or once the origin closed. interests: Option, + /// Never hold two requests where one covers the other. + exclusive: bool, runtime: Deadlines, /// Decided but not yet handed to the session, in order. changes: VecDeque, @@ -73,6 +79,7 @@ impl Discovery { pub fn new(origin: &origin::Producer, client: bool, runtime: Deadlines) -> Self { let mut discovery = Self { interests: None, + exclusive: false, timer: Deadline::new(&runtime), runtime, changes: VecDeque::new(), @@ -97,6 +104,15 @@ impl Discovery { discovery } + /// Never hold two requests where one covers the other: a broader request closes the + /// narrower ones beneath it before it opens, so the session can keep it off the wire + /// until they are. The covering set of wants and the granted heads never overlap, so + /// nothing else does. + pub fn exclusive(mut self) -> Self { + self.exclusive = true; + self + } + /// The next change to apply, registering `waiter` for the next one otherwise. pub fn poll(&mut self, waiter: &kio::Waiter) -> Poll { loop { @@ -182,12 +198,34 @@ impl Discovery { if want && !covered { let snapshot = snapshot.expect("a wanted prefix comes with its snapshot"); - let request = self.hold_lingering(&prefix); + let request = match self.exclusive { + true => self.replace_beneath(&prefix), + false => self.hold_lingering(&prefix), + }; self.requests.insert(prefix.clone(), request); self.changes.push_back(Change::Open(prefix, snapshot)); } } + /// A new request at `prefix` replaces every request open beneath it, wanted or + /// lingering, which closes first. + fn replace_beneath(&mut self, prefix: &PathOwned) -> Request { + let beneath: Vec<_> = self + .requests + .keys() + .filter(|other| *other != prefix && other.has_prefix(prefix)) + .cloned() + .collect(); + for other in beneath { + if let Some(Request { linger: Some(at), .. }) = self.requests.remove(&other) { + self.lingering.remove(&(at, other.clone())); + } + self.changes.push_back(Change::Close(other)); + } + self.timer.set(self.lingering.first().map(|(at, _)| *at)); + Request::default() + } + /// A new request at `prefix` holds the narrower ones still lingering beneath it, as it /// would had they been released after it opened, so none closes before it has synced. fn hold_lingering(&mut self, prefix: &PathOwned) -> Request { diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 8ca7c1a2d70..0277bc373b7 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -602,7 +602,7 @@ pub fn ietf_wire(data: &[u8]) -> bool { 24 => roundtrip::, _>(rest, version), 25 => roundtrip::(rest, version), 26 => roundtrip::, _>(rest, version), - 27 => roundtrip::(rest, version), + 27 => roundtrip::, _>(rest, version), 28 => roundtrip::, _>(rest, version), 29 => roundtrip::, _>(rest, version), 30 => roundtrip::(rest, version), diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index 6f6120baa1f..fe84ae1fdee 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -1,7 +1,7 @@ use crate::driver::Transport; use crate::transport::poll::Session as _; use std::{ - collections::{HashMap, VecDeque}, + collections::{HashMap, HashSet, VecDeque}, sync::{Arc, Mutex, Weak}, task::{Context, Poll}, }; @@ -41,12 +41,18 @@ impl Queue { /// Enqueue a message. Returns false once the queue is closed (the message is /// dropped), matching a write to a stream whose reader is gone. fn push(&self, item: T) -> bool { + self.offer(item).is_none() + } + + /// Enqueue a message, handing it back once the queue is closed, for a caller that + /// must not drop it where it is. + fn offer(&self, item: T) -> Option { let mut state = self.0.lock(); if state.closed { - return false; + return Some(item); } state.messages.push_back(item); - true + None } /// Dequeue the next message, waiting while the queue is open and empty. @@ -610,13 +616,17 @@ impl crate::transport::poll::Lease for AdapterLease<'_> { // === Control Stream Adapter === -/// Which side of the session advertised a namespace. +/// Which request a namespace names: an advertisement, by the side that made it, or the +/// peer's namespace subscription. #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum Direction { /// We advertised it to the peer, on a stream from `open_bi`. Outgoing, /// The peer advertised it to us, on a stream handed to `accept_bi`. Incoming, + /// The peer subscribed to it with SUBSCRIBE_NAMESPACE, on a stream handed to + /// `accept_bi`. Only draft-14's UNSUBSCRIBE_NAMESPACE names one. + Subscribed, } /// Namespace → request_id reverse lookup for the v14/v15 messages that name a @@ -626,6 +636,8 @@ enum Direction { /// advertising the same namespace to each other. One map would let the second /// advertisement overwrite the first, and a terminal message would then tear /// down the stream travelling the other way while its real target stayed open. +/// A subscription keeps a map of its own for the same reason: a peer may both +/// advertise a namespace and subscribe to it. #[derive(Default)] struct Namespaces { state: Mutex, @@ -635,6 +647,7 @@ struct Namespaces { struct NamespacesState { outgoing: HashMap, incoming: HashMap, + subscribed: HashMap, /// request_id → the namespace that request advertised, one entry per request /// including queued duplicates, so [`Namespaces::forget`] removes only that @@ -660,6 +673,7 @@ impl NamespacesState { match direction { Direction::Outgoing => &mut self.outgoing, Direction::Incoming => &mut self.incoming, + Direction::Subscribed => &mut self.subscribed, } } } @@ -702,6 +716,13 @@ impl Namespaces { ); } + /// The namespace the peer's request `request_id` advertises, if it is a PUBLISH_NAMESPACE. + fn incoming(&self, request_id: RequestId) -> Option { + let state = self.state.lock().unwrap(); + let registration = state.by_request.get(&request_id)?; + (registration.direction == Direction::Incoming).then(|| registration.namespace.clone()) + } + fn get(&self, direction: Direction, namespace: &PathOwned) -> Option { self.state .lock() @@ -760,6 +781,39 @@ struct Shared { /// Namespace → request_id reverse lookup (for v14/v15 namespace-keyed messages). namespaces: Namespaces, + + /// The peer's advertisements held back behind a withdrawal still being handled. + readvertised: Mutex, +} + +/// Keeps the peer's withdrawal and re-advertisement of a namespace in wire order. +/// +/// Each request is served by its own handler, so a PUBLISH_NAMESPACE that follows a +/// PUBLISH_NAMESPACE_DONE for the same namespace could otherwise be handled first. It +/// would then read as a second advertisement of a namespace still advertised, and the +/// withdrawal as the end of the first, so the route never ends and starts again. The +/// re-advertisement reaches `accept_bi` only once the withdrawn request's handler drops +/// its stream, which it does after releasing the route. Bounded by the peer's requests. +#[derive(Default)] +struct Readvertised { + namespaces: HashMap, + /// Each withdrawn request a handler still holds, to the namespace it advertised. + withdrawn: HashMap, +} + +#[derive(Default)] +struct Withdrawing { + /// Withdrawn requests whose handlers still hold their streams. + handling: HashSet, + /// The advertisements behind them, in wire order. + held: VecDeque, +} + +struct HeldRequest { + request_id: RequestId, + stream: (VirtualSendStream, VirtualRecvStream), + /// Withdrawn while held, so the advertisements behind it wait for its handler too. + withdrawn: bool, } impl Shared { @@ -799,6 +853,17 @@ impl Shared { recv._permit = permit; let send = VirtualSendStream::new(self.control.clone()); self.streams.lock().unwrap().insert(request_id, follow.writer()); + if let Some(namespace) = self.namespaces.incoming(request_id) { + let mut readvertised = self.readvertised.lock().unwrap(); + if let Some(withdrawing) = readvertised.namespaces.get_mut(&namespace) { + withdrawing.held.push_back(HeldRequest { + request_id, + stream: (send, recv), + withdrawn: false, + }); + return Ok(()); + } + } if !self.incoming.push((send, recv)) { return Err(Error::Closed); } @@ -825,9 +890,27 @@ impl Shared { return; } let tx = self.streams.lock().unwrap().remove(&request_id); + let namespace = self.namespaces.incoming(request_id); self.namespaces.forget(request_id); - if let Some(tx) = tx { + let Some(tx) = tx else { return }; + let Some(namespace) = namespace else { tx.push(raw); + return; + }; + + // Under the lock, so a handler dropping its stream either refuses the push or + // finds the withdrawal recorded. + let mut readvertised = self.readvertised.lock().unwrap(); + if !tx.push(raw) { + return; + } + let withdrawing = readvertised.namespaces.entry(namespace.clone()).or_default(); + match withdrawing.held.iter_mut().find(|held| held.request_id == request_id) { + Some(held) => held.withdrawn = true, + None => { + withdrawing.handling.insert(request_id); + readvertised.withdrawn.insert(request_id, namespace); + } } } @@ -849,6 +932,44 @@ impl Shared { } drop((updates, streams)); self.namespaces.forget(request_id); + self.handled(request_id); + } + + /// A withdrawn request's handler is done with it: once none of the namespace's + /// withdrawals is still being handled, hand over what was held behind them, up to and + /// including the next one withdrawn. + fn handled(&self, request_id: RequestId) { + // What a closed session refused, dropped only once the lock is released, since + // dropping a stream lands back here. + let mut refused = Vec::new(); + let mut readvertised = self.readvertised.lock().unwrap(); + let Some(namespace) = readvertised.withdrawn.remove(&request_id) else { + return; + }; + let Readvertised { namespaces, withdrawn } = &mut *readvertised; + let Some(withdrawing) = namespaces.get_mut(&namespace) else { + return; + }; + withdrawing.handling.remove(&request_id); + while withdrawing.handling.is_empty() { + let Some(held) = withdrawing.held.pop_front() else { + namespaces.remove(&namespace); + break; + }; + if let Some(stream) = self.incoming.offer(held.stream) { + // The session is gone, so nothing held behind this one is accepted either. + refused.push(stream); + refused.extend(withdrawing.held.drain(..).map(|held| held.stream)); + namespaces.remove(&namespace); + break; + } + if held.withdrawn { + withdrawing.handling.insert(held.request_id); + withdrawn.insert(held.request_id, namespace.clone()); + } + } + drop(readvertised); + drop(refused); } } @@ -1116,7 +1237,16 @@ fn classify(type_id: u64, body: &Bytes, version: Version, namespaces: &Namespace } // SubscribeNamespace on control stream (v14/v15 only) ietf::SubscribeNamespaceLegacy::ID => match version { - Version::Draft14 | Version::Draft15 => { + // Draft-14's UNSUBSCRIBE_NAMESPACE names the prefix rather than the request. + Version::Draft14 => { + let mut r = Decoder::new(body, version.into()); + let id = RequestId::decode(&mut r, version)?; + if let Ok(prefix) = crate::ietf::namespace::decode_namespace(&mut r) { + namespaces.register(Direction::Subscribed, prefix.into_owned(), id); + } + Ok(Route::NewRequest(id)) + } + Version::Draft15 => { let id = decode_request_id(body, version)?; Ok(Route::NewRequest(id)) } @@ -1264,7 +1394,11 @@ fn classify(type_id: u64, body: &Bytes, version: Version, namespaces: &Namespace _ => Err(Error::UnexpectedMessage), }, ietf::UnsubscribeNamespace::ID => match version { - Version::Draft14 | Version::Draft15 => { + Version::Draft14 => match lookup_namespace_request_id(body, version, namespaces, Direction::Subscribed)? { + Some(id) => Ok(Route::CloseStream(id)), + None => Ok(Route::Ignore), + }, + Version::Draft15 => { let id = decode_request_id(body, version)?; Ok(Route::CloseStream(id)) } @@ -1913,6 +2047,7 @@ mod tests { let ours = advertise(&shared, namespace, version, RequestId(4)).await; (ours, receive(&shared, namespace, version, RequestId(7)).await) } + Direction::Subscribed => unreachable!("a mesh only advertises"), }; (shared, ours, theirs) @@ -2076,6 +2211,7 @@ mod tests { shared.close(RequestId(7), encode_msg(&msg, version)); assert!(drained(&mut theirs, 4).await); assert!(!shared.streams.lock().unwrap().contains_key(&RequestId(7))); + drop(theirs); let _again = receive(&shared, "cluster/ns", version, RequestId(11)).await; assert!(matches!( @@ -2084,6 +2220,98 @@ mod tests { )); } + /// The peer's next request on `namespace`, advertising it, as the read loop routes it. + fn readvertise(shared: &Arc, namespace: &str, version: Version, request_id: RequestId) { + let msg = publish_namespace(request_id, namespace); + let route = classify( + ietf::PublishNamespace::ID, + &encode_body(&msg, version), + version, + &shared.namespaces, + ) + .unwrap(); + assert!(matches!(route, Route::NewRequest(id) if id == request_id)); + shared + .open_incoming(request_id, encode_msg(&msg, version), None) + .unwrap(); + } + + /// The peer withdrawing `namespace` on `request_id`, as the read loop routes it. + fn withdraw(shared: &Arc, namespace: &str, version: Version, request_id: RequestId) { + let msg = ietf::PublishNamespaceDone { + track_namespace: crate::Path::new(namespace), + request_id, + }; + let route = classify( + ietf::PublishNamespaceDone::ID, + &encode_body(&msg, version), + version, + &shared.namespaces, + ) + .unwrap(); + assert!(matches!(route, Route::CloseStream(id) if id == request_id)); + shared.close(request_id, encode_msg(&msg, version)); + } + + /// The request of the next stream `accept_bi` would hand out, if one is ready. + fn accepted(shared: &Arc) -> Option<(RequestId, VirtualRecvStream)> { + let (_, recv) = shared.incoming.pop().now_or_never()??; + let request_id = recv.request_id.lock().unwrap().unwrap(); + Some((request_id, recv)) + } + + #[moq_clock::test] + async fn a_readvertisement_waits_for_the_withdrawal_to_be_handled() { + // Each request has its own handler, so a re-advertisement handed out beside the + // withdrawal before it could be handled first and read as a second advertisement. + for version in [Version::Draft14, Version::Draft15, Version::Draft16] { + let shared = Arc::new(Shared::default()); + let first = receive(&shared, "live/media", version, RequestId(3)).await; + let other = receive(&shared, "live/other", version, RequestId(5)).await; + + withdraw(&shared, "live/media", version, RequestId(3)); + readvertise(&shared, "live/media", version, RequestId(7)); + withdraw(&shared, "live/media", version, RequestId(7)); + readvertise(&shared, "live/media", version, RequestId(9)); + // Another namespace is not held up. + withdraw(&shared, "live/other", version, RequestId(5)); + drop(other); + readvertise(&shared, "live/other", version, RequestId(11)); + assert!(matches!(accepted(&shared), Some((RequestId(11), _))), "{version:?}"); + assert!(accepted(&shared).is_none(), "{version:?}"); + + // Handling the first withdrawal hands over the next advertisement, which was + // withdrawn too, so the one after it waits for that withdrawal in turn. + drop(first); + let (id, mut second) = accepted(&shared).expect("released"); + assert_eq!(id, RequestId(7), "{version:?}"); + assert!(accepted(&shared).is_none(), "{version:?}"); + assert!(drained(&mut second, 4).await, "{version:?}"); + drop(second); + assert!(matches!(accepted(&shared), Some((RequestId(9), _))), "{version:?}"); + assert!(shared.readvertised.lock().unwrap().namespaces.is_empty(), "{version:?}"); + } + } + + #[moq_clock::test] + async fn a_session_closed_behind_a_withdrawal_drops_what_it_held() { + // Dropping a refused stream re-enters the release, which must not still hold its lock. + let version = Version::Draft14; + let shared = Arc::new(Shared::default()); + let first = receive(&shared, "live/media", version, RequestId(3)).await; + withdraw(&shared, "live/media", version, RequestId(3)); + readvertise(&shared, "live/media", version, RequestId(7)); + withdraw(&shared, "live/media", version, RequestId(7)); + readvertise(&shared, "live/media", version, RequestId(9)); + shared.incoming.close(); + + drop(first); + let readvertised = shared.readvertised.lock().unwrap(); + assert!(readvertised.namespaces.is_empty() && readvertised.withdrawn.is_empty()); + drop(readvertised); + assert!(shared.streams.lock().unwrap().is_empty()); + } + #[moq_clock::test] async fn test_local_withdrawal_releases_the_namespace() { // We withdraw our own advertisement by dropping the request that carried it, so a @@ -2126,7 +2354,7 @@ mod tests { #[test] fn namespace_removal_preserves_links_in_every_order() { - for direction in [Direction::Incoming, Direction::Outgoing] { + for direction in [Direction::Incoming, Direction::Outgoing, Direction::Subscribed] { for first in 0..4 { for second in 0..4 { for third in 0..4 { @@ -2148,7 +2376,7 @@ mod tests { } let state = namespaces.state.lock().unwrap(); assert!(state.by_request.is_empty()); - assert!(state.incoming.is_empty() && state.outgoing.is_empty()); + assert!(state.incoming.is_empty() && state.outgoing.is_empty() && state.subscribed.is_empty()); } } } diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 5c977605bf5..1490041dea3 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -10,12 +10,27 @@ use crate::{ }; use super::{ - Control, Message, Publisher, Subscriber, Version, active_count, - adapter::ControlStreamAdapter, - auth, cluster, epoch, hidden, max_age, peer, request_update, solicit, - subscriber::{is_protocol_violation, subscribe_prefixes}, + Control, Message, Publisher, Subscriber, Version, active_count, adapter::ControlStreamAdapter, auth, cluster, + epoch, hidden, max_age, peer, request_update, solicit, subscriber::is_protocol_violation, }; +/// Keep the session's SUBSCRIBE_NAMESPACE requests open as `discovery` decides, each on +/// a stream from `open`; with no subscribe half, ask nothing. Only an error ends it. +async fn run_namespaces( + subscriber: Subscriber, + discovery: Option, + open: F, +) -> Result<(), Error> +where + F: Fn() -> O, + O: std::future::Future, Error>>, +{ + match discovery { + Some(discovery) => subscriber.run_namespaces(discovery, open).await, + None => std::future::pending().await, + } +} + /// Everything one moq-transport session needs to start. pub struct Config { /// The runtime that arms the session's timers. @@ -166,19 +181,13 @@ pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> { false => None, }; - // One SUBSCRIBE_NAMESPACE per permitted prefix, like `lite::Subscriber`: the - // scope is what we may ask for, and it is not the origin's root. - // Each holds a snapshot from now, so the origin's announce consumers never read as - // synced before the request goes out. - let namespaces: Vec<_> = subscribe + // Which prefixes to SUBSCRIBE_NAMESPACE, as for moq-lite: a client asks about what + // its origin's readers watch, a server or relay link about its whole scope. Decided + // now, so the origin's announce consumers never read as synced before a request goes + // out. An unset half asks nothing. + let discovery = subscribe .as_ref() - .map(|origin| { - subscribe_prefixes(origin) - .into_iter() - .map(|prefix| (origin.snapshot(&prefix), prefix)) - .collect() - }) - .unwrap_or_default(); + .map(|origin| crate::discovery::Discovery::new(origin, client, runtime.clone())); let withdrawal = crate::session::Withdrawal::default(); let withdrawing = withdrawal.clone(); @@ -306,8 +315,6 @@ pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> { drop(tasks); let dispatch_session = virtual_session.clone(); - let sub_ns = subscriber.clone(); - let sub_ns_adapter = adapter.clone(); // Every half only ends the session on error (err_only parks on clean // completion); the task set draining is the one clean exit. @@ -337,64 +344,33 @@ pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> { // Unsolicited PUBLISH_NAMESPACE unless the peer requires solicitation; // see `Publisher::run_publish_namespaces`. let mut pub_ns_run = std::pin::pin!(err_only(publisher.clone().run_publish_namespaces())); - let mut sub_ns_run = std::pin::pin!(err_only(async { - let mut prefixes = futures::stream::FuturesUnordered::new(); - for (snapshot, prefix) in namespaces { - let mut sub_ns = sub_ns.clone(); - let sub_ns_adapter = sub_ns_adapter.clone(); - prefixes.push(async move { - let opened = async { - match version { - Version::Draft16 => { - let mut sub_ns_adapter = sub_ns_adapter; - let (send, recv) = sub_ns_adapter.open_native_bi().await?; - let mut reader = crate::coding::Reader::new( - crate::transport::boxed::RecvStream::new(recv), - version, - ); - reader.set_priority(crate::priority::RECV_CONTROL); - Ok(Stream { - writer: crate::coding::Writer::new( - crate::transport::boxed::SendStream::new(send), - version, - ), - reader, - }) - } - _ => { - Stream::open_request( - &mut crate::driver::Transport::new(sub_ns_adapter), + let sub_ns_run = run_namespaces(subscriber.clone(), discovery, { + let adapter = adapter.clone(); + move || { + let mut adapter = adapter.clone(); + async move { + match version { + Version::Draft16 => { + let (send, recv) = adapter.open_native_bi().await?; + let mut reader = crate::coding::Reader::new( + crate::transport::boxed::RecvStream::new(recv), + version, + ); + reader.set_priority(crate::priority::RECV_CONTROL); + Ok(Stream { + writer: crate::coding::Writer::new( + crate::transport::boxed::SendStream::new(send), version, - ) - .await - } + ), + reader, + }) } + _ => Stream::open_request(&mut crate::driver::Transport::new(adapter), version).await, } - .await; - let stream = match opened { - Ok(stream) => stream, - Err(Error::NoStreamCredit) => { - tracing::warn!(%prefix, "subscribe_namespace refused without stream credit"); - return Ok(()); - } - Err(err) => return Err(err), - }; - if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix, snapshot).await { - // The peer breaking the protocol is fatal, and the driver - // below turns this into the session close the draft wants. - if is_protocol_violation(&err) { - return Err(err); - } - tracing::warn!(%err, "subscribe_namespace failed, continuing without"); - } - Ok::<(), Error>(()) - }); - } - while let Some(result) = futures::StreamExt::next(&mut prefixes).await { - result?; + } } - Ok(()) - })); + }); + let mut sub_ns_run = std::pin::pin!(err_only(sub_ns_run)); let res = kio::wait(|waiter| { use std::task::Poll; @@ -521,9 +497,6 @@ pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> { }); subscriber.announces = announces; - let sub_ns_session = session.clone(); - let sub_ns = subscriber.clone(); - // When the peer's SETUP was pre-read (a gated server accept), monitor // GOAWAY on that stream here; otherwise `run_unis` does it when the SETUP // arrives on the wire. @@ -572,37 +545,14 @@ pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> { // Unsolicited PUBLISH_NAMESPACE unless the peer requires solicitation; // see `Publisher::run_publish_namespaces`. let mut pub_ns_run = std::pin::pin!(err_only(publisher.clone().run_publish_namespaces())); - let mut sub_ns_run = std::pin::pin!(err_only(async { - let mut prefixes = futures::stream::FuturesUnordered::new(); - for (snapshot, prefix) in namespaces { - let mut sub_ns = sub_ns.clone(); - let sub_ns_session = sub_ns_session.clone(); - prefixes.push(async move { - let mut sub_ns_session = sub_ns_session; - let stream = match Stream::open_request(&mut sub_ns_session, version).await { - Ok(stream) => stream, - Err(Error::NoStreamCredit) => { - tracing::warn!(%prefix, "subscribe_namespace refused without stream credit"); - return Ok(()); - } - Err(err) => return Err(err), - }; - if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix, snapshot).await { - // The peer breaking the protocol is fatal, and the driver - // below turns this into the session close the draft wants. - if is_protocol_violation(&err) { - return Err(err); - } - tracing::warn!(%err, "subscribe_namespace failed, continuing without"); - } - Ok::<(), Error>(()) - }); - } - while let Some(result) = futures::StreamExt::next(&mut prefixes).await { - result?; + let sub_ns_run = run_namespaces(subscriber.clone(), discovery, { + let session = session.clone(); + move || { + let mut session = session.clone(); + async move { Stream::open_request(&mut session, version).await } } - Ok(()) - })); + }); + let mut sub_ns_run = std::pin::pin!(err_only(sub_ns_run)); let res = kio::wait(|waiter| { use std::task::Poll; @@ -1949,6 +1899,8 @@ mod tests { // bound it: paused time makes the deadline fire the moment nothing else can run. let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + // A client asks only about what is watched. + let _watch = origin.consume().announced(); let session = crate::lite::test_transport::ScriptedSession::new(namespace_without_hop_path(VERSION).await); let log = session.log.clone(); @@ -1998,10 +1950,10 @@ mod tests { assert_eq!(reason, err.to_string()); } - /// A subscriber issues one SUBSCRIBE_NAMESPACE per PERMITTED PREFIX, and asks for - /// those prefixes rather than the root it mounts replies under. Driven through - /// `start` so the per-prefix fan-out is exercised, not just one stream in - /// isolation: a loop that opened a single stream would still satisfy a test that + /// A subscriber watching everything issues one SUBSCRIBE_NAMESPACE per PERMITTED + /// PREFIX, and asks for those prefixes rather than the root it mounts replies under. + /// Driven through `start` so the per-prefix fan-out is exercised, not just one stream + /// in isolation: a loop that opened a single stream would still satisfy a test that /// called `run_subscribe_namespace` itself. #[moq_clock::test] async fn every_permitted_prefix_gets_its_own_subscribe_namespace() { @@ -2013,6 +1965,7 @@ mod tests { let scoped = origin .scope("rootns", &scope) .expect("scope the origin to two prefixes"); + let _watch = scoped.consume().announced(); let gate = kio::Producer::new(true); let session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); @@ -2059,14 +2012,14 @@ mod tests { } /// The bytes of one legacy SUBSCRIBE_NAMESPACE, as the session writes them. - async fn subscribe_namespace_legacy(version: Version, namespace: &str) -> Vec { + async fn subscribe_namespace_legacy(version: Version, namespace: &str, request_id: RequestId) -> Vec { let log = crate::lite::test_transport::Log::default(); let mut writer = Writer::new( crate::transport::boxed::SendStream::new(crate::lite::test_transport::SinkSend::new(log.clone())), version, ); let msg = ietf::SubscribeNamespaceLegacy { - request_id: RequestId(0), + request_id, namespace: crate::Path::new(namespace), subscribe_options: ietf::SubscribeOptions::Namespace, hidden: false, @@ -2076,33 +2029,69 @@ mod tests { log.writes.lock().unwrap().clone() } - /// How many times a session with this scope writes SUBSCRIBE_NAMESPACE for it. - /// - /// `prefix: None` is an unscoped origin, whose only interest head is empty. `solicit` - /// is what the peer's SETUP declared (MoQ Solicit). - async fn asked_namespace(version: Version, prefix: Option<&str>, solicit: Option) -> usize { - let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); - let subscribe = match prefix { - Some(name) => { - let scope: crate::Patterns = [crate::Pattern::subtree(name).unwrap()].into_iter().collect(); - origin.scope("rootns", &scope).expect("scope the origin") - } - None => origin, + /// The bytes of one UNSUBSCRIBE_NAMESPACE, as the session writes them. + async fn unsubscribe_namespace(version: Version, prefix: &str, request_id: RequestId) -> Vec { + let log = crate::lite::test_transport::Log::default(); + let mut writer = Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), version); + let msg = ietf::UnsubscribeNamespace { + prefix: crate::Path::new(prefix), + request_id, }; + writer.varint(ietf::UnsubscribeNamespace::ID).await.unwrap(); + writer.encode(&msg).await.unwrap(); + log.writes.lock().unwrap().clone() + } + + /// Watch everything under `prefix`. + fn watch(origin: &crate::origin::Producer, prefix: &str) -> crate::announce::Consumer { + origin + .consume() + .scope(prefix, &crate::Patterns::from(crate::Pattern::all())) + .unwrap() + .announced() + } + /// Start a client feeding `subscribe` against a peer that answers nothing and declared + /// `solicit` (MoQ Solicit). + async fn client( + version: Version, + subscribe: crate::origin::Producer, + solicit: Option, + ) -> ( + crate::lite::test_transport::Log, + moq_clock::sim::JoinHandle>, + ) { let gate = kio::Producer::new(true); let session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); + client_on(version, session, subscribe, solicit).await + } + + /// [`client`] over `session`. + async fn client_on( + version: Version, + session: crate::lite::test_transport::SinkSession, + subscribe: crate::origin::Producer, + solicit: Option, + ) -> ( + crate::lite::test_transport::Log, + moq_clock::sim::JoinHandle>, + ) { let log = session.log.clone(); - let session = crate::driver::Transport::new(session.clone()); - let setup = Stream::open(&mut crate::driver::Transport::new(session.clone()), version) - .await - .expect("open the control stream"); + let mut session = crate::driver::Transport::new(session); + let setup = match version { + Version::Draft14 | Version::Draft15 | Version::Draft16 => Some( + Stream::open(&mut session, version) + .await + .expect("open the control stream"), + ), + _ => None, + }; let (driver, _goaway) = start(Config { runtime: crate::time::sim(), stats: Default::default(), session, - setup: Some(setup), + setup, request_id_max: None, limits: Default::default(), client: true, @@ -2125,46 +2114,222 @@ mod tests { extensions: Default::default(), }) .expect("start the session"); - let driver = moq_clock::sim::spawn(driver); + (log, moq_clock::sim::spawn(driver)) + } - let needle = subscribe_namespace_legacy(version, prefix.unwrap_or("")).await; + /// Let the driver run until `needle` reaches the wire, or give up after a while. + async fn sent(log: &crate::lite::test_transport::Log, needle: &[u8]) -> bool { for _ in 0..ANNOUNCE_TURNS { - if occurrences(&log, &needle) > 0 { - break; + if occurrences(log, needle) > 0 { + return true; } moq_clock::sleep(std::time::Duration::from_millis(1)).await; } + false + } + + /// How many times a client watching everything under `prefix` writes SUBSCRIBE_NAMESPACE + /// for it. `solicit` is what the peer's SETUP declared (MoQ Solicit). + async fn asked_namespace(version: Version, prefix: &str, solicit: Option) -> usize { + let subscribe = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let _watch = watch(&subscribe, prefix); + let (log, driver) = client(version, subscribe, solicit).await; + + let needle = subscribe_namespace_legacy(version, prefix, RequestId(0)).await; + sent(&log, &needle).await; assert!(!driver.is_finished(), "{version:?} ended the session"); assert!(log.closes().is_empty(), "{version:?} closed: {:?}", log.closes()); occurrences(&log, &needle) } + /// A client asks nothing until something is watched, then only about what is. + #[moq_clock::test] + async fn a_client_asks_only_about_what_is_watched() { + for version in [Version::Draft14, Version::Draft15, Version::Draft16, Version::Draft17] { + let subscribe = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let (log, _driver) = client(version, subscribe.clone(), Some(true)).await; + let first = subscribe_namespace_legacy(version, "cam", RequestId(0)).await; + assert!( + !sent(&log, &first).await, + "{version:?} asked before anything was watched" + ); + + let _cam = watch(&subscribe, "cam"); + assert!(sent(&log, &first).await, "{version:?} did not ask about the watch"); + let second = subscribe_namespace_legacy(version, "", RequestId(2)).await; + assert!( + !sent(&log, &second).await, + "{version:?} asked about more than the watch" + ); + } + } + + /// Once nothing has watched a prefix for the linger, the request ends. Draft-14 and 15 + /// carry it on the control stream, where only UNSUBSCRIBE_NAMESPACE tells the peer; it + /// names the prefix on draft-14 and the request on draft-15. + #[moq_clock::test] + async fn an_unwatched_prefix_is_unsubscribed_after_the_linger() { + for version in [Version::Draft14, Version::Draft15] { + let subscribe = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let cam = watch(&subscribe, "cam"); + let (log, _driver) = client(version, subscribe.clone(), None).await; + let asked = subscribe_namespace_legacy(version, "cam", RequestId(0)).await; + assert!(sent(&log, &asked).await, "{version:?}"); + let unsubscribe = unsubscribe_namespace(version, "cam", RequestId(0)).await; + + drop(cam); + moq_clock::sleep(crate::discovery::LINGER / 2).await; + assert!( + !sent(&log, &unsubscribe).await, + "{version:?} ended the request before the linger" + ); + moq_clock::sleep(crate::discovery::LINGER).await; + assert!(sent(&log, &unsubscribe).await, "{version:?} never unsubscribed"); + } + } + + /// A broader watch replaces the narrower request at a foreign peer rather than + /// overlapping it, which that peer refuses: on draft-14 and 15 the UNSUBSCRIBE_NAMESPACE + /// goes out on the ordered control stream ahead of the broader SUBSCRIBE_NAMESPACE. + #[moq_clock::test] + async fn a_broader_watch_replaces_the_narrower_at_a_foreign_peer() { + for version in [Version::Draft14, Version::Draft15] { + let subscribe = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let _front = watch(&subscribe, "cam/front"); + let (log, _driver) = client(version, subscribe.clone(), None).await; + let asked = subscribe_namespace_legacy(version, "cam/front", RequestId(0)).await; + assert!(sent(&log, &asked).await, "{version:?}"); + + let _cam = watch(&subscribe, "cam"); + let broader = subscribe_namespace_legacy(version, "cam", RequestId(2)).await; + assert!(sent(&log, &broader).await, "{version:?} never asked the broader prefix"); + let unsubscribe = unsubscribe_namespace(version, "cam/front", RequestId(0)).await; + let writes = log.writes.lock().unwrap().clone(); + let at = |needle: &[u8]| writes.windows(needle.len()).position(|window| window == needle); + assert!( + at(&unsubscribe).expect("unsubscribed the narrower request") < at(&broader).unwrap(), + "{version:?} overlapped the narrower request" + ); + } + } + + /// From draft-16 each request has a stream of its own, so at a foreign peer the broader + /// one waits for the peer to close the narrower one's, which ends with a FIN until + /// draft-19 makes it a reset: a peer that never closes it never sees the two overlap. + #[moq_clock::test] + async fn a_broader_watch_waits_for_the_narrower_to_leave_the_wire() { + for version in [ + Version::Draft16, + Version::Draft17, + Version::Draft18, + Version::Draft19, + Version::Draft22, + ] { + let subscribe = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let _front = watch(&subscribe, "cam/front"); + let (log, _driver) = client(version, subscribe.clone(), None).await; + assert!(sent(&log, b"front").await, "{version:?}"); + let opened = log.bi_opens(); + + let _cam = watch(&subscribe, "cam"); + for _ in 0..ANNOUNCE_TURNS { + moq_clock::sleep(std::time::Duration::from_millis(1)).await; + } + assert_eq!(log.bi_opens(), opened, "{version:?} overlapped the narrower request"); + let ended = log.trail().into_iter().any(|(_, sent)| match sent { + crate::lite::test_transport::Sent::Finish => super::super::request_stream::fin_cancels(version), + crate::lite::test_transport::Sent::Reset(_) => !super::super::request_stream::fin_cancels(version), + crate::lite::test_transport::Sent::Write(_) => false, + }); + assert!(ended, "{version:?} did not end the narrower request: {:?}", log.trail()); + } + } + + /// A request ended while it waits on the peer for stream credit holds nothing: the + /// broader request that replaced it goes out alone once the credit arrives. + #[moq_clock::test] + async fn a_request_ended_before_it_opened_does_not_block_the_broader_one() { + for version in [Version::Draft16, Version::Draft17, Version::Draft22] { + let subscribe = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let gate = kio::Producer::new(true); + let credit = kio::Producer::new(true); + let session = + crate::lite::test_transport::SinkSession::gated_bi(gate.consume()).with_open_bi_gate(credit.consume()); + // Credit for draft-16's control stream only, opened before the driver runs. + let (log, _driver) = client_on(version, session, subscribe.clone(), None).await; + *credit.write().ok().unwrap() = false; + let opened = log.bi_opens(); + + let _front = watch(&subscribe, "cam/front"); + for _ in 0..ANNOUNCE_TURNS { + moq_clock::sleep(std::time::Duration::from_millis(1)).await; + } + let _cam = watch(&subscribe, "cam"); + for _ in 0..ANNOUNCE_TURNS { + moq_clock::sleep(std::time::Duration::from_millis(1)).await; + } + *credit.write().ok().unwrap() = true; + assert!(sent(&log, b"cam").await, "{version:?} never asked the broader prefix"); + assert_eq!(log.bi_opens(), opened + 1, "{version:?} opened the replaced request"); + assert!(!sent(&log, b"front").await, "{version:?} sent the replaced request"); + } + } + + /// A peer that declared MoQ Solicit is ours and holds overlapping requests, so a broader + /// watch leaves the narrower request open, as on moq-lite: neither takes the other's + /// routes over before it has synced. + #[moq_clock::test] + async fn a_broader_watch_keeps_the_narrower_open_at_our_peer() { + for version in [Version::Draft14, Version::Draft15, Version::Draft16, Version::Draft17] { + let subscribe = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let _front = watch(&subscribe, "cam/front"); + let (log, _driver) = client(version, subscribe.clone(), Some(true)).await; + let asked = subscribe_namespace_legacy(version, "cam/front", RequestId(0)).await; + assert!(sent(&log, &asked).await, "{version:?}"); + + let _cam = watch(&subscribe, "cam"); + let broader = subscribe_namespace_legacy(version, "cam", RequestId(2)).await; + assert!(sent(&log, &broader).await, "{version:?} never asked the broader prefix"); + let ended = match version { + Version::Draft14 | Version::Draft15 => { + occurrences(&log, &unsubscribe_namespace(version, "cam/front", RequestId(0)).await) > 0 + } + _ => log + .trail() + .iter() + .any(|(_, sent)| !matches!(sent, crate::lite::test_transport::Sent::Write(_))), + }; + assert!(!ended, "{version:?} ended the narrower request: {:?}", log.trail()); + } + } + /// Draft-14 and draft-15 reject a zero-field track namespace, so a foreign peer (one - /// that declared no MoQ Solicit) is not asked for it, and a real prefix still goes - /// out. A peer that declared Solicit is ours: it only tells when asked, so it still + /// that declared no MoQ Solicit) is not asked for it, and a watched prefix still goes + /// out, which is how a client discovers anything through a peer that only tells when + /// asked. A peer that declared Solicit is ours: it only tells when asked, so it still /// gets the empty prefix. Draft-16 made the empty prefix legal, so it stays. #[moq_clock::test] async fn an_empty_namespace_is_not_asked_of_a_foreign_peer_before_draft_16() { for version in [Version::Draft14, Version::Draft15] { assert_eq!( - asked_namespace(version, None, None).await, + asked_namespace(version, "", None).await, 0, "{version:?} asked a foreign peer for every namespace" ); assert_eq!( - asked_namespace(version, Some("cam"), None).await, + asked_namespace(version, "cam", None).await, 1, - "{version:?} skipped a real prefix" + "{version:?} skipped a watched prefix" ); assert_eq!( - asked_namespace(version, None, Some(true)).await, + asked_namespace(version, "", Some(true)).await, 1, "{version:?} did not ask a soliciting peer for every namespace" ); } assert_eq!( - asked_namespace(Version::Draft16, None, None).await, + asked_namespace(Version::Draft16, "", None).await, 1, "draft-16 dropped the empty prefix" ); diff --git a/rs/moq-net/src/ietf/solicit.rs b/rs/moq-net/src/ietf/solicit.rs index 470577094e1..b1c07d55817 100644 --- a/rs/moq-net/src/ietf/solicit.rs +++ b/rs/moq-net/src/ietf/solicit.rs @@ -40,9 +40,9 @@ pub fn from_setup(params: &super::Parameters, _version: Version) -> Result { } } -/// UnsubscribeNamespace message (0x14): v14/v15 only (v16 uses stream close) +/// UnsubscribeNamespace message (0x14): v14/v15 only (v16 uses stream close). +/// v14 names the prefix it subscribed, v15 the request. #[derive(Clone, Debug)] -pub struct UnsubscribeNamespace { +pub struct UnsubscribeNamespace<'a> { + /// v14: the Track Namespace Prefix of the SUBSCRIBE_NAMESPACE it cancels. + pub prefix: Path<'a>, + /// v15: the Request ID of the SUBSCRIBE_NAMESPACE it cancels. pub request_id: RequestId, } -impl Message for UnsubscribeNamespace { +impl Message for UnsubscribeNamespace<'_> { const ID: u64 = 0x14; fn encode_msg(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { - self.request_id.encode(w, version)?; + match version { + Version::Draft14 => encode_namespace(w, &self.prefix)?, + Version::Draft15 => self.request_id.encode(w, version)?, + _ => return Err(EncodeError::Version), + } Ok(()) } fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { - let request_id = RequestId::decode(r, version)?; - Ok(Self { request_id }) + match version { + Version::Draft14 => Ok(Self { + prefix: decode_namespace(r)?, + request_id: RequestId(0), + }), + Version::Draft15 => Ok(Self { + prefix: Path::default(), + request_id: RequestId::decode(r, version)?, + }), + _ => Err(DecodeError::Version), + } } } @@ -370,6 +387,29 @@ mod tests { Ok((decoded, buf.is_empty())) } + /// Draft-14 names the prefix it cancels and draft-15 the request; later drafts close the + /// stream instead. + #[test] + fn unsubscribe_namespace_names_the_prefix_then_the_request() { + let msg = UnsubscribeNamespace { + prefix: Path::new("cam/front"), + request_id: RequestId(6), + }; + let mut v14 = bytes::Bytes::from(body(&msg, Version::Draft14)); + let named = crate::coding::decode_buf(&mut v14, Version::Draft14, UnsubscribeNamespace::decode_msg).unwrap(); + assert_eq!(named.prefix.as_str(), "cam/front"); + + let mut v15 = bytes::Bytes::from(body(&msg, Version::Draft15)); + let numbered = crate::coding::decode_buf(&mut v15, Version::Draft15, UnsubscribeNamespace::decode_msg).unwrap(); + assert_eq!(numbered.request_id, RequestId(6)); + + let mut buf = Vec::new(); + assert!( + msg.encode_msg(&mut Encoder::new(&mut buf, Version::Draft16.into()), Version::Draft16) + .is_err() + ); + } + fn clustered(hops: &[u64], cost: u64) -> Option { Some(AdvertParams { cluster: Some(cluster::Advert { diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 05a080104cb..377f615d7c6 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -21,6 +21,7 @@ use crate::tail::{Reading, Settle, Tail}; use kio::Lock; +mod namespaces; mod tracks; const TRACK_ALIAS_TIMEOUT: Duration = Duration::from_secs(1); @@ -805,19 +806,17 @@ pub(super) struct Subscriber { listings: Lock, } -/// The prefixes to issue SUBSCRIBE_NAMESPACE for: `origin`'s permitted scope, -/// relative to its root. -/// -/// The scope is what we may ASK the peer for; the root is where what comes back -/// MOUNTS locally. Those are independent, and only coincide when the peer shares -/// our namespace -- a peer outside it has never heard of our root, so a rooted -/// subscriber asks for its scope and mounts the replies under the root. -/// -/// Asked unconditionally: a peer with nothing to advertise answers with an empty set, -/// which costs one stream. [`Subscriber::run_subscribe_namespace`] holds back the empty -/// prefix from a foreign draft-14/15 peer. -pub(super) fn subscribe_prefixes(origin: &origin::Producer) -> Vec { - crate::model::interest_prefixes(&origin.allowed()) +/// One SUBSCRIBE_NAMESPACE on its stream. +struct NamespaceRequest { + stream: Stream, + messages: super::request_stream::Messages, + prefix: PathOwned, + /// Set once the request goes out. + request_id: Option, + /// The suffixes this stream advertised, with the advertisement each one holds, so a + /// repeat is recognized as an update to it rather than a second one (which would + /// leak the refcount). + live: HashMap, } /// The initial set of a SUBSCRIBE_NAMESPACE still to read, under MoQ Active Count. @@ -1152,7 +1151,8 @@ impl Subscriber { self.origin.stats() } - /// Send SUBSCRIBE_NAMESPACE for one prefix on a bidi stream. + /// Send SUBSCRIBE_NAMESPACE for one prefix on a bidi stream, and read what it reports + /// until the peer ends it or `live` closes. /// The caller is responsible for opening the appropriate stream type /// (virtual for v14/v15, real bidi for v16+), one per prefix. /// @@ -1163,23 +1163,60 @@ impl Subscriber { /// `snapshot` holds the origin's announce consumers under the prefix short of /// synced. It drops once the NAMESPACE messages MoQ Active Count promised are /// in, on REQUEST_OK without the extension, or when no request goes out at all. - pub async fn run_subscribe_namespace( + /// + /// [`Ended::Local`](namespaces::Ended::Local) means `live` closed, and the request is + /// off the wire by the time this returns. + async fn run_subscribe_namespace( &mut self, - mut stream: Stream, + stream: Stream, prefix: PathOwned, snapshot: crate::model::Snapshot, - ) -> Result<(), Error> { - stream = if self.version == Version::Draft16 { + live: kio::Consumer<()>, + ) -> Result { + let stream = if self.version == Version::Draft16 { stream.with_stats(self.origin.stats()) } else { super::request_stream::with_stats(stream, self.origin.stats(), self.version) }; + let mut request = NamespaceRequest { + stream, + messages: super::request_stream::Messages::new(self.session.clone(), self.control.client()), + prefix, + request_id: None, + live: HashMap::new(), + }; + let res = self.subscribe_namespace(&mut request, snapshot, &live).await; + + // The stream owns every advertisement it carried, so release them however it + // ends: a clean close, a decode error, the peer resetting it, or the session + // ending it. Without this each namespace keeps its refcount and the source never + // detaches. + for (path, advert) in std::mem::take(&mut request.live) { + if let Some(id) = advert.id { + let _ = self.stop_announce(path, id, false); + } + } + + if let Ok(namespaces::Ended::Local) = res { + self.unsubscribe_namespace(request).await; + } + res + } + + /// Send the request and read its answers until either side ends it. + async fn subscribe_namespace( + &mut self, + request: &mut NamespaceRequest, + snapshot: crate::model::Snapshot, + live: &kio::Consumer<()>, + ) -> Result { // Hidden namespaces are requested too, as on moq-lite: the session mirrors the // peer into the origin and each local reader opts in on its own. The parameter // fails decoding at a peer that doesn't know it, so it waits on the peer's SETUP // to say whether it does (MoQ Hidden). let declared = self.peer_setup.get().await; let hidden = declared.hidden; + let prefix = request.prefix.clone(); // Draft-16 is the first to allow a zero-field namespace, so the empty prefix is a // protocol violation on draft-14/15. A peer that never declared MoQ Solicit is not @@ -1190,10 +1227,24 @@ impl Subscriber { && matches!(self.version, Version::Draft14 | Version::Draft15) { tracing::debug!(version = ?self.version, "not asking a foreign peer for the empty namespace"); - return Ok(()); + return Ok(namespaces::Ended::Peer); } - let request_id = self.control.next_request_id(&self.runtime).await?; + // No ID is spent on a request that ends while it waits for credit, so the peer + // never sees a gap in our sequential request IDs. + let mut timeout = crate::time::Deadline::after(&self.runtime, Control::REQUEST_ID_TIMEOUT); + let request_id = kio::wait(|waiter| { + if live.poll_closed(waiter).is_ready() { + return Poll::Ready(Ok(None)); + } + self.control.poll_credit(waiter, &mut timeout).map(|id| id.map(Some)) + }) + .await?; + let Some(request_id) = request_id else { + return Ok(namespaces::Ended::Local); + }; + request.request_id = Some(request_id); + let stream = &mut request.stream; // Draft-18+ uses SUBSCRIBE_NAMESPACE (0x50); earlier drafts use the legacy // 0x11 message with a Subscribe Options field. @@ -1220,8 +1271,11 @@ impl Subscriber { tracing::debug!(%prefix, "subscribe_namespace sent"); // Read response - let mut messages = super::request_stream::Messages::new(self.session.clone(), self.control.client()); - let super::request_stream::Framed(type_id, body) = messages.message(&mut stream.reader).await?; + let Some(framed) = self.next_namespace_message(request, live).await? else { + return Ok(namespaces::Ended::Local); + }; + let super::request_stream::Framed(type_id, body) = + framed.ok_or_else(|| request.messages.failed(Error::ProtocolViolation))?; let mut data = body.decoder(self.version); let count = match type_id { @@ -1264,66 +1318,114 @@ impl Subscriber { epoch: declared.epoch, }; - // Suffixes live on this stream, with the advertisement each one holds, so a repeat - // is recognized as an update to it rather than a second one (which would leak the - // refcount). - let mut live: HashMap = HashMap::new(); + self.run_namespace_entries(request, &peer, negotiated, Initial { initial, snapshot }, live) + .await + } - // The stream owns every advertisement it carried, so release them however it - // ends: a clean close, a decode error, or the peer resetting it. Without this - // each namespace keeps its refcount and the source never detaches. - // - // This is what moq-lite already does, where the equivalent map is a local whose - // guards drop. - let res = self - .run_namespace_entries( - &mut stream, - &prefix, - &peer, - negotiated, - &mut live, - Initial { initial, snapshot }, - &mut messages, - ) - .await; - for (path, advert) in live { - if let Some(id) = advert.id { - let _ = self.stop_announce(path, id, false); + /// The next message on a SUBSCRIBE_NAMESPACE stream: `None` once `live` closes, + /// `Some(None)` once the peer finishes the stream. + async fn next_namespace_message( + &self, + request: &mut NamespaceRequest, + live: &kio::Consumer<()>, + ) -> Result>, Error> { + let NamespaceRequest { stream, messages, .. } = request; + kio::wait(|waiter| { + if live.poll_closed(waiter).is_ready() { + return Poll::Ready(Ok(None)); } + let mut cx = waiter.context(); + messages + .poll_next(&mut stream.reader, &mut cx) + .map(|next| next.map(Some)) + }) + .await + } + + /// End a request the session ended, and wait for it to leave the wire, so an + /// overlapping one can go out: a peer may refuse a prefix that overlaps one it still + /// holds. + /// + /// Draft-14 and 15 ride the control stream, whose virtual streams have no close of + /// their own, so UNSUBSCRIBE_NAMESPACE is all the peer ever sees, and anything after + /// it on that ordered stream reaches the peer later. Later drafts close the stream + /// (a FIN until draft-19 makes it a reset) and read until the peer closes its half. + async fn unsubscribe_namespace(&self, request: NamespaceRequest) { + let NamespaceRequest { + stream: Stream { mut writer, mut reader }, + mut messages, + prefix, + request_id, + .. + } = request; + // Nothing went out, so there is nothing to end. + let Some(request_id) = request_id else { + return; + }; + tracing::debug!(%prefix, "unsubscribe_namespace"); + + if matches!(self.version, Version::Draft14 | Version::Draft15) { + let msg = ietf::UnsubscribeNamespace { prefix, request_id }; + let written = async { + writer.varint(ietf::UnsubscribeNamespace::ID).await?; + writer.encode(&msg).await + }; + if let Err(err) = written.await { + tracing::debug!(%err, "failed to write unsubscribe_namespace"); + } + let _ = writer.close().await; + return; } - res + + match super::request_stream::fin_cancels(self.version) { + true => { + let _ = writer.finish(); + } + false => writer.abort(&Error::Cancel), + } + let mut session = self.session.clone(); + kio::wait(|waiter| { + let mut cx = waiter.context(); + if session.poll_closed(&mut cx).is_ready() { + return Poll::Ready(()); + } + loop { + match ready!(messages.poll_next(&mut reader, &mut cx)) { + Ok(Some(_)) => continue, + Ok(None) | Err(_) => return Poll::Ready(()), + } + } + }) + .await; } - /// Read NAMESPACE / NAMESPACE_DONE entries until the stream closes. + /// Read NAMESPACE / NAMESPACE_DONE entries until either side ends the request. /// - /// `live` tracks the suffixes this stream has advertised, so a repeat is recognized - /// as an update rather than a second advertisement, and the caller can release - /// whatever is still held when the stream ends. - #[expect( - clippy::too_many_arguments, - reason = "private plumbing; a single-use argument struct would only rename the parameters" - )] + /// `request.live` tracks the suffixes this stream has advertised, so a repeat is + /// recognized as an update rather than a second advertisement, and the caller can + /// release whatever is still held when the stream ends. async fn run_namespace_entries( &mut self, - stream: &mut Stream, - prefix: &PathOwned, + request: &mut NamespaceRequest, peer: &cluster::Peer, negotiated: ietf::Negotiated, - live: &mut HashMap, mut initial: Initial, - messages: &mut super::request_stream::Messages, - ) -> Result<(), Error> { + live: &kio::Consumer<()>, + ) -> Result { loop { // Every counted NAMESPACE is in the origin by now, dropped ones included. if initial.initial == 0 { initial.snapshot = None; } - let super::request_stream::Framed(type_id, body) = match messages.next(&mut stream.reader).await? { - Some(message) => message, + let super::request_stream::Framed(type_id, body) = match self.next_namespace_message(request, live).await? { + Some(Some(message)) => message, + None => return Ok(namespaces::Ended::Local), // A stream ending inside the counted set cut it short. - None if initial.initial > 0 => return Err(Error::ProtocolViolation), - None => break, + Some(None) if initial.initial > 0 => return Err(Error::ProtocolViolation), + Some(None) => return Ok(namespaces::Ended::Peer), }; + let prefix = &request.prefix; + let live = &mut request.live; let mut data = body.decoder(self.version); match type_id { @@ -1408,8 +1510,6 @@ impl Subscriber { } } } - - Ok(()) } /// Handle an incoming bidi stream dispatched by the session. @@ -6016,6 +6116,13 @@ mod tests { use super::*; + /// What keeps a SUBSCRIBE_NAMESPACE open until the peer ends it, while the first lives. + fn live() -> (kio::Producer<()>, kio::Consumer<()>) { + let live = kio::Producer::new(()); + let consumer = live.consume(); + (live, consumer) + } + #[moq_clock::test] async fn internal_alias_resolution_ends_with_its_session() { let aliases = TrackAliases::default(); @@ -6304,7 +6411,7 @@ mod tests { Default::default(), ); - let mut namespaces = subscribe_prefixes(&subscriber.origin); + let mut namespaces = crate::model::interest_prefixes(&subscriber.origin.allowed()); assert_eq!( namespaces, vec![crate::Path::new("cam").to_owned()], @@ -6312,11 +6419,12 @@ mod tests { ); let prefix = namespaces.pop().unwrap(); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), Version::Draft16) .await .unwrap(); let mut run = - std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed())); + std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed(), live)); // Parks awaiting the peer's response; the request is already on the wire. assert!(futures::poll!(run.as_mut()).is_pending()); @@ -6391,13 +6499,16 @@ mod tests { Default::default(), ); - let prefix = subscribe_prefixes(&subscriber.origin).pop().expect("one prefix"); + let prefix = crate::model::interest_prefixes(&subscriber.origin.allowed()) + .pop() + .expect("one prefix"); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); // Parks on the read after the scripted NAMESPACE is consumed. let mut run = - std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed())); + std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed(), live)); for _ in 0..100 { // The result is deliberately ignored: a regressed mount lands out of scope // and errors here, which the assertions below name far better than a poll @@ -6489,12 +6600,19 @@ mod tests { Default::default(), ); - let prefix = subscribe_prefixes(&subscriber.origin).pop().expect("one prefix"); + let prefix = crate::model::interest_prefixes(&subscriber.origin.allowed()) + .pop() + .expect("one prefix"); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); - let mut run = - std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed())); + let mut run = std::pin::pin!(subscriber.run_subscribe_namespace( + stream, + prefix, + crate::model::Snapshot::landed(), + live + )); let mut result = None; for _ in 0..100 { if let std::task::Poll::Ready(done) = futures::poll!(run.as_mut()) { @@ -6578,12 +6696,15 @@ mod tests { ); subscriber.announces = crate::session::Slots::new(1); - let prefix = subscribe_prefixes(&subscriber.origin).pop().expect("one prefix"); + let prefix = crate::model::interest_prefixes(&subscriber.origin.allowed()) + .pop() + .expect("one prefix"); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); let mut run = - std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed())); + std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, crate::model::Snapshot::landed(), live)); let mut result = None; for _ in 0..100 { if let std::task::Poll::Ready(done) = futures::poll!(run.as_mut()) { @@ -6644,6 +6765,7 @@ mod tests { Default::default(), ); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); @@ -6652,6 +6774,7 @@ mod tests { stream, crate::Path::new("").to_owned(), crate::model::Snapshot::landed(), + live, ) .await; assert!( @@ -6746,11 +6869,16 @@ mod tests { crate::lite::test_transport::ScriptedSession::new(counted_response(VERSION, Some(2), entries).await); let (mut subscriber, consumer, snapshot) = counted_subscriber(VERSION, true, session.clone()); let mut cursor = consumer.announced(); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); - let mut run = - std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("").to_owned(), snapshot)); + let mut run = std::pin::pin!(subscriber.run_subscribe_namespace( + stream, + crate::Path::new("").to_owned(), + snapshot, + live + )); for _ in 0..10 { assert!(futures::poll!(run.as_mut()).is_pending()); settle().await; @@ -6770,11 +6898,12 @@ mod tests { let (mut subscriber, consumer, snapshot) = counted_subscriber(VERSION, false, session.clone()); let cursor = consumer.announced(); assert!(!cursor.is_synced()); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); let mut run = - std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("").to_owned(), snapshot)); + std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("").to_owned(), snapshot, live)); for _ in 0..10 { assert!(futures::poll!(run.as_mut()).is_pending()); settle().await; @@ -6793,11 +6922,12 @@ mod tests { session.close(crate::lite::test_transport::Close::Fin); } let (mut subscriber, _consumer, snapshot) = counted_subscriber(VERSION, true, session.clone()); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); let res = subscriber - .run_subscribe_namespace(stream, crate::Path::new("").to_owned(), snapshot) + .run_subscribe_namespace(stream, crate::Path::new("").to_owned(), snapshot, live) .await; assert!(matches!(res, Err(Error::ProtocolViolation)), "{entries:?}: {res:?}"); } @@ -9124,6 +9254,7 @@ mod tests { Default::default(), ); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); @@ -9132,6 +9263,7 @@ mod tests { stream, crate::Path::new("").to_owned(), crate::model::Snapshot::landed(), + live, ) .await .expect("a clean FIN is not an error"); @@ -9213,13 +9345,15 @@ mod tests { Default::default(), ); + let (_live, live) = live(); let stream = Stream::open(&mut crate::driver::Transport::new(session.clone()), VERSION) .await .unwrap(); let mut run = std::pin::pin!(subscriber.run_subscribe_namespace( stream, crate::Path::new("").to_owned(), - crate::model::Snapshot::landed() + crate::model::Snapshot::landed(), + live )); for _ in 0..100 { assert!( diff --git a/rs/moq-net/src/ietf/subscriber/namespaces.rs b/rs/moq-net/src/ietf/subscriber/namespaces.rs new file mode 100644 index 00000000000..15b8f03f347 --- /dev/null +++ b/rs/moq-net/src/ietf/subscriber/namespaces.rs @@ -0,0 +1,187 @@ +//! Drives a session's SUBSCRIBE_NAMESPACE requests from [`Discovery`]. +//! +//! A client asks only about the prefixes its origin's readers watch, as moq-lite does, +//! and ends a request once nothing has watched it for a while. +//! +//! moq-transport refuses a SUBSCRIBE_NAMESPACE that shares a prefix with one still open. +//! A peer that declared MoQ Solicit is ours and holds both, so it gets moq-lite's +//! overlapping requests, which never take each other's routes over. Any other peer gets +//! exclusive discovery: a broader request replaces the narrower ones beneath it, each of +//! which leaves the wire before the broader one goes out. The routes a replaced request +//! carried go with it, and come back once the broader request reports them. + +use std::future::Future; + +use super::*; +use crate::discovery::{Change, Discovery}; + +/// How a SUBSCRIBE_NAMESPACE ended. +#[derive(Debug)] +pub(super) enum Ended { + /// The peer ended it, or it never went out. + Peer, + /// The session ended it. + Local, +} + +/// A request from when discovery opens it until it is off the wire. +struct Request { + prefix: PathOwned, + /// Dropped to end it; `None` once the session did. + live: Option>, + /// Closes once it is off the wire, so an overlapping one may go out. + gone: kio::Consumer<()>, +} + +/// A request waiting to go out. +struct Pending { + /// Close once each request it overlaps is off the wire. + blockers: Vec>, + /// Opens its stream. + opening: O, + /// Closes when the session ends it. + live: kio::Consumer<()>, + gone: kio::Producer<()>, +} + +impl Subscriber { + /// Keep one SUBSCRIBE_NAMESPACE open for each prefix `discovery` decides on, each on a + /// stream from `open`, for the session's life. Only an error ends it: a request the + /// peer breaks the protocol on. + pub(in crate::ietf) async fn run_namespaces(self, discovery: Discovery, open: F) -> Result<(), Error> + where + F: Fn() -> O, + O: Future, Error>>, + { + let mut discovery = match self.peer_setup.get().await.solicit { + Some(_) => discovery, + None => discovery.exclusive(), + }; + let mut requests: HashMap = HashMap::new(); + // The request discovery holds open for each prefix. + let mut open_at: HashMap = HashMap::new(); + let mut running = futures::stream::FuturesUnordered::new(); + let mut next = 0u64; + + loop { + enum Event { + Change(Change), + Done(T), + } + let event = kio::wait(|waiter| { + if let Poll::Ready(change) = discovery.poll(waiter) { + return Poll::Ready(Event::Change(change)); + } + let mut cx = waiter.context(); + match futures::Stream::poll_next(std::pin::Pin::new(&mut running), &mut cx) { + Poll::Ready(Some(done)) => Poll::Ready(Event::Done(done)), + _ => Poll::Pending, + } + }) + .await; + + match event { + Event::Change(Change::Open(prefix, snapshot)) => { + // Wait out the overlapping requests on their way out, such as the narrower + // ones an exclusive session just closed for this one. Those still open + // overlap it only where the peer holds both. + let blockers = requests + .values() + .filter(|request| request.live.is_none()) + .filter(|request| request.prefix.has_prefix(&prefix) || prefix.has_prefix(&request.prefix)) + .map(|request| request.gone.clone()) + .collect(); + let live = kio::Producer::new(()); + let gone = kio::Producer::new(()); + let key = next; + next += 1; + requests.insert( + key, + Request { + prefix: prefix.clone(), + live: Some(live.clone()), + gone: gone.consume(), + }, + ); + open_at.insert(prefix.clone(), key); + let pending = Pending { + blockers, + opening: open(), + live: live.consume(), + gone, + }; + running.push(self.clone().namespace_request(key, prefix, snapshot, pending)); + } + // Dropping its `live` ends the request. + Event::Change(Change::Close(prefix)) => { + if let Some(key) = open_at.remove(&prefix) + && let Some(request) = requests.get_mut(&key) + { + request.live = None; + } + } + Event::Done((key, res)) => { + let Some(request) = requests.remove(&key) else { + continue; + }; + // Ended on its own, rather than by the session. + if open_at.get(&request.prefix) == Some(&key) { + open_at.remove(&request.prefix); + discovery.ended(&request.prefix); + } + if let Err(err) = res { + // The peer breaking the protocol is fatal, and the driver turns + // this into the session close the draft wants. + if is_protocol_violation(&err) { + return Err(err); + } + tracing::warn!(%err, prefix = %request.prefix, "subscribe_namespace failed, continuing without"); + } + } + } + } + } + + /// Send one request once the ones it overlaps are off the wire, and read it until + /// either side ends it. + async fn namespace_request( + mut self, + key: u64, + prefix: PathOwned, + snapshot: crate::model::Snapshot, + pending: Pending, + ) -> (u64, Result) + where + O: Future, Error>>, + { + let Pending { + blockers, + opening, + live, + gone, + } = pending; + // The open races `live` too: a request ended before it went out holds nothing, nor + // the broader one it blocks. Without stream credit it is refused, not parked. + let mut opening = std::pin::pin!(opening); + let opened = kio::wait(|waiter| { + if live.poll_closed(waiter).is_ready() { + return Poll::Ready(None); + } + for blocker in &blockers { + if blocker.poll_closed(waiter).is_pending() { + return Poll::Pending; + } + } + waiter.poll_future(opening.as_mut()).map(Some) + }) + .await; + let res = match opened { + Some(Ok(stream)) => self.run_subscribe_namespace(stream, prefix, snapshot, live).await, + Some(Err(err)) => Err(err), + None => Ok(Ended::Local), + }; + // Off the wire: an overlapping request may go out. + drop(gone); + (key, res) + } +} diff --git a/rs/moq-net/src/lite/test_transport.rs b/rs/moq-net/src/lite/test_transport.rs index 603214e1b20..5614b0897a7 100644 --- a/rs/moq-net/src/lite/test_transport.rs +++ b/rs/moq-net/src/lite/test_transport.rs @@ -592,6 +592,9 @@ pub struct SinkSession { /// Set by [`Self::accepted_bi`]. `None` parks `accept_bi` forever, which is what /// every session that only ever opens its own streams expects. accept_gate: Option>, + /// Set by [`Self::with_open_bi_gate`] to withhold bidi stream credit. + bi_open_gate: Option>, + bi_open_park: kio::Park, /// Set by [`Self::gated_uni`] to hold unidirectional stream writes. uni_gate: Option>, /// Set by [`Self::gated_open_uni`] to withhold unidirectional stream credit. @@ -714,6 +717,12 @@ impl SinkSession { self.uni_open_gate = Some(gate); self } + + /// Also hold bidi stream opens until `gate` grants stream credit. + pub fn with_open_bi_gate(mut self, gate: kio::Consumer) -> Self { + self.bi_open_gate = Some(gate); + self + } } impl poll::Session for SinkSession { @@ -736,12 +745,19 @@ impl poll::Session for SinkSession { fn poll_open_bi( &mut self, - _cx: &mut Context<'_>, + cx: &mut Context<'_>, options: poll::SendOptions<'_, SinkGroup>, ) -> Poll, Self::Error>> { let Some(gate) = self.bi_gate.clone() else { return Poll::Pending; }; + if let Some(credit) = &self.bi_open_gate { + let waiter = self.bi_open_park.hold(cx); + match credit.poll(waiter, |open| (**open).then_some(()).map_or(Poll::Pending, Poll::Ready)) { + Poll::Ready(Ok(())) => {} + Poll::Ready(Err(_)) | Poll::Pending => return Poll::Pending, + } + } self.log.bi_opens.fetch_add(1, Ordering::Relaxed); let mut send = SinkSend::gated(self.log.clone(), gate).opened(options); diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index e1787be71de..ba6ffbb3f68 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -6151,8 +6151,8 @@ impl Consumer { /// relative to this consumer's root; its patterns only filter visibility. /// Routes with a segment starting with `.` below the literal head of those /// patterns are hidden unless [`with_hidden`](Self::with_hidden) opted in. - /// The stream also watches its scope: a moq-lite client session fed by this - /// origin asks its peer about the scope while any stream watches it. + /// The stream also watches its scope: a client session fed by this origin asks + /// its peer about the scope while any stream watches it. /// Drop the returned [`AnnounceConsumer`] to unregister. pub fn announced(&self) -> AnnounceConsumer { let state = kio::Producer::::default(); @@ -6443,7 +6443,7 @@ impl Consumer { /// /// The returned future fails with [`Error::Unroutable`] at once when no /// announced route covers the path, including a broadcast created on this - /// origin but not announced. A moq-lite client session only discovers what this + /// origin but not announced. A client session only discovers what this /// origin's readers watch, so a path no [`Self::announced`], /// [`Self::routed_broadcast`], or [`Self::announced_broadcast`] watches is /// unroutable on it. A broadcast resolved through a discovered route keeps its diff --git a/rs/moq-net/tests/integration/lazy_discovery.rs b/rs/moq-net/tests/integration/lazy_discovery.rs index 85fd2340e24..42075bce0fb 100644 --- a/rs/moq-net/tests/integration/lazy_discovery.rs +++ b/rs/moq-net/tests/integration/lazy_discovery.rs @@ -358,3 +358,147 @@ async fn serving_the_peer_asks_nothing() { )); } } + +/// Every moq-transport draft. Not every one marks where an initial set ends (draft-14 and +/// 15 answer with PUBLISH_NAMESPACE requests after the OK), so these read what arrived once +/// everything in flight has. +const IETF: &[&str] = &[ + "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", +]; + +/// Apply what `announced` reports once everything in flight has arrived to `held`, and +/// say whether any of it ended a prefix. +async fn settled(announced: &mut announce::Consumer, held: &mut Vec) -> bool { + moq_clock::sleep(SETTLE).await; + let mut ended = false; + while let Some(event) = announced.try_next() { + match event { + announce::Event::Start(announce) | announce::Event::Restart(announce) => { + held.push(announce.prefix.to_string()) + } + announce::Event::End(announce) => { + ended = true; + held.retain(|prefix| *prefix != announce.prefix.as_str()) + } + announce::Event::Update(_) => {} + } + } + held.sort(); + held.dedup(); + ended +} + +/// Nothing watched, nothing asked: the peer's broadcasts stay undiscovered. +#[moq_clock::test] +async fn ietf_asks_nothing_until_watched() { + for version in IETF { + let lazy = Lazy::new(version).await; + assert!(!lazy.routes("a/x").await, "{version}"); + } +} + +/// A watch asks about its own prefix and nothing else. +#[moq_clock::test] +async fn ietf_asks_only_about_the_watched_prefix() { + for version in IETF { + let lazy = Lazy::new(version).await; + let mut watch = lazy.watch("a"); + let mut held = Vec::new(); + settled(&mut watch, &mut held).await; + assert_eq!(held, ["x"], "{version}"); + assert!(lazy.routes("a/x").await, "{version}"); + assert!(!lazy.routes("b/y").await, "{version}"); + } +} + +/// The request outlives its last watcher by the linger, then ends, and what it reported +/// goes with it. +#[moq_clock::test] +async fn ietf_lingers_after_the_last_watcher() { + for version in IETF { + let lazy = Lazy::new(version).await; + let mut watch = lazy.watch("a"); + settled(&mut watch, &mut Vec::new()).await; + drop(watch); + + moq_clock::sleep(Duration::from_secs(1)).await; + assert!(lazy.routes("a/x").await, "{version}: still lingering"); + + let again = lazy.watch("a"); + moq_clock::sleep(LINGERED).await; + assert!(lazy.routes("a/x").await, "{version}: rewatched"); + + drop(again); + moq_clock::sleep(LINGERED).await; + assert!(!lazy.routes("a/x").await, "{version}: ended"); + } +} + +/// A broader watch after a narrower one discovers everything, and retracts nothing the +/// narrower one holds. +#[moq_clock::test] +async fn ietf_a_broader_watch_retracts_nothing() { + for version in IETF { + let lazy = Lazy::new(version).await; + let mut narrower = lazy.watch("a"); + let mut held = Vec::new(); + settled(&mut narrower, &mut held).await; + assert_eq!(held, ["x"], "{version}"); + + let mut root = lazy.watch(""); + let mut all = Vec::new(); + settled(&mut root, &mut all).await; + assert_eq!(all, ["a/x", "b/y"], "{version}"); + assert!( + !settled(&mut narrower, &mut held).await, + "{version}: retracted {held:?}" + ); + + drop(root); + moq_clock::sleep(LINGERED).await; + assert!(lazy.routes("a/x").await, "{version}: the narrower watch still holds it"); + assert!( + !settled(&mut narrower, &mut held).await, + "{version}: retracted {held:?}" + ); + + drop(narrower); + moq_clock::sleep(LINGERED * 2).await; + assert!(!lazy.routes("a/x").await, "{version}"); + assert!(!lazy.routes("b/y").await, "{version}"); + } +} + +/// Waiting on a path asks about it. +#[moq_clock::test] +async fn ietf_waiting_on_a_path_asks_about_it() { + for version in IETF { + let lazy = Lazy::new(version).await; + let consumer = lazy.subscriber.consume(); + moq_clock::timeout(SETTLE, consumer.routed_broadcast("a/x")) + .await + .expect("routed in time") + .unwrap(); + assert!(!lazy.routes("b/y").await, "{version}"); + } +} + +/// A relay's link still asks for everything it may. +#[moq_clock::test] +async fn ietf_a_relay_link_asks_for_everything() { + for version in IETF { + let relay = produce_origin(2); + let _lazy = Lazy::with(version, relay.clone().with_link(moq_net::origin::Link::Peer)).await; + for path in ["a/x", "b/y"] { + relay.consume().request_broadcast(path, None).await.unwrap(); + } + } +}