diff --git a/pulsebeam-agent/src/agent/driver.rs b/pulsebeam-agent/src/agent/driver.rs index 8a1cd179..1d40079b 100644 --- a/pulsebeam-agent/src/agent/driver.rs +++ b/pulsebeam-agent/src/agent/driver.rs @@ -109,6 +109,28 @@ pub(crate) enum DataTrackDirection { Subscribe, } +#[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), + } + } +} + #[derive(Debug, Clone)] struct DataTrackBinding { direction: DataTrackDirection, @@ -296,7 +318,15 @@ impl AgentDriver { } pub fn declare_publish_topic(&mut self, topic: &str) -> Result { - let cid = self.ensure_data_topic(DataTrackDirection::Publish, topic)?; + self.declare_publish_topic_with(topic, DeliveryClass::default()) + } + + pub fn declare_publish_topic_with( + &mut self, + topic: &str, + delivery: DeliveryClass, + ) -> Result { + let cid = self.ensure_data_topic(DataTrackDirection::Publish, topic, delivery)?; self.data.data_pub_topics.insert( topic.to_string(), DataPublisher::new(cid, topic.to_string(), self.outgoing_tx.clone()), @@ -305,7 +335,15 @@ impl AgentDriver { } pub fn declare_subscribe_topic(&mut self, topic: &str) -> Result { - let cid = self.ensure_data_topic(DataTrackDirection::Subscribe, topic)?; + self.declare_subscribe_topic_with(topic, DeliveryClass::default()) + } + + pub fn declare_subscribe_topic_with( + &mut self, + topic: &str, + delivery: DeliveryClass, + ) -> Result { + let cid = self.ensure_data_topic(DataTrackDirection::Subscribe, topic, delivery)?; let (tx, rx) = mailbox::bounded(8); self.data.data_sub_topics.insert( topic.to_string(), @@ -800,6 +838,7 @@ impl AgentDriver { &mut self, direction: DataTrackDirection, topic: &str, + delivery: DeliveryClass, ) -> Result { let existing = match direction { DataTrackDirection::Publish => { @@ -814,10 +853,11 @@ impl AgentDriver { } let topic_owned = topic.to_string(); + let (ordered, reliability) = delivery.channel_settings(); let cfg = ChannelConfig { label: data_track_label(direction, &topic_owned), - ordered: false, - reliability: Reliability::MaxRetransmits { retransmits: 0 }, + ordered, + reliability, negotiated: None, protocol: "".to_string(), }; diff --git a/pulsebeam-agent/src/agent/mod.rs b/pulsebeam-agent/src/agent/mod.rs index b842cebf..26a04d05 100644 --- a/pulsebeam-agent/src/agent/mod.rs +++ b/pulsebeam-agent/src/agent/mod.rs @@ -7,7 +7,8 @@ mod slots; pub use builder::AgentBuilder; pub use driver::{ - AgentDriver, AgentError, AgentEvent, AgentStats, ParticipantId, TrackStats, VideoPreset, + AgentDriver, AgentError, AgentEvent, AgentStats, DeliveryClass, ParticipantId, TrackStats, + VideoPreset, }; pub use handles::*; pub use mailbox::*; diff --git a/pulsebeam-simulator/src/tests/data_channel.rs b/pulsebeam-simulator/src/tests/data_channel.rs index d10a1ae2..827245e8 100644 --- a/pulsebeam-simulator/src/tests/data_channel.rs +++ b/pulsebeam-simulator/src/tests/data_channel.rs @@ -1,4 +1,5 @@ use super::common; +use pulsebeam_agent::agent::DeliveryClass; use std::{ sync::{Arc, Mutex}, time::Duration, @@ -90,6 +91,115 @@ fn data_channel_pubsub_forwarding_test() -> turmoil::Result { Ok(()) } +#[test] +fn data_channel_reliable_ordered_forwarding_test() -> turmoil::Result { + common::setup_tracing(); + + let mut sim = turmoil::Builder::new() + .tick_duration(Duration::from_micros(100)) + .rng_seed(0x0DDBA11) + .build(); + + let subnet = common::reserve_subnet(); + let server_ip = common::subnet_ip(subnet, 1); + let pub_ip = common::subnet_ip(subnet, 2); + let sub_ip = common::subnet_ip(subnet, 3); + + let topic = "reliable_topic".to_string(); + const TOTAL_MESSAGES: usize = 32; + + sim.host(server_ip, move || async move { + common::start_sfu_node(server_ip, pulsebeam_runtime::rand::seeded_rng(0xDEADBEEF)) + .await + .map_err(|e| e.into()) + }); + + let received_all = Arc::new(Mutex::new(false)); + { + let topic = topic.clone(); + let received_all = received_all.clone(); + sim.client(pub_ip, async move { + let mut client = common::client::SimClientBuilder::bind(pub_ip, server_ip) + .await? + .connect("room-data-reliable") + .await?; + + client + .ctx + .driver + .declare_publish_topic_with(&topic, DeliveryClass::ReliableOrdered)?; + // Messages published before the subscriber attaches are not replayed. + let mut next_seq = 0usize; + client + .drive_with(|ctx| { + let Some(publisher) = ctx.published_topics.get_mut(&topic) else { + return false; + }; + + let payload = format!("seq-{next_seq}").into_bytes(); + if publisher.try_send(payload).is_ok() { + next_seq += 1; + } + *received_all.lock().unwrap() + }) + .await?; + Ok(()) + }); + } + + { + let topic = topic.clone(); + let received_all = received_all.clone(); + sim.client(sub_ip, async move { + let mut client = common::client::SimClientBuilder::bind(sub_ip, server_ip) + .await? + .connect("room-data-reliable") + .await?; + + client + .ctx + .driver + .declare_subscribe_topic_with(&topic, DeliveryClass::ReliableOrdered)?; + let mut first_seq = None; + let mut count = 0usize; + client + .drive_with(|ctx| { + let Some(subscriber) = ctx.subscribed_topics.get_mut(&topic) else { + return false; + }; + + while let Ok(payload) = subscriber.try_recv() { + let text = String::from_utf8(payload).expect("utf8 payload"); + let seq: usize = text + .strip_prefix("seq-") + .expect("seq- prefix") + .parse() + .expect("sequence number"); + let start = *first_seq.get_or_insert(seq); + assert_eq!( + seq, + start + count, + "reliable-ordered topic delivered out of order or dropped" + ); + count += 1; + } + + if count >= TOTAL_MESSAGES { + *received_all.lock().unwrap() = true; + true + } else { + false + } + }) + .await?; + Ok(()) + }); + } + + sim.run().unwrap(); + Ok(()) +} + #[test] fn data_channel_latency_regression_test() -> turmoil::Result { common::setup_tracing(); diff --git a/pulsebeam/src/participant/core.rs b/pulsebeam/src/participant/core.rs index 8b70ff25..55348bfb 100644 --- a/pulsebeam/src/participant/core.rs +++ b/pulsebeam/src/participant/core.rs @@ -28,11 +28,20 @@ use crate::participant::{ use crate::rtp::RtpPacket; use crate::track::{ self, DataTopicChannel, DataTrackDirection, DataTrackIntent, DataTrackIntentError, - KEYFRAME_DEBOUNCE, StreamId, StreamWriter, Topic, Track, + DeliveryClass, KEYFRAME_DEBOUNCE, StreamId, StreamWriter, Topic, Track, }; const SLOW_POLL_INTERVAL: Duration = Duration::from_millis(100); +const MAX_RELIABLE_BACKLOG_BYTES: usize = 8 * 1024 * 1024; +const RELIABLE_FLUSH_THRESHOLD: usize = 64 * 1024; + +#[derive(Default)] +struct ReliableBacklog { + queue: VecDeque>, + bytes: usize, +} + struct TrackAvailability { in_topology: bool, } @@ -110,6 +119,7 @@ pub struct ParticipantCore { data_topic_channels: HashMap, data_pub_channels: HashMap, data_sub_channels: HashMap, + data_reliable_backlogs: HashMap, // Cold: touched rarely disconnect_reason: Option, @@ -150,6 +160,7 @@ impl ParticipantCore { data_topic_channels: HashMap::new(), data_pub_channels: HashMap::new(), data_sub_channels: HashMap::new(), + data_reliable_backlogs: HashMap::new(), room_id: cfg.room_id, shard_id, }; @@ -192,14 +203,112 @@ impl ParticipantCore { return; }; + let reliable = self + .data_topic_channels + .get(&cid) + .is_some_and(|ch| ch.delivery == DeliveryClass::Reliable); + + if !reliable { + let Some(mut ch) = self.rtc.channel(cid) else { + return; + }; + if let Err(err) = ch.write(true, pkt) { + tracing::warn!(?topic, ?cid, ?err, "failed to forward data topic packet"); + } + return; + } + + if !self.flush_reliable_backlog(cid) { + self.enqueue_reliable(cid, pkt.to_vec()); + return; + } + let Some(mut ch) = self.rtc.channel(cid) else { return; }; - if let Err(err) = ch.write(true, pkt) { - tracing::warn!(?topic, ?cid, ?err, "failed to forward data topic packet"); + match ch.write(true, pkt) { + Ok(true) => {} + Ok(false) => self.enqueue_reliable(cid, pkt.to_vec()), + Err(err) => { + tracing::warn!( + ?topic, + ?cid, + ?err, + "failed to forward reliable data topic packet, closing channel" + ); + self.close_reliable_channel(cid); + } } } + fn flush_reliable_backlog(&mut self, cid: ChannelId) -> bool { + let Some(backlog) = self.data_reliable_backlogs.get_mut(&cid) else { + return true; + }; + + let Some(mut ch) = self.rtc.channel(cid) else { + self.data_reliable_backlogs.remove(&cid); + return false; + }; + + let mut write_error = None; + while let Some(pkt) = backlog.queue.front() { + match ch.write(true, pkt) { + Ok(true) => { + backlog.bytes -= pkt.len(); + backlog.queue.pop_front(); + } + Ok(false) => return false, + Err(err) => { + write_error = Some(err); + break; + } + } + } + + if let Some(err) = write_error { + tracing::warn!( + ?cid, + ?err, + "failed to flush reliable data topic backlog, closing channel" + ); + self.close_reliable_channel(cid); + return false; + } + + self.data_reliable_backlogs.remove(&cid); + true + } + + fn flush_reliable_backlogs(&mut self) { + if self.data_reliable_backlogs.is_empty() { + return; + } + let cids: Vec = self.data_reliable_backlogs.keys().copied().collect(); + for cid in cids { + self.flush_reliable_backlog(cid); + } + } + + fn enqueue_reliable(&mut self, cid: ChannelId, pkt: Vec) { + let backlog = self.data_reliable_backlogs.entry(cid).or_default(); + backlog.bytes += pkt.len(); + backlog.queue.push_back(pkt); + if backlog.bytes > MAX_RELIABLE_BACKLOG_BYTES { + tracing::warn!( + ?cid, + bytes = backlog.bytes, + "reliable data topic backlog exceeded budget, closing channel" + ); + self.close_reliable_channel(cid); + } + } + + fn close_reliable_channel(&mut self, cid: ChannelId) { + self.data_reliable_backlogs.remove(&cid); + self.rtc.direct_api().close_data_channel(cid); + } + #[tracing::instrument(skip_all, fields(participant_id = %self.participant_id))] pub fn on_tracks_published(&mut self, tracks: &[Track]) { for track in tracks { @@ -277,6 +386,8 @@ impl ParticipantCore { } fn poll_slow(&mut self, now: Instant, events: &mut impl ParticipantSink) { + // Buffered-amount-low only fires per stream, backlogs may wait behind other streams. + self.flush_reliable_backlogs(); let assignments_changed = self.downstream.poll_slow(now, &mut self.rtc.bwe(), events); if assignments_changed { self.signaling.mark_assignments_dirty(); @@ -496,13 +607,23 @@ impl ParticipantCore { } DataTrackDirection::Subscribe => { self.data_sub_channels.insert(e.topic.clone(), cid); + if e.delivery == DeliveryClass::Reliable + && let Some(mut ch) = self.rtc.channel(cid) + { + ch.set_buffered_amount_low_threshold(RELIABLE_FLUSH_THRESHOLD); + } events.subscribe_data_topic(e.topic); } } } } } + Event::ChannelBufferedAmountLow(_) => { + // The freed budget is association-wide, retry every backlog. + self.flush_reliable_backlogs(); + } Event::ChannelClose(cid) => { + self.data_reliable_backlogs.remove(&cid); let Some(ch) = self.data_topic_channels.remove(&cid) else { return; }; @@ -753,6 +874,7 @@ impl ParticipantCore { self.data_pub_channels.clear(); self.data_sub_channels.clear(); + self.data_reliable_backlogs.clear(); } fn release_data_topic_channel( diff --git a/pulsebeam/src/track.rs b/pulsebeam/src/track.rs index 3e399488..4f259f77 100644 --- a/pulsebeam/src/track.rs +++ b/pulsebeam/src/track.rs @@ -454,10 +454,30 @@ mod data_track { } } + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] + pub enum DeliveryClass { + Lossy, + SemiReliable, + Reliable, + } + + impl From<&ChannelConfig> for DeliveryClass { + fn from(cfg: &ChannelConfig) -> Self { + match cfg.reliability { + Reliability::Reliable => Self::Reliable, + Reliability::MaxRetransmits { retransmits: 0 } => Self::Lossy, + Reliability::MaxRetransmits { .. } | Reliability::MaxPacketLifetime { .. } => { + Self::SemiReliable + } + } + } + } + #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct DataTopicChannel { pub direction: DataTrackDirection, pub topic: crate::track::Topic, + pub delivery: DeliveryClass, } impl Display for DataTopicChannel { @@ -489,15 +509,6 @@ mod data_track { #[error("The target user asset label component is missing or empty")] MissingLabel, - #[error( - "Unsupported data channel configuration for label '{label}': expected unordered with MaxRetransmits(0), but got ordered={ordered}, reliability={reliability:?}" - )] - UnsupportedDataChannelConfig { - label: String, - ordered: bool, - reliability: Reliability, - }, - #[error( "The label contains illegal characters (only alphanumeric, dashes, and underscores allowed)" )] @@ -529,18 +540,6 @@ mod data_track { } } Some("rt") => { - let supported_delivery_guarantee = matches!( - cfg.reliability, - Reliability::MaxRetransmits { retransmits: 0 } - ) && !cfg.ordered; - if !supported_delivery_guarantee { - return Err(DataTrackIntentError::UnsupportedDataChannelConfig { - label: s.clone(), - ordered: cfg.ordered, - reliability: cfg.reliability, - }); - } - let direction = match parts.next() { Some("pub") => DataTrackDirection::Publish, Some("sub") => DataTrackDirection::Subscribe, @@ -563,6 +562,7 @@ mod data_track { let topic = DataTopicChannel { direction, topic: Topic(topic_slice.to_string()), + delivery: DeliveryClass::from(cfg), }; Ok(Self::UserTopic(topic)) } @@ -664,6 +664,44 @@ mod data_track { assert_eq!(err, DataTrackIntentError::IllegalCharacters); } + #[test] + fn test_delivery_class_from_config() { + let mut reliable = cfg("v1/rt/pub/operation"); + reliable.ordered = true; + reliable.reliability = Reliability::Reliable; + let res = DataTrackIntent::try_from(&reliable).unwrap(); + if let DataTrackIntent::UserTopic(e) = res { + assert_eq!(e.delivery, DeliveryClass::Reliable); + } else { + panic!("Expected UserTopic variant"); + } + + let mut semi = cfg("v1/rt/sub/telemetry"); + semi.reliability = Reliability::MaxRetransmits { retransmits: 5 }; + let res = DataTrackIntent::try_from(&semi).unwrap(); + if let DataTrackIntent::UserTopic(e) = res { + assert_eq!(e.delivery, DeliveryClass::SemiReliable); + } else { + panic!("Expected UserTopic variant"); + } + + let mut lifetime = cfg("v1/rt/sub/telemetry"); + lifetime.reliability = Reliability::MaxPacketLifetime { lifetime: 200 }; + let res = DataTrackIntent::try_from(&lifetime).unwrap(); + if let DataTrackIntent::UserTopic(e) = res { + assert_eq!(e.delivery, DeliveryClass::SemiReliable); + } else { + panic!("Expected UserTopic variant"); + } + + let res = DataTrackIntent::try_from(&cfg("v1/rt/pub/control_input")).unwrap(); + if let DataTrackIntent::UserTopic(e) = res { + assert_eq!(e.delivery, DeliveryClass::Lossy); + } else { + panic!("Expected UserTopic variant"); + } + } + #[test] fn test_max_length_boundary() { // Exact boundary (10 bytes prefix + 54 bytes topic = 64 total)