feat(moq-hls): Producer/Consumer cursors for recording a broadcast

[kixelated]

Supersedes #2381 (which exposed raw observers); please review this instead.

Why

The export module is built to serve HLS on demand. A consumer that records a broadcast (mirrors every rendition + segment to storage) needs the same state the server has, but #2381 exposed it as raw internals:

• Broadcaster::updated() -> watch::Receiver<usize> and Rendition::updated() -> watch::Receiver<u64> put internal encodings (rendition count, timeline epoch) in the public API. The doc comment had to say "the value is the count, so go re-read renditions()" -- a signal masquerading as a value.
• Because only signals + snapshot getters were exposed, every consumer re-derived the same state machine: reconcile-by-Arc::ptr_eq, a generation counter, and diffing successive playlist() snapshots to find newly-closed groups. Segment discovery is this crate's logic; having each consumer rebuild it is the bug factory.

What

Two cursors in the moq-net idiom (a handle that owns its cursor and yields the next element; None = done). No try_next/poll_next/is_closed -- nothing needed them.

// Broadcaster::renditions() -> renditions::Consumer
pub enum Event { Added(Arc<Rendition>), Removed { kind: Kind, name: String } }
pub async fn next(&mut self) -> Option<Event>;

// Rendition::segments() -> segments::Consumer
pub struct Segment { pub group: u64, pub media: Bytes, pub duration: f64,
                     pub program_date_time: Option<SystemTime>, pub discontinuity: bool }
pub async fn next(&mut self) -> Result<Option<Segment>>;
pub async fn init(&self) -> Result<Option<Bytes>>;

• renditions::{Producer, Consumer} replaces the snapshot getter + count-watch + closed()/is_closed(). The Producer owns the current set (the catalog reconcile moves off Broadcaster); the Consumer diffs against its own view, so it replays the current set as Added and a reconfigure is a Removed then an Added -- callers no longer do pointer-identity detection.
• segments::{Producer, Consumer} rehomes timeline.rs. The Producer is the timeline window, now backed by kio::Producer instead of a tokio::sync::watch (also in the direction of refactor(net)!: drop moq-net's direct tokio dependency #2377). The Consumer's next() awaits the next finalized segment, FETCHes and transmuxes it through the same on-demand path the serve path uses (so no new standing traffic), and yields the CMAF bytes. Evicted segments are skipped and flagged via discontinuity -- the caller stops re-deriving gaps. A transient fetch error is returned without advancing the cursor, so the caller can retry the same segment.

The serve path keeps its pull API unchanged in behavior, but it is now gated behind the server feature (only in-crate callers use it), so a recording-only build's surface is just the cursors. No watch types remain in the public API.

Waiting is timeout-free -- bounding a wait is the caller's policy, not the primitive's, so Broadcaster::ready() and Rendition::playable() just resolve and the serve path wraps them in its own tokio::time::timeout. Those follow moq-net's sync-predicate/async-wait pairing (is_closed()/closed()): Rendition::is_playable() is the bool, playable() is the wait.

Validation

cargo test -p moq-hls 44/44 and clippy clean with the server feature on, and clean as a dependency with it off. Includes a new end-to-end test driving both cursors (renditions -> segments -> media bytes -> run dry).

Consumer-side, this was written against a real rewrite of moq.pro's VOD recorder: it deletes that recorder's ptr_eq reconcile, generation-based re-enumeration, and playlist-snapshot diffing (16 recorder + 31 package tests green). That port also surfaced a latent issue worth calling out: on a reconfigure the predecessor rendition's cursor does not end, because its source media continues -- the old snapshot-diff loop had been depending on the timeline ending to release its state. Consumers that hand off state across a reconfigure need to drive that themselves.

Risk

Additive for recording; the serve path's behavior is unchanged (its methods only changed visibility/feature-gating). timeline.rs -> segments.rs is a rehome, with its window/eviction logic and tests carried over intact.

(written by Opus 4.8)

[sourcery-ai]

Sorry @kixelated, you have reached your weekly rate limit of 500000 diff characters.

Please try again later or upgrade to continue using Sourcery