Split OriginConsumer into cheap read handle and announcement cursor#1434
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (12)
WalkthroughThis PR refactors the origin announcement streaming API by separating the cheap read handle ( 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
✨ Simplify code
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
106d480 to
9588bfd
Compare
OriginConsumer used to bundle a tree-read handle with an mpsc cursor and a registered ConsumerId. Every clone (and every transient .consume() / .scope() / .with_root() chain) paid for a fresh channel plus a consume_initial() walk that replayed every active broadcast as backlog. Callers like get_broadcast, announced_broadcast, and the FFI wrappers paid that cost just to look up one path or re-root a handle. Split announcement subscription into AnnounceProducer and AnnounceConsumer. OriginConsumer is now a cheap, Clone-friendly read handle; call .announced() on it to allocate the channel and register the cursor. AnnounceConsumer owns the Drop-side unregistration. The deprecated OriginProducer::get_broadcast shim and its libmoq Updates the lite/ietf publishers, FFI, libmoq, relay, clock, moq-boy, hang examples, and integration tests to grab a cursor explicitly (consumer.announced() -> AnnounceConsumer::next/try_next). Adds a regression test that cloning OriginConsumer does not drain another cursor's channel and that a freshly-built AnnounceConsumer still receives the active backlog. https://claude.ai/code/session_01BSuKtGPPntcBMEQiGb3Pn6
9588bfd to
f2b8dd9
Compare
Rebasing onto dev picked up #1495 (moq-mux::Error is thiserror, not anyhow) and #1434 (OriginConsumer lost announced_broadcast, gained get_broadcast/announced). Track both in our error enum and binary; the WHIP-client path is gated on the per-codec re-packetizer anyway, so first-miss is good enough until that lands. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…sor API broadcast.rs's linger_resubscribe_keeps_flowing_moq_lite_03 still awaited OriginConsumer::announced() as a future, but #1434 made it return an AnnounceConsumer cursor. The three sibling tests were updated; this one was missed, leaving the workspace test build broken on dev. Mirror the sibling pattern: build the cursor up front and poll it with .next().
Reconcile main into dev. Key conflict resolutions: - conducer crate renamed to kio (main #1547): applied across all of dev's newer code; dropped the stale conducer path-dep, kept dev's new flate2 dep. - moq-mux: kept dev's thiserror Result (#1495); dropped main's CatalogSource as dead code since dev's catalog::Consumer already unifies Hang/MSF. - moq-net: kept dev's OriginConsumer/AnnounceConsumer split (#1434) and the TrackConsumer end_at cap; kept dev's non-optional auto-created origins on the lite session/publisher (#e770). - stats: combined main's StatsConfig + liveness retention (#1537, #1548) with dev's AnnounceConsumer usage. - libmoq + moq-native: kept main's auto-reconnect (#1544), terminal-callback contract (#1546), and consume_announced (#1552), adapted to dev's AnnounceConsumer and OriginProducer connect API. Restored the InitFailed error variant and made moq-rtc handle the now-fallible Log::init. cargo check/clippy/test all pass on the merged workspace.
Reconciles the PR with dev's landed work: * `moq-net: runtime Timescale/Timestamp; container::Frame keeps source scale (#1473)` — adopts dev's canonical `Timescale(NonZero<u64>)` and `Timestamp` design. The PR keeps `Track.timescale: Option<Timescale>`, `Frame.timestamp: Option<Timestamp>`, and `SubscribeOk.timescale` for the Lite05 wire format on top. * `moq-mux: catalog filter/target and Annex-B exporters (#1487)` and the surrounding mux reorganization (codec/* + container/{fmp4,mkv,loc} split). The PR's per-file Lite05 additions to these layers are dropped in favor of dev's structure; per-frame timestamp encoding stays opt-in at the moq-net layer until the mux gets explicit timescale support. * `Split OriginConsumer into cheap read handle and announcement cursor (#1434)` and other broadcast/origin updates land via dev. * `moq-net: add Lite05Wip version variant (unadvertised) (#1518)` — the PR's wire format additions now hang off dev's `Lite05Wip` name (was `Lite05` in the PR), keeping a single ALPN identifier (`moq-lite-05-wip`) across both sides of the merge. * Lite05 publisher logic and Lite05 subscriber timestamp decoding kept from the PR. `SubscribeOk.timescale: Option<Timescale>` carries the negotiated scale on Lite05+; older versions decode it as `None`. * `subscribe_track` reverts to dev's sync `(&Track)` signature. The PR's async `Subscription`-based variant is dropped here because it cascaded async through every mux exporter and tipped over dev's poll-driven exporters. Timescale negotiation still works: the publisher carries it on the `Track` it creates, and Lite05 subscribers stash the value from `SubscribeOk` on the `TrackEntry` before frame decoding begins. * `lite05_timestamp_roundtrip` integration test ported to dev's API and passes over WebTransport (negative delta frames included). JS net (`js/net`) is still at Lite04, so the Lite05 wire field doesn't need a JS-side update in this merge. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Summary
Refactor the announcement subscription API to separate concerns:
OriginConsumeris now a cheap, cloneable read handle over the broadcast tree, whileAnnounceConsumeris the stateful cursor that receives announcement events. This eliminates the expensive per-clone channel allocation and allows multiple independent announcement streams from the same consumer.Key Changes
OriginConsumerrefactored to be cheap to clone: Removed the per-instancempsc::UnboundedReceiverandConsumerId. Cloning now shares the underlying tree state without allocating any per-cursor resources.New
AnnounceConsumertype: Extracted the announcement stream logic into a separate struct that holds the channel receiver and consumer ID. Created viaOriginConsumer::announced()orAnnounceProducer::consume().New
AnnounceProducertype: Symmetric counterpart toAnnounceConsumer. Provides a cheap read handle to the announcement stream for a subtree, similar to howOriginConsumerworks for the broadcast tree.API method renames:
OriginConsumer::announced()now returnsAnnounceConsumer(was async method returningOption<OriginAnnounce>)OriginConsumer::try_announced()→AnnounceConsumer::try_next()OriginConsumer::announced()(async) →AnnounceConsumer::next()(async)AnnounceConsumer::is_closed()to check cursor stateOriginProducer::announces()method: New method to get anAnnounceProducerfor the subtree, providing symmetric API toconsume().Updated internal struct: Renamed
OriginConsumerNotifytoAnnounceConsumerNotifyfor clarity.Documentation improvements: Updated doc comments to clarify the cheap-to-clone nature of
OriginConsumerand the allocation semantics ofAnnounceConsumer.Implementation Details
OriginConsumernow derivesCloneand implements it directly (no custom impl needed).AnnounceConsumermaintains the original per-cursor state: channel receiver, consumer ID, and root path.assert_next,assert_try_next, etc.) moved toAnnounceConsumerimpl block..announced()onOriginConsumerto get anAnnounceConsumerbefore awaiting announcements.test_consumer_clone_is_side_effect_freeto verify that cloningOriginConsumerdoes not drain announcement channels and that freshAnnounceConsumerinstances still receive the active backlog.https://claude.ai/code/session_01BSuKtGPPntcBMEQiGb3Pn6