From 557e0cb4090954b6927097de7ed80b73beaa4734 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 23 Oct 2026 00:56:26 -0700 Subject: [PATCH 1/9] chore(quest): claim quest/m1/ietf-lazy-discovery.md Quest-Claim: b0d9aedee400e57e From beaa6dd9e153779cca35b2c8443726776cc4b5c3 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 23 Oct 2026 01:39:16 -0700 Subject: [PATCH 2/9] feat(net): ask a moq-transport peer only about watched prefixes Drive SUBSCRIBE_NAMESPACE from the origin's watched prefixes, as moq-lite does, on every draft. End a request with UNSUBSCRIBE_NAMESPACE on drafts 14/15 (prefix-keyed on 14, as the draft says) or by closing its stream. A peer that did not declare MoQ Solicit never sees overlapping requests. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EeyiEy5FWz3z44DYZVo7vU --- cpp/moq/MIGRATE.md | 4 +- dart/moq/MIGRATE.md | 4 +- dart/moq_ffi/MIGRATE.md | 4 +- doc/concept/standard.md | 2 +- doc/lib/rs/moq-net.md | 2 +- go/ffi/MIGRATE.md | 4 +- go/wrapper/MIGRATE.md | 4 +- kt/moq-ffi/MIGRATE.md | 4 +- kt/moq/MIGRATE.md | 4 +- py/moq-ffi/MIGRATE.md | 4 +- py/moq-rs/MIGRATE.md | 4 +- rs/moq-c/MIGRATE.md | 4 +- rs/moq-c/src/api.rs | 4 +- rs/moq-ffi/MIGRATE.md | 4 +- rs/moq-ffi/src/origin.rs | 4 +- rs/moq-net/MIGRATE.md | 6 +- rs/moq-net/src/discovery.rs | 44 ++- rs/moq-net/src/fuzz.rs | 2 +- rs/moq-net/src/ietf/adapter.rs | 32 +- rs/moq-net/src/ietf/session.rs | 355 +++++++++++++----- rs/moq-net/src/ietf/solicit.rs | 6 +- rs/moq-net/src/ietf/subscribe_namespace.rs | 29 +- rs/moq-net/src/ietf/subscriber.rs | 284 ++++++++++---- rs/moq-net/src/ietf/subscriber/namespaces.rs | 189 ++++++++++ rs/moq-net/src/model/origin.rs | 6 +- .../tests/integration/lazy_discovery.rs | 144 +++++++ swift/MIGRATE.md | 4 +- swift/ffi/MIGRATE.md | 4 +- 28 files changed, 926 insertions(+), 235 deletions(-) create mode 100644 rs/moq-net/src/ietf/subscriber/namespaces.rs diff --git a/cpp/moq/MIGRATE.md b/cpp/moq/MIGRATE.md index f7e71829805..541e202d8aa 100644 --- a/cpp/moq/MIGRATE.md +++ b/cpp/moq/MIGRATE.md @@ -17,10 +17,10 @@ To keep a value's moq-lite byte, subtract 128 from it, widening first (`static_c ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced` or `announced_broadcast`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced` or `announced_broadcast`, instead of under its whole scope at connect. `OriginConsumer::request_broadcast` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announced_broadcast`, or hold `announced` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. ### `moq::ContainerFormat::kFmp4` unwraps to Legacy (#6250) diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index 2421cf9d9c8..ca2b3c222fb 100644 --- a/dart/moq/MIGRATE.md +++ b/dart/moq/MIGRATE.md @@ -23,10 +23,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. ### `ContainerFormat.fmp4` unwraps to Legacy (#6250) diff --git a/dart/moq_ffi/MIGRATE.md b/dart/moq_ffi/MIGRATE.md index 81942441036..5b502e73d52 100644 --- a/dart/moq_ffi/MIGRATE.md +++ b/dart/moq_ffi/MIGRATE.md @@ -19,10 +19,10 @@ Regenerate the bindings for the changed checksums. ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch diff --git a/doc/concept/standard.md b/doc/concept/standard.md index a8df3896d0a..bd2af32b008 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. A peer that did not declare solicit never holds two overlapping requests of ours: 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. Each paused `PUBLISH` holds a subscription slot, 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; any other end, including the error our own answerer sends for an unlisted track, 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 d9295f9ad4a..422856e91b4 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 a `peer()` or `upstream()` origin handle, 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 a `peer()` or `upstream()` origin handle, 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/go/ffi/MIGRATE.md b/go/ffi/MIGRATE.md index 8b8121ffdd1..67e7a6703ea 100644 --- a/go/ffi/MIGRATE.md +++ b/go/ffi/MIGRATE.md @@ -19,10 +19,10 @@ Regenerate the bindings for the changed checksums. ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `Announced()` or `AnnouncedBroadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `Announced()` or `AnnouncedBroadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.RequestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `AnnouncedBroadcast()`, or hold `Announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index b76850ea132..f00ad93b3d2 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -23,10 +23,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `Announced()` or `AnnouncedBroadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `Announced()` or `AnnouncedBroadcast()`, instead of under its whole scope at connect. `OriginConsumer.RequestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `AnnouncedBroadcast()`, or hold `Announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. ### `media.ContainerFormatFmp4` unwraps to Legacy (#6250) diff --git a/kt/moq-ffi/MIGRATE.md b/kt/moq-ffi/MIGRATE.md index 26889124bb9..4ecf55f0dd5 100644 --- a/kt/moq-ffi/MIGRATE.md +++ b/kt/moq-ffi/MIGRATE.md @@ -12,10 +12,10 @@ Regenerate the bindings for the changed `resolve` checksum. ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index 995d262f220..21a0a53edb7 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -23,10 +23,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. ### `ContainerFormat.FMP4` unwraps to Legacy (#6250) diff --git a/py/moq-ffi/MIGRATE.md b/py/moq-ffi/MIGRATE.md index 44d3a96c07a..6f5bb2adec4 100644 --- a/py/moq-ffi/MIGRATE.md +++ b/py/moq-ffi/MIGRATE.md @@ -12,10 +12,10 @@ Regenerate the bindings for the changed `resolve` checksum. ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announced_broadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced()` or `announced_broadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.request_broadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announced_broadcast()`, or hold `announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index b8ec8a192fa..ed4378a59d8 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -23,10 +23,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announced_broadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced()` or `announced_broadcast()`, instead of under its whole scope at connect. `OriginConsumer.request_broadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announced_broadcast()`, or hold `announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. ### `ContainerFormat.FMP4` unwraps to Legacy (#6250) diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index 73a236f8faa..c2947610e84 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -23,10 +23,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `moq_origin_announced` or `moq_origin_announced_broadcast`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `moq_origin_announced` or `moq_origin_announced_broadcast`, instead of under its whole scope at connect. `moq_origin_request` watches nothing, so on a client it fails with a negative code for a path nothing watches, even when the relay announces it. Resolve the path with `moq_origin_announced_broadcast`, or hold `moq_origin_announced` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. ### `MOQ_CONTAINER_FORMAT_FMP4` unwraps to Legacy (#6250) diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index aa7170afaa1..96ff9cf28fb 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -2101,8 +2101,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/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index 63a09626841..a086a2336cf 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -24,10 +24,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced` or `announced_broadcast`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced` or `announced_broadcast`, instead of under its whole scope at connect. `MoqOriginConsumer::request_broadcast` watches nothing, so on a client it fails with `MoqError::Unroutable` for a path nothing watches, even when the relay announces it. Resolve the path with `announced_broadcast`, or hold `announced` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. ### `MoqContainerFormat::Fmp4` unwraps to Legacy (#6250) diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 6bbc6af4af4..274b8e0fa81 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -498,8 +498,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/MIGRATE.md b/rs/moq-net/MIGRATE.md index cbd8d10165a..0cc3685a5cc 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -98,16 +98,16 @@ The publisher's `Info::priority` now breaks ties between subscriptions at one su On the wire, moq-lite carries `priority + 128` and moq-transport `127 - priority`, so an unset priority goes out as 128 on moq-lite (was 127) and 127 on moq-transport (was 128), and an absent moq-transport priority reads as 0. -### A client session asks only about what its readers watch (#6165) +### A client session asks only about what its readers watch (#6165, #6530) -On moq-lite, a client session no longer asks its peer for announcements under its whole subscribe scope at connect. +A client session no longer asks its peer for announcements under its whole subscribe scope at connect. It asks only under the prefixes its subscribe origin's readers watch, with `announced()`, `follow`, `routed`, `routed_broadcast`, or `announced_broadcast`, and stops asking 2s after nothing at or beneath a prefix is watched. `request_broadcast` watches nothing, so on a client it fails with `Error::Unroutable` for a path no reader watches, even when the peer announces it. Resolve a path with `routed_broadcast` or `announced_broadcast`, or hold an `announced()` stream over its prefix while requesting, the way `routed` already did after connecting. A broadcast resolved through a discovered route keeps its path watched until dropped. A server session, and a client fed into a `Producer::peer` or `Producer::upstream` handle, still asks for everything it may, so a relay discovers as before. Serving another session's request is not a watch, so a client that relays what it discovers to its own sessions should feed its client session `origin.upstream()` (or `peer()`) instead of a plain handle. -moq-transport sessions, and moq-lite-03/04 sessions (whose announce streams mark no initial set), are unchanged. +moq-lite-03/04 sessions, whose announce streams mark no initial set, are unchanged. ### Frames carry their arrival time 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 2c752971a7f..51a3b2222f2 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 bbab81b152e..4148952a25c 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -605,13 +605,17 @@ impl crate::transport::poll::Lease for Adapt // === 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 @@ -621,6 +625,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, @@ -630,6 +636,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 @@ -655,6 +662,7 @@ impl NamespacesState { match direction { Direction::Outgoing => &mut self.outgoing, Direction::Incoming => &mut self.incoming, + Direction::Subscribed => &mut self.subscribed, } } } @@ -1112,7 +1120,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)) } @@ -1260,7 +1277,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)) } @@ -1898,6 +1919,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) @@ -2111,7 +2133,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 { @@ -2133,7 +2155,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 ca2e15e11d3..2533e21fcf9 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -8,12 +8,28 @@ 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 + S: crate::transport::poll::Boxable, + 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. @@ -157,19 +173,13 @@ where 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(); @@ -294,8 +304,6 @@ where drop(tasks); let dispatch_session = adapter.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. @@ -324,41 +332,27 @@ where // 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 stream = match version { + 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 mut sub_ns_adapter = sub_ns_adapter; - let (send, recv) = sub_ns_adapter.open_native_bi().await?; + let (send, recv) = adapter.open_native_bi().await?; let mut reader = crate::coding::Reader::new(recv, version); reader.set_priority(crate::priority::RECV_CONTROL); - Stream { + Ok(Stream { writer: crate::coding::Writer::new(send, version), reader, - } - } - _ => Stream::open(&mut sub_ns_adapter.clone(), version).await?, - }; - 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"); + _ => Stream::open(&mut adapter, version).await, } - 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; @@ -485,9 +479,6 @@ where }); 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. @@ -536,30 +527,14 @@ where // 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 = Stream::open(&mut sub_ns_session, version).await?; - 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(&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; @@ -1803,6 +1778,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(); @@ -1850,10 +1827,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() { @@ -1865,6 +1842,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()); @@ -1909,11 +1887,11 @@ 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::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, @@ -1923,32 +1901,55 @@ 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 mut session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); let log = session.log.clone(); - let setup = Stream::open(&mut session, version) - .await - .expect("open the control stream"); + 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, @@ -1969,46 +1970,192 @@ 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 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), + } } } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index e8835fbebc4..d8864c748db 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -19,6 +19,7 @@ use crate::tail::{Reading, Settle, Tail}; use kio::Lock; +mod namespaces; mod tracks; const TRACK_ALIAS_TIMEOUT: Duration = Duration::from_secs(1); @@ -799,19 +800,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. @@ -1146,7 +1145,8 @@ where 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. /// @@ -1157,23 +1157,60 @@ where /// `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 @@ -1184,10 +1221,24 @@ where && 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. @@ -1216,8 +1267,11 @@ where 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 { @@ -1260,63 +1314,114 @@ where 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. - #[allow(clippy::too_many_arguments)] + /// `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 { @@ -1398,8 +1503,6 @@ where } } } - - Ok(()) } /// Handle an incoming bidi stream dispatched by the session. @@ -6003,6 +6106,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(); @@ -6291,7 +6401,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()], @@ -6299,9 +6409,10 @@ mod tests { ); let prefix = namespaces.pop().unwrap(); + let (_live, live) = live(); let stream = Stream::open(&mut 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()); @@ -6373,11 +6484,14 @@ 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 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 @@ -6467,10 +6581,17 @@ 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 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()) { @@ -6528,12 +6649,14 @@ mod tests { Default::default(), ); + let (_live, live) = live(); let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); let res = subscriber .run_subscribe_namespace( stream, crate::Path::new("").to_owned(), crate::model::Snapshot::landed(), + live, ) .await; assert!( @@ -6629,9 +6752,14 @@ 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 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; @@ -6651,9 +6779,10 @@ 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 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; @@ -6672,9 +6801,10 @@ 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 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:?}"); } @@ -8915,12 +9045,14 @@ mod tests { Default::default(), ); + let (_live, live) = live(); let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); subscriber .run_subscribe_namespace( stream, crate::Path::new("").to_owned(), crate::model::Snapshot::landed(), + live, ) .await .expect("a clean FIN is not an error"); @@ -9000,11 +9132,13 @@ mod tests { Default::default(), ); + let (_live, live) = live(); let stream = Stream::open(&mut 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..71f183df141 --- /dev/null +++ b/rs/moq-net/src/ietf/subscriber/namespaces.rs @@ -0,0 +1,189 @@ +//! 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 +where + S: crate::transport::poll::Boxable, +{ + /// 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; + let proceed = kio::wait(|waiter| { + if live.poll_closed(waiter).is_ready() { + return Poll::Ready(false); + } + for blocker in &blockers { + if blocker.poll_closed(waiter).is_pending() { + return Poll::Pending; + } + } + Poll::Ready(true) + }) + .await; + let res = match proceed { + true => match opening.await { + Ok(stream) => self.run_subscribe_namespace(stream, prefix, snapshot, live).await, + Err(err) => Err(err), + }, + false => Ok(Ended::Local), + }; + // Off the wire: an overlapping request may go out. + drop(gone); + (key, res) + } +} diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 128800ae229..2421b68e93a 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -6125,8 +6125,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(); @@ -6418,7 +6418,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 92cbc1ca2e4..022f97fdaf4 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().peer()).await; + for path in ["a/x", "b/y"] { + relay.consume().request_broadcast(path, None).await.unwrap(); + } + } +} diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 976a5a725ad..896696f8620 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -23,10 +23,10 @@ To keep a value's moq-lite byte, subtract 128 from it, widening first so it does ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `OriginConsumer.requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. ### `ContainerFormat.fmp4` unwraps to Legacy (#6250) diff --git a/swift/ffi/MIGRATE.md b/swift/ffi/MIGRATE.md index f27cf7ce292..bda91abf24b 100644 --- a/swift/ffi/MIGRATE.md +++ b/swift/ffi/MIGRATE.md @@ -12,10 +12,10 @@ Regenerate the bindings for the changed `resolve` checksum. ### A client asks only about what it watches (#6165) -On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-transport and moq-lite-03/04 sessions are unchanged. +moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch From 57871cc4da6d1409c6d469e7eb15bb8edbc6eea1 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 23 Oct 2026 01:45:16 -0700 Subject: [PATCH 3/9] fix(net): name the prefix in a draft-14 UNSUBSCRIBE_NAMESPACE Draft 14 carries the Track Namespace Prefix, not a request ID; route an incoming one by prefix in both adapters, and leave the JS overlap fix as the rest of the quest. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EeyiEy5FWz3z44DYZVo7vU --- doc/concept/standard.md | 2 +- js/net/src/ietf/adapter.ts | 48 ++++++++++++++------ js/net/src/ietf/ietf.test.ts | 14 ++++++ js/net/src/ietf/publisher.test.ts | 2 +- js/net/src/ietf/subscribe_namespace.ts | 32 ++++++++----- js/net/src/ietf/subscriber.ts | 2 +- quest/m1/README.md | 2 +- quest/m1/audit-interop.md | 8 ++-- quest/m1/ietf-lazy-discovery.md | 52 +++++++++------------- quest/m2/keyed-prefix-polling.md | 4 -- rs/moq-net/src/ietf/subscribe_namespace.rs | 23 ++++++++++ 11 files changed, 122 insertions(+), 67 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index bd2af32b008..eb3719e41e4 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. A client asks only for the prefixes its readers watch, and a server or relay link for every prefix it may discover. A peer that did not declare solicit never holds two overlapping requests of ours: 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. +- **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. Each paused `PUBLISH` holds a subscription slot, 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; any other end, including the error our own answerer sends for an unlisted track, 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/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index 281dc96f587..7493668544c 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 }; } @@ -1110,11 +1128,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 }; } @@ -1167,6 +1188,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 e3231756f45..d3f1853c579 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 35ef3a24553..b998d0b7a75 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -2558,7 +2558,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 8a4ca40b462..f3fecd2bd1d 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -561,7 +561,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 55f947f4924..9c54c0ccb5e 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -148,7 +148,7 @@ tracks, so per-group stats stop costing cluster announces. - [FromStr parsers](/quest/m1/from-str-trait.md) - hang, moq-auth, and moq-msf implement `FromStr` instead of inherent `from_str` - [Drop dead items](/quest/m1/drop-dead-items.md) - the audit's uncalled public items and test helpers are removed - [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 -- [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 - [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 - [Runner nightly](/quest/m1/interop-runner-nightly.md) - a nightly job runs the upstream interop runner against our relay and clients, so newly registered implementations with a published image are tested without a change here - [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/audit-interop.md b/quest/m1/audit-interop.md index 4d93aee879e..42754540ebf 100644 --- a/quest/m1/audit-interop.md +++ b/quest/m1/audit-interop.md @@ -54,10 +54,9 @@ released: - moxygen: the Rust client offered `moq-00` over WebTransport, so moxygen picked draft-14 over draft-16. That cell passed live from `main`. The - draft-14 QUIC cell stays red until [IETF lazy discovery](/quest/m1/ietf-lazy-discovery.md) - lands: the CLI asks draft-14 for no namespace, so it discovers nothing - from moxygen. A CLI subtree scope was tried and dropped, since it lost - the sibling broadcasts a catalog references. + draft-14 QUIC cell 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)). - js/net dialed `moqt://` as a WebTransport CONNECT with `:scheme: moqt`, which moq-relay refuses; the runner's moq-dev-js `remote-quic` cells failed with an empty message. @@ -120,4 +119,3 @@ waits on [JavaScript FETCH](/quest/m1/js-fetch.md). - [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 diff --git a/quest/m1/ietf-lazy-discovery.md b/quest/m1/ietf-lazy-discovery.md index aa9f21c0653..c976a3e081b 100644 --- a/quest/m1/ietf-lazy-discovery.md +++ b/quest/m1/ietf-lazy-discovery.md @@ -1,37 +1,27 @@ -# [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. + +Public API: none expected. Wire: none. 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-net/src/ietf/subscribe_namespace.rs b/rs/moq-net/src/ietf/subscribe_namespace.rs index 07d7f151227..79cb3233061 100644 --- a/rs/moq-net/src/ietf/subscribe_namespace.rs +++ b/rs/moq-net/src/ietf/subscribe_namespace.rs @@ -387,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 { From 8ad038030f59d76f4a667e59c574a24b3881091c Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Fri, 23 Oct 2026 01:52:29 -0700 Subject: [PATCH 4/9] chore(dart): regenerate bindings for the request_broadcast doc Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EeyiEy5FWz3z44DYZVo7vU --- dart/moq_ffi/lib/src/moq.dart | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index a7f3a9a6bf6..b893f081e23 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -19609,7 +19609,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) { From fe0992a607fbe3012cffaeac0d439f25ea6cca7e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 10:37:57 -0700 Subject: [PATCH 5/9] chore(quest): record lazy discovery decisions; fmt merged test Co-Authored-By: Claude Opus 5.5 --- quest/m1/ietf-lazy-discovery.md | 14 ++++++++++++++ rs/moq-net/src/ietf/subscriber.rs | 8 ++------ 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/quest/m1/ietf-lazy-discovery.md b/quest/m1/ietf-lazy-discovery.md index c976a3e081b..aa86be17d8e 100644 --- a/quest/m1/ietf-lazy-discovery.md +++ b/quest/m1/ietf-lazy-discovery.md @@ -24,4 +24,18 @@ narrower request open under a broader one on every version. - `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/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 5197b5460b6..168a04148d9 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -6687,12 +6687,8 @@ mod tests { .expect("one prefix"); let (_live, live) = live(); let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); - let mut run = std::pin::pin!(subscriber.run_subscribe_namespace( - stream, - prefix, - crate::model::Snapshot::landed(), - live - )); + 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()) { From a6afd36a4ac16e311475903be3f046d0f13900ca Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 23 Oct 2026 13:17:48 -0700 Subject: [PATCH 6/9] fix(net): end a namespace request still waiting for stream credit A request discovery closes while its stream open waits on the peer's credit now drops the open, so it holds no state and never blocks the broader request that replaced it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/session.rs | 43 ++++++++++++++++++++ rs/moq-net/src/ietf/subscriber/namespaces.rs | 19 +++++---- rs/moq-net/src/lite/test_transport.rs | 18 +++++++- 3 files changed, 70 insertions(+), 10 deletions(-) diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 9d7d69b6033..4ae49daa621 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1946,6 +1946,19 @@ mod tests { ) { 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 mut session = crate::driver::Transport::new(session); let setup = match version { @@ -2114,6 +2127,36 @@ mod tests { } } + /// 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. diff --git a/rs/moq-net/src/ietf/subscriber/namespaces.rs b/rs/moq-net/src/ietf/subscriber/namespaces.rs index 71f183df141..89e120910c4 100644 --- a/rs/moq-net/src/ietf/subscriber/namespaces.rs +++ b/rs/moq-net/src/ietf/subscriber/namespaces.rs @@ -163,24 +163,25 @@ where live, gone, } = pending; - let proceed = kio::wait(|waiter| { + // The open waits on stream credit the peer grants, so it races `live` too: a request + // ended before it went out holds nothing, nor the broader one it blocks. + let mut opening = std::pin::pin!(opening); + let opened = kio::wait(|waiter| { if live.poll_closed(waiter).is_ready() { - return Poll::Ready(false); + return Poll::Ready(None); } for blocker in &blockers { if blocker.poll_closed(waiter).is_pending() { return Poll::Pending; } } - Poll::Ready(true) + waiter.poll_future(opening.as_mut()).map(Some) }) .await; - let res = match proceed { - true => match opening.await { - Ok(stream) => self.run_subscribe_namespace(stream, prefix, snapshot, live).await, - Err(err) => Err(err), - }, - false => Ok(Ended::Local), + 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); diff --git a/rs/moq-net/src/lite/test_transport.rs b/rs/moq-net/src/lite/test_transport.rs index 0a7644ad1c3..2e5305b4ae8 100644 --- a/rs/moq-net/src/lite/test_transport.rs +++ b/rs/moq-net/src/lite/test_transport.rs @@ -580,6 +580,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. @@ -702,6 +705,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 { @@ -724,12 +733,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); From f17b07c5905a73e817c9f0d5c1e8a3de3c2fb60b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 00:59:35 -0700 Subject: [PATCH 7/9] docs: move migration notes to the PR's Impact section Main no longer writes MIGRATE.md per PR (#6610); the breaking cut writes it from each PR's Impact section. Co-Authored-By: Claude Opus 5.5 --- cpp/moq/MIGRATE.md | 4 ++-- dart/moq/MIGRATE.md | 4 ++-- dart/moq_ffi/MIGRATE.md | 4 ++-- go/ffi/MIGRATE.md | 4 ++-- go/wrapper/MIGRATE.md | 4 ++-- kt/moq-ffi/MIGRATE.md | 4 ++-- kt/moq/MIGRATE.md | 4 ++-- py/moq-ffi/MIGRATE.md | 4 ++-- py/moq-rs/MIGRATE.md | 4 ++-- rs/moq-c/MIGRATE.md | 4 ++-- rs/moq-ffi/MIGRATE.md | 4 ++-- rs/moq-net/MIGRATE.md | 6 +++--- swift/MIGRATE.md | 4 ++-- swift/ffi/MIGRATE.md | 4 ++-- 14 files changed, 29 insertions(+), 29 deletions(-) diff --git a/cpp/moq/MIGRATE.md b/cpp/moq/MIGRATE.md index d9989fe886a..50c37346b2f 100644 --- a/cpp/moq/MIGRATE.md +++ b/cpp/moq/MIGRATE.md @@ -37,10 +37,10 @@ To keep a value's moq-lite byte, subtract 128 from it, widening first (`static_c ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced` or `announced_broadcast`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced` or `announced_broadcast`, instead of under its whole scope at connect. `OriginConsumer::request_broadcast` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announced_broadcast`, or hold `announced` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. ### `moq::ContainerFormat::kFmp4` unwraps to Legacy (#6250) diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index 56235f1917e..fcae653ce62 100644 --- a/dart/moq/MIGRATE.md +++ b/dart/moq/MIGRATE.md @@ -50,10 +50,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. ### `ContainerFormat.fmp4` unwraps to Legacy (#6250) diff --git a/dart/moq_ffi/MIGRATE.md b/dart/moq_ffi/MIGRATE.md index 7a5b704b969..3d827bd5232 100644 --- a/dart/moq_ffi/MIGRATE.md +++ b/dart/moq_ffi/MIGRATE.md @@ -53,10 +53,10 @@ Regenerate the bindings for the changed checksums. ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch diff --git a/go/ffi/MIGRATE.md b/go/ffi/MIGRATE.md index fa139d62621..a9d2c9beff3 100644 --- a/go/ffi/MIGRATE.md +++ b/go/ffi/MIGRATE.md @@ -43,10 +43,10 @@ Regenerate the bindings for the changed checksums. ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `Announced()` or `AnnouncedBroadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `Announced()` or `AnnouncedBroadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.RequestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `AnnouncedBroadcast()`, or hold `Announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index ba9a1cd5b4f..68bab16c90b 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -43,10 +43,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `Announced()` or `AnnouncedBroadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `Announced()` or `AnnouncedBroadcast()`, instead of under its whole scope at connect. `OriginConsumer.RequestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `AnnouncedBroadcast()`, or hold `Announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. ### `media.ContainerFormatFmp4` unwraps to Legacy (#6250) diff --git a/kt/moq-ffi/MIGRATE.md b/kt/moq-ffi/MIGRATE.md index 2913e2e9c67..81fb99c14cd 100644 --- a/kt/moq-ffi/MIGRATE.md +++ b/kt/moq-ffi/MIGRATE.md @@ -36,10 +36,10 @@ Regenerate the bindings for the changed `resolve` checksum. ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index 6586ca4b37c..f15d69691be 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -43,10 +43,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. ### `ContainerFormat.FMP4` unwraps to Legacy (#6250) diff --git a/py/moq-ffi/MIGRATE.md b/py/moq-ffi/MIGRATE.md index c56206902fe..a71e2fe9f9b 100644 --- a/py/moq-ffi/MIGRATE.md +++ b/py/moq-ffi/MIGRATE.md @@ -36,10 +36,10 @@ Regenerate the bindings for the changed `resolve` checksum. ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced()` or `announced_broadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announced_broadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.request_broadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announced_broadcast()`, or hold `announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index 6c65e6631da..cb060da1319 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -43,10 +43,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced()` or `announced_broadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announced_broadcast()`, instead of under its whole scope at connect. `OriginConsumer.request_broadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announced_broadcast()`, or hold `announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. ### `ContainerFormat.FMP4` unwraps to Legacy (#6250) diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index 4a301f7a0c1..2139f8017c6 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -43,10 +43,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `moq_origin_announced` or `moq_origin_announced_broadcast`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `moq_origin_announced` or `moq_origin_announced_broadcast`, instead of under its whole scope at connect. `moq_origin_request` watches nothing, so on a client it fails with a negative code for a path nothing watches, even when the relay announces it. Resolve the path with `moq_origin_announced_broadcast`, or hold `moq_origin_announced` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. ### `MOQ_CONTAINER_FORMAT_FMP4` unwraps to Legacy (#6250) diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index 778a79c191f..970199b6486 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -49,10 +49,10 @@ To keep a value's moq-lite byte, subtract 128 from it: 255 becomes 127, 128 beco ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced` or `announced_broadcast`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced` or `announced_broadcast`, instead of under its whole scope at connect. `MoqOriginConsumer::request_broadcast` watches nothing, so on a client it fails with `MoqError::Unroutable` for a path nothing watches, even when the relay announces it. Resolve the path with `announced_broadcast`, or hold `announced` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. ### `MoqContainerFormat::Fmp4` unwraps to Legacy (#6250) diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index 2fc5e29a3ed..a2d342bdfe6 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -127,16 +127,16 @@ The publisher's `Info::priority` now breaks ties between subscriptions at one su On the wire, moq-lite carries `priority + 128` and moq-transport `127 - priority`, so an unset priority goes out as 128 on moq-lite (was 127) and 127 on moq-transport (was 128), and an absent moq-transport priority reads as 0. -### A client session asks only about what its readers watch (#6165, #6530) +### A client session asks only about what its readers watch (#6165) -A client session no longer asks its peer for announcements under its whole subscribe scope at connect. +On moq-lite, a client session no longer asks its peer for announcements under its whole subscribe scope at connect. It asks only under the prefixes its subscribe origin's readers watch, with `announced()`, `follow`, `routed`, `routed_broadcast`, or `announced_broadcast`, and stops asking 2s after nothing at or beneath a prefix is watched. `request_broadcast` watches nothing, so on a client it fails with `Error::Unroutable` for a path no reader watches, even when the peer announces it. Resolve a path with `routed_broadcast` or `announced_broadcast`, or hold an `announced()` stream over its prefix while requesting, the way `routed` already did after connecting. A broadcast resolved through a discovered route keeps its path watched until dropped. A server session, and a client fed into a `Producer::peer` or `Producer::upstream` handle, still asks for everything it may, so a relay discovers as before. Serving another session's request is not a watch, so a client that relays what it discovers to its own sessions should feed its client session `origin.upstream()` (or `peer()`) instead of a plain handle. -moq-lite-03/04 sessions, whose announce streams mark no initial set, are unchanged. +moq-transport sessions, and moq-lite-03/04 sessions (whose announce streams mark no initial set), are unchanged. ### Frames carry their arrival time diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index a3679b5a310..df8d029a81c 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -43,10 +43,10 @@ To keep a value's moq-lite byte, subtract 128 from it, widening first so it does ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `OriginConsumer.requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. ### `ContainerFormat.fmp4` unwraps to Legacy (#6250) diff --git a/swift/ffi/MIGRATE.md b/swift/ffi/MIGRATE.md index 463090199ff..1869c0a2795 100644 --- a/swift/ffi/MIGRATE.md +++ b/swift/ffi/MIGRATE.md @@ -36,10 +36,10 @@ Regenerate the bindings for the changed `resolve` checksum. ### A client asks only about what it watches (#6165) -A client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. +On moq-lite, a client session asks its relay for announcements only under paths something watches, through `announced()` or `announcedBroadcast()`, instead of under its whole scope at connect. `MoqOriginConsumer.requestBroadcast()` watches nothing, so on a client it fails with an unroutable error for a path nothing watches, even when the relay announces it. Resolve the path with `announcedBroadcast()`, or hold `announced()` over its prefix while requesting. -moq-lite-03/04 sessions are unchanged. +moq-transport and moq-lite-03/04 sessions are unchanged. Regenerate the bindings for the changed `request_broadcast` checksum. ### Catalog references carry their epoch From c1d5feaab5b189925411481312af8a0d461ae0c4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 01:13:27 -0700 Subject: [PATCH 8/9] fix(net): apply a namespace's withdrawal before its re-advertisement The control-stream adapter hands each moq-transport request to its own handler, so a PUBLISH_NAMESPACE that followed a PUBLISH_NAMESPACE_DONE for the same namespace could be handled first, read as a second advertisement, and hide the replacement: the route never ended and started again. Asking only about watched prefixes made it routine, since each request's announcements now interleave on the wire. Hold the re-advertisement until the withdrawn request's handler drops its stream. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/adapter.rs | 175 ++++++++++++++++++++++++++++++++- 1 file changed, 173 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index 94bfae2222d..df99468f261 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}, }; @@ -710,6 +710,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() @@ -768,6 +775,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 { @@ -807,6 +847,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); } @@ -833,9 +884,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); + } } } @@ -857,6 +926,34 @@ 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) { + 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); + return; + }; + if held.withdrawn { + withdrawing.handling.insert(held.request_id); + withdrawn.insert(held.request_id, namespace.clone()); + } + // A closed queue means the session is gone, and the stream with it. + let _ = self.incoming.push(held.stream); + } } } @@ -2088,6 +2185,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!( @@ -2096,6 +2194,79 @@ 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 test_local_withdrawal_releases_the_namespace() { // We withdraw our own advertisement by dropping the request that carried it, so a From 3608f3b18634d97613c62536e0ff614d541522b7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 01:49:09 -0700 Subject: [PATCH 9/9] fix(net): drop a held re-advertisement outside the withdrawal lock A held stream the closed session refuses re-entered the release when dropped, under the lock it still held. Drop what the session refused once the lock is released. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/adapter.rs | 45 ++++++++++++++++++++++++++++++---- 1 file changed, 40 insertions(+), 5 deletions(-) diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index df99468f261..21f5de980b9 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -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. @@ -933,6 +939,9 @@ impl Shared { /// 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; @@ -945,15 +954,22 @@ impl Shared { while withdrawing.handling.is_empty() { let Some(held) = withdrawing.held.pop_front() else { namespaces.remove(&namespace); - return; + 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()); } - // A closed queue means the session is gone, and the stream with it. - let _ = self.incoming.push(held.stream); } + drop(readvertised); + drop(refused); } } @@ -2267,6 +2283,25 @@ mod tests { } } + #[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