Replace mpsc with conducer for coalesced origin consumer updates

[kixelated]

Replace the unbounded mpsc channel in OriginConsumer with a conducer-backed state machine that coalesces redundant announce/unannounce pairs. This bounds the pending update queue for slow consumers and simplifies the delivery logic.

Summary

The previous implementation used tokio::sync::mpsc::UnboundedReceiver to queue all announce/unannounce events for consumers. A slow consumer that doesn't drain between updates could accumulate many redundant messages (e.g., announce, unannounce, announce of the same path). The new implementation uses conducer::Producer to manage a coalesced state where at most one pending update exists per path.

Key Changes

• Replaced mpsc with conducer: OriginConsumerNotify now holds a conducer::Producer<OriginConsumerState> instead of an mpsc::UnboundedSender. The OriginConsumer holds the producer directly for polling.

• Added PendingUpdate enum: Represents the four possible states for a queued update:

  • Announce(BroadcastConsumer): a new broadcast at a path
  • Unannounce: removal of a broadcast
  • UnannounceAnnounce(BroadcastConsumer): a removal followed by a new broadcast (preserves the signal that the origin changed)

• Added OriginConsumerState: Manages coalescing logic with:

  • pending: HashMap<PathOwned, PendingUpdate>: at most one update per path
  • queue: VecDeque<PathOwned>: FIFO delivery order across paths
  • apply_announce() and apply_unannounce() methods that collapse redundant updates
  • take() method that yields the next update to deliver, skipping stale tombstones

• Updated announced() and try_announced(): Now poll the conducer state directly instead of receiving from a channel.

• Fixed test expectations: Updated test_128 (now test_many_announces) to reflect that the conducer-backed implementation can deliver all updates synchronously without the previous mpsc limitation. Removed the #[should_panic] attribute and added proper assertions.

• Added coalescing tests: New test suite validates the coalescing behavior:

  • test_coalesce_announce_then_unannounce: unobserved announce+unannounce collapses to nothing
  • test_coalesce_announce_unannounce_announce: multiple updates to the same path coalesce to the final state
  • test_coalesce_unannounce_announce_preserved: unannounce followed by announce of a different broadcast is preserved as two deliveries
  • test_coalesce_churn_bounded: rapid churn on a single path keeps the pending set bounded to one entry

Implementation Details

The coalescing logic preserves correctness by distinguishing between updates the consumer has already observed and those still pending. When an announce is replaced before delivery, it's simply overwritten. When an unannounce arrives after an unobserved announce, the pair collapses. However, when an announce follows an unannounce that the consumer must see, both are preserved as UnannounceAnnounce to signal the origin changed.

The queue uses a FIFO deque to maintain cross-path delivery order matching publish order, with stale entries (from collapsed pairs) skipped during take().

https://claude.ai/code/session_01KZDLekCBPsYpQzjYCdbfnj

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 4b2f3ee2-ef61-40cc-9e34-bc802df9c880

📥 Commits

Reviewing files that changed from the base of the PR and between f177e60 and 451db79.

📒 Files selected for processing (1)

• rs/moq-net/src/model/origin.rs

---

Walkthrough

This PR refactors the update-delivery mechanism for OriginConsumer by replacing a Tokio mpsc::UnboundedReceiver with a conducer::Producer that coalesces per-path announce/unannounce events. It adds a PendingUpdate state machine and OriginConsumerState keyed by PathOwned to collapse stale sequences and provide deterministic lexicographic delivery ordering. OriginConsumerNotify now writes transitions into the shared producer. OriginConsumer::announced(), try_announced(), and a new poll_announced() are reimplemented to use conducer waiting/polling. Tests were updated and new regression tests verify coalescing behavior and bounded pending updates. Path<'a> now derives PartialOrd and Ord.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title accurately and specifically describes the main change: replacing mpsc with conducer for coalesced updates in OriginConsumer.
Description check	✅ Passed	The description is well-detailed and directly related to the changeset, explaining the motivation, key changes, and implementation details of the mpsc-to-conducer refactoring.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch claude/slow-origin-consumers-gt5ay

✨ Simplify code

• Create PR with simplified code
• Commit simplified code in branch claude/slow-origin-consumers-gt5ay

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@rs/moq-net/src/model/origin.rs`:
- Around line 2094-2101: Replace the brittle check that expects exactly one
immediate try_announced() with logic that drains all responses from
consumer.try_announced() into a collection and then assert that the collection's
length is <= 1 and that every announced path (if any) equals Path::new("test");
this accepts zero entries (fully coalesced) and one entry (including the
PendingUpdate::UnannounceAnnounce case which can produce two successive
announcements) and prevents flapping due to cleanup progress — use
consumer.try_announced() in a loop to collect results, then assert
collected.len() <= 1 and collected.iter().all(|p| p.0 == Path::new("test")).
- Around line 2001-2013: These tests (test_coalesce_announce_then_unannounce,
test_coalesce_announce_unannounce_announce,
test_coalesce_unannounce_announce_preserved,
test_coalesce_unannounce_announce_unannounce, test_coalesce_churn_bounded) use
tokio::time::sleep(...) but never call tokio::time::pause(), making them depend
on real-world timing; fix each test by calling tokio::time::pause() near the
start of the async test (before any sleeps) and then replace or supplement the
sleep usage with tokio::time::advance(Duration::from_millis(...)) so the test
scheduler advances virtual time deterministically while keeping the existing
assertions in functions like Origin::random().produce(), origin.consume(), and
origin.publish_broadcast(...) unchanged.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 8ac6637c-847b-42e9-aa3a-efb23eba1b8a

📥 Commits

Reviewing files that changed from the base of the PR and between 9021565 and f177e60.

📒 Files selected for processing (2)

• rs/moq-net/src/model/origin.rs
• rs/moq-net/src/path.rs