fix: audio loopback

Author: lherman-cs

pulsebeam/src/audio_selector.rs

@@ -1,4 +1,3 @@
-
 //! Room-level Top-N audio selector.
 //!
 //! A single async task per room that:
@@ -50,12 +49,14 @@ use std::{
 use futures_concurrency::stream::StreamGroup;
 use futures_concurrency::stream::stream_group::Key;
 use futures_lite::stream::Stream as _;
-use pulsebeam_runtime::sync::spmc;
+use pulsebeam_runtime::sync::{Arc, spmc};
 use tokio::{sync::mpsc, time::Instant};
 
 use crate::{
-    controller::MAX_SEND_AUDIO_SLOTS, entity::TrackId, rtp::RtpPacket, rtp::monitor::StreamState,
-    track::TrackReceiver,
+    controller::MAX_SEND_AUDIO_SLOTS,
+    entity::TrackId,
+    rtp::{AudioRtpPacket, RtpPacket, monitor::StreamState},
+    track::{TrackMeta, TrackReceiver},
 };
 
 /// Number of output slots produced by the selector; matches the controller's
@@ -84,15 +85,15 @@ pub struct AudioSelectorSubscription {
     /// slots.  Slot 0 is the loudest speaker, slot 1 the second-loudest, etc.
     /// (assignments shift after each re-rank, but the slot ordering is stable
     /// within a re-rank window).
-    pub receivers: Vec<spmc::Receiver<RtpPacket>>,
+    pub receivers: Vec<spmc::Receiver<AudioRtpPacket>>,
 }
 
 /// Room-side handle: send track commands and create per-participant subscriptions.
 pub struct AudioSelectorHandle {
     /// Send [`AudioSelectorCmd`]s to the background task.
     pub cmd_tx: mpsc::Sender<AudioSelectorCmd>,
     /// One prototype receiver per slot; cloned for every subscriber.
-    receivers: Vec<spmc::Receiver<RtpPacket>>,
+    receivers: Vec<spmc::Receiver<AudioRtpPacket>>,
 }
 
 impl AudioSelectorHandle {
@@ -129,7 +130,7 @@ pub fn create(
     for _ in 0..SELECTOR_SLOTS {
         // 256-packet ring per slot.  At 50 pkt/s that is 5 s of runway;
         // enough to absorb any downstream stall without dropping audio.
-        let (tx, rx) = spmc::channel::<RtpPacket>(256);
+        let (tx, rx) = spmc::channel::<AudioRtpPacket>(256);
         senders.push(tx);
         receivers.push(rx);
     }
@@ -162,6 +163,7 @@ struct InputTrackMeta {
     /// Shared speech-intensity state maintained by the upstream `StreamMonitor`.
     /// Atomic loads — zero overhead on the hot path.
     state: StreamState,
+    meta: Arc<TrackMeta>,
 }
 
 /// One `spmc` receiver wrapped as a `Stream<Item=(TrackId, RtpPacket)>` so it
@@ -213,7 +215,7 @@ impl futures_lite::stream::Stream for InputStream {
 /// One of the N output slots produced by the selector.
 struct OutputSlot {
     /// Broadcast channel; all participant subscriptions share the same ring.
-    sender: spmc::Sender<RtpPacket>,
+    sender: spmc::Sender<AudioRtpPacket>,
     /// Which input track is currently assigned to this slot.
     track_id: Option<TrackId>,
 }
@@ -310,12 +312,13 @@ impl TopNAudioSelector {
                 }
                 let sim = track.lowest_quality();
                 let state = sim.state.clone();
+                let meta = track.meta.clone();
                 let receiver = sim.channel.clone();
                 let key = self.inputs.insert(InputStream {
                     track_id: id,
                     receiver,
                 });
-                self.tracks.insert(id, InputTrackMeta { key, state });
+                self.tracks.insert(id, InputTrackMeta { key, state, meta });
             }
             AudioSelectorCmd::RemoveTrack(id) => {
                 self.remove_track(id);
@@ -342,9 +345,19 @@ impl TopNAudioSelector {
     /// only occurs in the ≤200 ms window between a speaker becoming active and
     /// the next re-rank pass assigning them a slot.
     fn forward(&mut self, track_id: TrackId, packet: RtpPacket) {
-        if let Some(slot) = self.slots.iter_mut().find(|s| s.track_id == Some(track_id)) {
-            slot.sender.send(packet);
-        }
+        let Some(slot) = self.slots.iter_mut().find(|s| s.track_id == Some(track_id)) else {
+            return;
+        };
+
+        let Some(track) = self.tracks.get(&track_id) else {
+            return;
+        };
+
+        slot.sender.send(AudioRtpPacket {
+            participant_id: track.meta.origin_participant,
+            track_id,
+            packet,
+        });
     }
 
     // ── Re-ranking ────────────────────────────────────────────────────────────

pulsebeam/src/participant/core.rs

@@ -111,7 +111,7 @@ impl ParticipantCore {
             udp_batcher,
             tcp_batcher,
             upstream: UpstreamAllocator::new(),
-            downstream: DownstreamAllocator::new(manual_sub),
+            downstream: DownstreamAllocator::new(participant_id, manual_sub),
             slot_meta: HashMap::new(),
             disconnect_reason: None,
             events: Vec::with_capacity(32),

pulsebeam/src/participant/downstream/audio.rs

@@ -5,9 +5,10 @@ use std::task::{Context, Poll};
 
 use crate::audio_selector::AudioSelectorSubscription;
 use crate::controller::MAX_SEND_AUDIO_SLOTS;
+use crate::entity::ParticipantId;
 use crate::rtp;
-use crate::rtp::RtpPacket;
 use crate::rtp::timeline::Timeline;
+use crate::rtp::{AudioRtpPacket, RtpPacket};
 use str0m::media::Mid;
 use str0m::rtp::Ssrc;
 
@@ -18,6 +19,7 @@ use str0m::rtp::Ssrc;
 /// every audio `Mid` negotiated with the client.  The allocator pairs each
 /// slot index with the matching subscription receiver index.
 pub struct AudioAllocator {
+    participant_id: ParticipantId,
     /// One stream per negotiated audio Mid.  Each stream wraps one selector
     /// output receiver + one `AudioSlot` (timeline rewriter).
     ///
@@ -37,8 +39,9 @@ pub struct AudioAllocator {
 }
 
 impl AudioAllocator {
-    pub fn new() -> Self {
+    pub fn new(participant_id: ParticipantId) -> Self {
         Self {
+            participant_id,
             slots: SlotGroup::with_capacity(MAX_SEND_AUDIO_SLOTS),
             pending_sub: None,
         }
@@ -67,7 +70,11 @@ impl AudioAllocator {
     /// regardless of the Top-N ranking.  Returns `false` if `slot_index` has
     /// not yet been registered via [`add_slot`].
     #[allow(unused)]
-    pub fn pin_slot(&mut self, slot_index: usize, receiver: spmc::Receiver<RtpPacket>) -> bool {
+    pub fn pin_slot(
+        &mut self,
+        slot_index: usize,
+        receiver: spmc::Receiver<AudioRtpPacket>,
+    ) -> bool {
         let Some(mut stream) = self.slots.get_mut(slot_index) else {
             return false;
         };
@@ -113,9 +120,20 @@ impl AudioAllocator {
     #[inline]
     pub(super) fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<(Mid, RtpPacket)>> {
         use futures_lite::stream::Stream as _;
-        match Pin::new(&mut self.slots).poll_next(cx) {
-            Poll::Ready(Some(item)) => Poll::Ready(Some(item)),
-            Poll::Ready(None) | Poll::Pending => Poll::Pending,
+
+        // Drop packets that originate from this participant to avoid sending
+        // loopback audio back to the sender.
+        loop {
+            match Pin::new(&mut self.slots).poll_next(cx) {
+                Poll::Ready(Some((mid, packet))) => {
+                    if packet.participant_id == self.participant_id {
+                        // Keep draining until we find a packet from another participant.
+                        continue;
+                    }
+                    return Poll::Ready(Some((mid, packet.packet)));
+                }
+                Poll::Ready(None) | Poll::Pending => return Poll::Pending,
+            }
         }
     }
 }
@@ -132,7 +150,7 @@ impl AudioAllocator {
 /// the session lifetime; only the underlying receiver is swapped on
 /// subscription/pin changes.
 struct AudioInputStream {
-    receiver: Option<spmc::Receiver<RtpPacket>>,
+    receiver: Option<spmc::Receiver<AudioRtpPacket>>,
     slot: AudioSlot,
 }
 
@@ -147,14 +165,14 @@ impl AudioInputStream {
     /// Swap in a new receiver and reset the timeline so the SSRC-change
     /// detection in `AudioSlot::process` triggers a clean rebase on the
     /// first packet from the new source.
-    fn set_receiver(&mut self, receiver: spmc::Receiver<RtpPacket>) {
+    fn set_receiver(&mut self, receiver: spmc::Receiver<AudioRtpPacket>) {
         self.receiver = Some(receiver);
         self.slot.last_ssrc = None;
     }
 }
 
 impl futures_lite::stream::Stream for AudioInputStream {
-    type Item = (Mid, RtpPacket);
+    type Item = (Mid, AudioRtpPacket);
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
@@ -203,11 +221,12 @@ impl AudioSlot {
         }
     }
 
-    fn process(&mut self, packet: RtpPacket) -> (Mid, RtpPacket) {
-        if self.last_ssrc != Some(packet.ssrc) {
-            self.last_ssrc = Some(packet.ssrc);
-            self.timeline.rebase(&packet);
+    fn process(&mut self, mut p: AudioRtpPacket) -> (Mid, AudioRtpPacket) {
+        if self.last_ssrc != Some(p.packet.ssrc) {
+            self.last_ssrc = Some(p.packet.ssrc);
+            self.timeline.rebase(&p.packet);
         }
-        (self.mid, self.timeline.rewrite(packet))
+        p.packet = self.timeline.rewrite(p.packet);
+        (self.mid, p)
     }
 }

pulsebeam/src/participant/downstream/mod.rs

@@ -2,6 +2,7 @@ mod audio;
 mod video;
 
 use crate::audio_selector::AudioSelectorSubscription;
+use crate::entity::ParticipantId;
 use crate::participant::downstream::audio::AudioAllocator;
 use crate::participant::downstream::video::{SlotConfig, VideoAllocator};
 use crate::rtp::RtpPacket;
@@ -26,10 +27,10 @@ pub struct DownstreamAllocator {
 }
 
 impl DownstreamAllocator {
-    pub fn new(manual_sub: bool) -> Self {
+    pub fn new(participant_id: ParticipantId, manual_sub: bool) -> Self {
         Self {
             available_bandwidth: MIN_BANDWIDTH,
-            audio: AudioAllocator::new(),
+            audio: AudioAllocator::new(participant_id),
             video: VideoAllocator::new(manual_sub),
             dirty_allocation: false,
         }

pulsebeam/src/rtp/mod.rs

@@ -12,6 +12,8 @@ use str0m::rtp::rtcp::SenderInfo;
 use str0m::rtp::{ExtensionValues, SeqNo, Ssrc};
 use tokio::time::Instant;
 
+use crate::entity::{ParticipantId, TrackId};
+
 /// Pool capacity: 16 384 slots × 2 048 bytes ≈ 32 MB resident.
 ///
 /// Trade memory for zero jemalloc on the hot path. RTP payload pool and
@@ -41,16 +43,20 @@ pub enum Codec {
     Opus,
 }
 
+#[derive(Clone, Debug)]
+pub struct AudioRtpPacket {
+    pub participant_id: ParticipantId,
+    pub track_id: TrackId,
+    pub packet: RtpPacket,
+}
+
 /// Unified internal RTP packet representation used across the SFU.
 /// This struct is designed for mutability and composition in middleware.
 /// Only the fields actually consumed by the forwarding pipeline are kept here;
 /// redundant header data (sequence_number, timestamp, csrc list, etc.) is dropped
 /// at ingress so every ring-slot stays as small as possible.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RtpPacket {
-    // NOTE: `payload` is a `PoolBuf` — a custom Arc-like handle backed by
-    // `rtp_payload_pool()`.  `.clone()` is a single atomic increment with no
-    // heap allocation, making SPMC fan-out essentially free.
     pub ssrc: Ssrc,
     pub marker: bool,
     pub ext_vals: ExtensionValues,