Configure delivery classes for data channel#160
Conversation
📝 WalkthroughWalkthroughThe change adds configurable data-channel delivery classes, propagates reliability and ordering through topic metadata, buffers reliable packets during RTC backpressure, and tests reliable ordered forwarding in the simulator. ChangesData-channel delivery semantics
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Publisher
participant ParticipantCore
participant RTCDataChannel
participant Subscriber
Publisher->>ParticipantCore: publish reliable packet
ParticipantCore->>RTCDataChannel: write packet
RTCDataChannel-->>ParticipantCore: signal backpressure
ParticipantCore->>ParticipantCore: queue packet in ReliableBacklog
ParticipantCore->>RTCDataChannel: flush queued packets
RTCDataChannel->>Subscriber: deliver ordered payload
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pulsebeam-agent/src/agent/driver.rs (1)
837-860: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winReject delivery changes for an already-declared topic.
The early return reuses the existing channel without verifying
delivery. For example, declaring a topic as lossy and then requestingReliableOrderedsucceeds but retains the lossy configuration. Store and compare the existing delivery settings, or explicitly recreate the channel.🤖 Prompt for 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. In `@pulsebeam-agent/src/agent/driver.rs` around lines 837 - 860, Update ensure_data_topic to compare the requested delivery with the existing topic’s delivery settings before returning its ChannelId. Reject mismatches with the appropriate AgentError, while preserving the current reuse behavior when delivery matches; apply this consistently to both publish and subscribe topic maps.
🤖 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 `@pulsebeam-agent/src/agent/driver.rs`:
- Around line 112-129: The ReliableOrdered send path must retain publisher
messages when channel writes return Ok(false) under backpressure. Update the
relevant forwarding/write logic near DeliveryClass::ReliableOrdered and the
referenced publisher-write section to queue locally and retry rejected writes,
matching the existing forwarding-side behavior while preserving immediate
handling of successful writes and errors.
In `@pulsebeam/src/participant/core.rs`:
- Around line 221-224: Update flush_reliable_backlog and its callers to return
distinct Flushed, Blocked, and Closed outcomes instead of a boolean; preserve
each outcome through channel-missing or closed handling, and change the packet
path around enqueue_reliable so the current packet is re-enqueued only for
Blocked, not for Closed.
---
Outside diff comments:
In `@pulsebeam-agent/src/agent/driver.rs`:
- Around line 837-860: Update ensure_data_topic to compare the requested
delivery with the existing topic’s delivery settings before returning its
ChannelId. Reject mismatches with the appropriate AgentError, while preserving
the current reuse behavior when delivery matches; apply this consistently to
both publish and subscribe topic maps.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 914ac1e0-e333-44b9-8822-712641926268
📒 Files selected for processing (5)
pulsebeam-agent/src/agent/driver.rspulsebeam-agent/src/agent/mod.rspulsebeam-simulator/src/tests/data_channel.rspulsebeam/src/participant/core.rspulsebeam/src/track.rs
| #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] | ||
| pub enum DeliveryClass { | ||
| #[default] | ||
| Lossy, | ||
| SemiReliable { | ||
| retransmits: u16, | ||
| }, | ||
| ReliableOrdered, | ||
| } | ||
|
|
||
| impl DeliveryClass { | ||
| fn channel_settings(self) -> (bool, Reliability) { | ||
| match self { | ||
| Self::Lossy => (false, Reliability::MaxRetransmits { retransmits: 0 }), | ||
| Self::SemiReliable { retransmits } => { | ||
| (false, Reliability::MaxRetransmits { retransmits }) | ||
| } | ||
| Self::ReliableOrdered => (true, Reliability::Reliable), |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Buffer locally backpressured ReliableOrdered publisher writes.
ReliableOrdered configures SCTP reliability, but the existing send path discards ch.write(...) == Ok(false). Under RTC backpressure, accepted publisher messages therefore vanish before reaching the reliable channel. Queue and retry these writes, as the forwarding side does.
Also applies to: 324-329
🤖 Prompt for 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.
In `@pulsebeam-agent/src/agent/driver.rs` around lines 112 - 129, The
ReliableOrdered send path must retain publisher messages when channel writes
return Ok(false) under backpressure. Update the relevant forwarding/write logic
near DeliveryClass::ReliableOrdered and the referenced publisher-write section
to queue locally and retry rejected writes, matching the existing
forwarding-side behavior while preserving immediate handling of successful
writes and errors.
| if !self.flush_reliable_backlog(cid) { | ||
| self.enqueue_reliable(cid, pkt.to_vec()); | ||
| return; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Distinguish backpressure from terminal flush failure.
false means both “temporarily blocked” and “channel missing/closed.” The caller consequently re-enqueues the current packet after flush_reliable_backlog has removed or closed the channel, recreating stale backlog state. Return separate Flushed, Blocked, and Closed outcomes, and enqueue only for Blocked.
Also applies to: 244-276
🤖 Prompt for 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.
In `@pulsebeam/src/participant/core.rs` around lines 221 - 224, Update
flush_reliable_backlog and its callers to return distinct Flushed, Blocked, and
Closed outcomes instead of a boolean; preserve each outcome through
channel-missing or closed handling, and change the packet path around
enqueue_reliable so the current packet is re-enqueued only for Blocked, not for
Closed.
|
@wrangelvid, as mentioned #155, a reliable and ordered mode, closer to "exactly once" guarantee, unfortunately, is harder than it looks. Most of the complexity comes from the fact that SFU terminates SCTP on the edge, and the relationship is many-to-many, meaning we need a way to also route each packet internally in reliable and ordered mode too. I want to be extra careful in claiming that we support this, when in reality, people can experience implicit data loss. Currently, the system is only single-node. But we want this to be able to scale horizontally as well, so deployment is easier. Even with a single node, communications between cores don't have this reliable and ordered delivery guarantee. It doesn't make sense to make it reliable and ordered, as this will lead to backpressure & potentially a deadlock. Alternatively, a "reliable" and "ordered" mode is best implemented at the application level on the client side, wrapping the current "unreliable" transport layer. The client is the ideal place to enforce this guarantee because it decouples session state from socket life. If a network migration occurs, or if an SFU node fails over, any transport-level protocol state (like native SCTP) is destroyed, resulting in immediate data loss and sequence resets. I also typically see people using MQTT alongside WebRTC in IOT because there's plenty of mature infrastructure for this. TL;DR: It’s not a flat no, but if we support a "reliable and ordered" channel, it needs to actually hold up cluster-wide. Right now, claiming we support it risks hidden data loss during node failovers or inter-core communication, where forcing strict ordering would just trigger backpressure and deadlocks. |
eedc5b1 to
730d253
Compare
2 participants
This should add the remaining delivery classes. Let me know what you think.
Summary by CodeRabbit
New Features
Bug Fixes
Tests