Skip to content

moq-lite-05: implement FETCH for past groups via TrackConsumer::fetch#1601

Open
kixelated wants to merge 2 commits into
devfrom
claude/silly-colden-96e186
Open

moq-lite-05: implement FETCH for past groups via TrackConsumer::fetch#1601
kixelated wants to merge 2 commits into
devfrom
claude/silly-colden-96e186

Conversation

@kixelated
Copy link
Copy Markdown
Collaborator

@kixelated kixelated commented Jun 3, 2026

What

Implements a one-shot fetch on TrackConsumer, so you can retrieve a single past group without holding a live subscription:

TrackConsumer::fetch(group: u64, options: impl Into<Option<Fetch>>) -> Result<GroupConsumer>
pub struct Fetch { pub priority: u8 }   // defaults priority 0

If the group is already cached it's returned directly; otherwise the request bridges to a wire moq-lite FETCH, blocking on a new FETCH_OK (mirroring how subscribe blocks on SUBSCRIBE_OK). The FETCH was previously a wire-only definition the publisher rejected with UnexpectedStream.

Why

The only way to read a group was to subscribe to a track and walk its cache. There was no first-class way to grab one past group. fetch lives on TrackConsumer (not a live subscription) on purpose: you no longer have to subscribe just to fetch.

How it works

The model→wire bridge reuses the existing dynamic track-request machinery, but for a single group:

TrackConsumer::fetch
  └─ BroadcastConsumer::request_fetch(name, group, Fetch)
       ├─ cache hit (complete/in-progress) → return GroupConsumer
       ├─ cached but aborted → bypass → dynamic group request
       ├─ not cached / no live track → dynamic group request
       └─ no dynamic handler → NotFound
            └─ FetchPending (Future → GroupConsumer)

BroadcastDynamic::requested_group() -> GroupRequest        (relay subscriber drains this)
  └─ GroupRequest::accept(timescale) -> GroupProducer      (filled from the wire)

The FETCH response streams on the same bidirectional control stream as the request (like IETF MoQT): the subscriber writes Fetch; the publisher replies FetchOk then streams the group's frames; stream FIN ends the group. Errors propagate via stream reset.

Changes

  • model (model/broadcast.rs): Fetch options struct, a dynamic group-request channel (request_fetch / FetchPending / GroupRequest / BroadcastDynamic::requested_group), and TrackConsumer::fetch. GroupConsumer::timescale() accessor (model/group.rs).
  • wire (lite/fetch.rs): new FetchOk (group echo + compression + timescale); Fetch.frame_start (lite-05+). The publisher honors frame_start by skipping earlier frames.
  • publisher (lite/publisher.rs): recv_fetch / run_fetch serve the group's frames on the fetch stream (gated to lite-05+).
  • subscriber (lite/subscriber.rs): serve group requests by issuing FETCH upstream and routing frames into the producer that resolves the fetcher.
  • relay (moq-relay/src/web.rs): GET /fetch/<broadcast>/<track>?group=N uses fetch() instead of subscribe(). ?group=latest keeps the subscribe path (fetch needs a concrete group).

Out of scope (follow-ups)

  • frame_start resume reseed. The publisher honors frame_start on the wire, but fetch() always requests 0; an aborted cached group is bypassed with a full re-fetch. True resume (request frame_start = N, stitch onto the N cached frames) needs a GroupProducer that begins at a logical frame offset — a separate model change.
  • Compressed fetch. v1 streams fetched groups uncompressed (FetchOk.compression = None); the model GroupConsumer carries no per-track compress hint.
  • JS + docs. Cross-Package Sync pairs rs/moq-net wire/API with js/net (which already has a lite/fetch.ts stub) and doc/concept. Deferred to a follow-up.

Built on #1595

#1595 (now merged into dev) added the Fetch.frame_start wire field. dev has been merged in and the fetch.rs overlap reconciled: this PR keeps #1595's field + tests and adds FetchOk plus the serving logic that actually uses frame_start.

Test plan

  • cargo test -p moq-net — wire roundtrip (FetchOk, frame_start gating) + 9 model unit tests (cache hit, miss-queues-request, past-final NotFound, aborted-cache bypass, coalescing, no-dynamic NotFound, deny, drop-cancels, dynamic-drop-aborts).
  • cargo test -p moq-native --test broadcast — new broadcast_moq_lite_05_fetch_webtransport end-to-end fetch over a real session (WebTransport; Lite05Wip isn't ALPN-advertised so raw QUIC is excluded, matching the other Lite05 tests).
  • cargo test -p moq-relay, cargo clippy --workspace --all-targets, cargo fmt --check.

Targets dev per branch targeting (wire-protocol change under rs/moq-net).

🤖 Generated with Claude Code

(Written by Claude)

kixelated and others added 2 commits June 2, 2026 17:55
Add a one-shot `TrackConsumer::fetch(group, options) -> GroupConsumer` that
retrieves a single past group without holding a subscription. A cached group is
returned directly; otherwise the request bridges to a wire moq-lite FETCH,
blocking on a new FETCH_OK (mirroring SUBSCRIBE/SUBSCRIBE_OK).

- model: a dynamic group-request channel parallel to dynamic track requests
  (request_fetch / FetchPending / BroadcastDynamic::requested_group / GroupRequest).
  Cache hit returns the group; an aborted cached group is bypassed.
- wire: new FetchOk message; Fetch.frame_start (lite-05+); the publisher honors
  frame_start by skipping earlier frames.
- publisher: recv_fetch / run_fetch serve the group's frames on the fetch stream.
- subscriber: serve group requests by issuing FETCH upstream and routing frames
  into the producer that resolves the fetcher.
- relay web.rs: /fetch?group=N uses fetch() instead of subscribe().

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…y-colden-96e186

# Conflicts:
#	rs/moq-net/src/lite/fetch.rs
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant