Configure delivery classes for data channel

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<ChannelId, AgentError> {
-        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<ChannelId, AgentError> {
+        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<ChannelId, AgentError> {
-        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<ChannelId, AgentError> {
+        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<ChannelId, AgentError> {
         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(),
         };

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::*;

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();