feat: synchronized playback

Author: lherman-cs

pulsebeam-runtime/src/collections/ring.rs

@@ -82,6 +82,30 @@ impl<T: Clone> RingBuffer<T> {
         slid
     }
 
+    /// Removes and returns an item at a specific sequence number, if it exists.
+    /// This does not affect the head or tail pointers. It is useful for cherry-picking
+    /// items from the buffer without advancing the read cursor (`tail`).
+    pub fn remove(&mut self, seq: u64) -> Option<T> {
+        if !self.initialized {
+            return None;
+        }
+
+        // A sequence number `s` is within the valid range `[tail, head)` if and only if
+        // the distance from the tail to `s` is less than the distance from the tail to the head.
+        // This check correctly handles the u64 wrap-around.
+        let distance_from_tail = seq.wrapping_sub(self.tail);
+        let buffer_length = self.head.wrapping_sub(self.tail);
+
+        if distance_from_tail >= buffer_length {
+            // The sequence number is outside the valid range (either too old or too new).
+            return None;
+        }
+
+        // The sequence number is within our valid range, so we can calculate the index.
+        let index = (seq % self.capacity()) as usize;
+        self.buffer[index].take()
+    }
+
     /// Takes the item at the current tail, advancing the tail if successful.
     pub fn pop_front(&mut self) -> Option<T> {
         if self.tail == self.head {
@@ -358,4 +382,50 @@ mod tests {
 
         assert_eq!(buffer.pop_front(), None);
     }
+
+    #[test]
+    fn test_remove() {
+        let mut buffer = RingBuffer::new(10);
+        buffer.insert(10, 100); // tail=10, head=11
+        buffer.insert(12, 120); // tail=10, head=13
+
+        // Remove an existing item
+        assert_eq!(buffer.remove(10), Some(100));
+        // The buffer should now contain a None at that position
+        assert!(buffer.buffer[(10 % 10) as usize].is_none());
+        // Pointers should be unchanged
+        assert_eq!(buffer.tail(), 10);
+        assert_eq!(buffer.head(), 13);
+
+        // Try to remove it again
+        assert_eq!(buffer.remove(10), None);
+
+        // Remove another existing item
+        assert_eq!(buffer.remove(12), Some(120));
+        assert!(buffer.buffer[(12 % 10) as usize].is_none());
+
+        // Try to remove an item that was never there (a gap)
+        assert_eq!(buffer.remove(11), None);
+
+        // Try to remove an item older than tail
+        assert_eq!(buffer.remove(9), None);
+
+        // Try to remove an item that is at or newer than head
+        assert_eq!(buffer.remove(13), None);
+        assert_eq!(buffer.remove(14), None);
+
+        // Test remove on wrap around
+        let mut buffer = RingBuffer::new(10);
+        let start_seq = u64::MAX - 2;
+        buffer.insert(start_seq, 1);
+        buffer.insert(start_seq.wrapping_add(1), 2);
+        buffer.insert(start_seq.wrapping_add(3), 4); // seq=0
+
+        assert_eq!(buffer.tail(), start_seq);
+        assert_eq!(buffer.head(), start_seq.wrapping_add(4)); // seq=1
+
+        assert_eq!(buffer.remove(start_seq.wrapping_add(3)), Some(4));
+        assert_eq!(buffer.remove(start_seq.wrapping_add(2)), None);
+        assert_eq!(buffer.remove(start_seq), Some(1));
+    }
 }

pulsebeam/src/participant/downstream.rs

@@ -446,8 +446,8 @@ impl DownstreamAllocator {
     ) -> TrackStream {
         async_stream::stream! {
             let mut reader = TrackReader::new(track, control_rx);
-            while let Some(item) = reader.next_packet().await {
-                yield item;
+            while let Some(pkt) = reader.next_packet().await {
+                yield (reader.id.clone(), pkt);
             }
         }
         .boxed()
@@ -467,8 +467,13 @@ impl Stream for DownstreamAllocator {
 enum TrackReaderState {
     /// The stream is paused and waiting for a new configuration.
     Paused,
+    Resuming {
+        active_index: usize,
+    },
     /// Actively streaming from a single, stable layer.
-    Streaming { active_index: usize },
+    Streaming {
+        active_index: usize,
+    },
     /// Attempting to switch to a new layer (`active_index`) while forwarding packets
     /// from the old layer (`fallback_index`) until the new one becomes active.
     Transitioning {
@@ -483,9 +488,9 @@ pub struct TrackReader {
     track: TrackReceiver,
     control_rx: watch::Receiver<StreamConfig>,
 
+    switcher: Switcher,
     state: TrackReaderState,
     config: StreamConfig,
-    switcher: Switcher,
 }
 
 impl TrackReader {
@@ -495,26 +500,25 @@ impl TrackReader {
             MediaKind::Audio => rtp::AUDIO_FREQUENCY,
             MediaKind::Video => rtp::VIDEO_FREQUENCY,
         };
-        let switcher = Switcher::new(clock_rate);
         let mut this = Self {
             id: track.meta.id.clone(),
             track,
             control_rx,
+            switcher: Switcher::new(clock_rate),
             state: TrackReaderState::Paused,
             config,
-            switcher,
         };
         // seed initial state
         this.update_state(config);
         this
     }
 
     /// Poll for the next RTP packet, handling fallback and layer switching.
-    pub async fn next_packet(&mut self) -> Option<TrackStreamItem> {
+    pub async fn next_packet(&mut self) -> Option<RtpPacket> {
         loop {
             self.maybe_update_state();
             if let Some(pkt) = self.switcher.pop() {
-                return Some((self.id.clone(), pkt));
+                return Some(pkt);
             }
 
             match self.state {
@@ -524,6 +528,37 @@ impl TrackReader {
                         return None;
                     }
                 }
+                TrackReaderState::Resuming { active_index } => {
+                    let receiver = &mut self.track.simulcast[active_index];
+                    tokio::select! {
+                        biased;
+
+                        res = self.control_rx.changed() => {
+                            if res.is_err() { return None; }
+                            continue; // Re-evaluate state on next loop iteration
+                        },
+
+                        res = receiver.channel.recv() => {
+                            match res {
+                                Ok(pkt) => {
+                                    self.switcher.stage(pkt.value.clone());
+                                },
+                                Err(spmc::RecvError::Lagged(n)) => {
+                                    tracing::warn!(track_id = %self.id, "Receiver lagged {n}, requesting keyframe");
+                                    receiver.request_keyframe(str0m::media::KeyframeRequestKind::Pli);
+                                }
+                                Err(spmc::RecvError::Closed) => {
+                                    tracing::warn!(track_id = %self.id, "Channel closed, ending stream");
+                                    return None;
+                                }
+                            }
+                        }
+                    }
+
+                    if self.switcher.is_ready() {
+                        self.state = TrackReaderState::Streaming { active_index };
+                    }
+                }
                 TrackReaderState::Streaming { active_index } => {
                     let receiver = &mut self.track.simulcast[active_index];
                     tokio::select! {
@@ -573,7 +608,7 @@ impl TrackReader {
                         res = active_receiver.channel.recv() => {
                             match res {
                                 Ok(pkt) => {
-                                    self.switcher.push(pkt.value.clone());
+                                    self.switcher.stage(pkt.value.clone());
                                 }
                                 Err(spmc::RecvError::Lagged(n)) => {
                                     tracing::warn!(track_id = %self.id, "New active stream lagged {n}, completing switch anyway");
@@ -606,7 +641,7 @@ impl TrackReader {
                         }
                     }
 
-                    if self.switcher.is_stable() {
+                    if self.switcher.is_ready() {
                         self.state = TrackReaderState::Streaming { active_index };
                     }
                 }
@@ -640,8 +675,16 @@ impl TrackReader {
         // Determine the next state based on the current state and the new config.
         let next_state = match self.state {
             TrackReaderState::Paused => new_active_index
-                .map(|idx| TrackReaderState::Streaming { active_index: idx })
+                .map(|idx| TrackReaderState::Resuming { active_index: idx })
                 .unwrap_or(TrackReaderState::Paused),
+            TrackReaderState::Resuming { active_index } => match new_active_index {
+                // TODO: double check if this is safe to resume to a different index here
+                Some(new_idx) if new_idx != active_index => {
+                    TrackReaderState::Resuming { active_index }
+                }
+                Some(_) => self.state,
+                None => TrackReaderState::Paused,
+            },
             TrackReaderState::Streaming { active_index } => {
                 match new_active_index {
                     Some(new_idx) if new_idx != active_index => TrackReaderState::Transitioning {
@@ -670,30 +713,27 @@ impl TrackReader {
         if self.state != next_state {
             tracing::debug!(track_id = %self.id, "TrackReader state changed from {:?} to {:?}", self.state, next_state);
 
-            match (self.state, next_state) {
-                (
-                    _,
-                    TrackReaderState::Transitioning {
-                        active_index,
-                        fallback_index,
-                    },
-                ) => {
-                    let fallback_receiver_rid = { self.track.simulcast[fallback_index].rid };
+            match next_state {
+                TrackReaderState::Resuming { active_index } => {
                     let active_receiver = &mut self.track.simulcast[active_index];
                     active_receiver.channel.reset();
                     active_receiver.request_keyframe(KeyframeRequestKind::Fir);
                     tracing::info!(track_id = %self.id,
-                        "switch simulcast layer: {:?} -> {:?}",
-                        fallback_receiver_rid, active_receiver.rid
+                        "resuming simulcast layer: {:?}",
+                        active_receiver.rid
                     );
                 }
-                (TrackReaderState::Paused, TrackReaderState::Streaming { active_index }) => {
+                TrackReaderState::Transitioning {
+                    active_index,
+                    fallback_index,
+                } => {
+                    let fallback_receiver_rid = { self.track.simulcast[fallback_index].rid };
                     let active_receiver = &mut self.track.simulcast[active_index];
                     active_receiver.channel.reset();
                     active_receiver.request_keyframe(KeyframeRequestKind::Fir);
                     tracing::info!(track_id = %self.id,
-                        "resuming simulcast layer: {:?}",
-                        active_receiver.rid
+                        "switch simulcast layer: {:?} -> {:?}",
+                        fallback_receiver_rid, active_receiver.rid
                     );
                 }
                 _ => {}