o_sfu_core/engine/room/
outbound.rs

1//! Bounded per-user room output.
2//!
3//! [`UserOutboundSender::send`] enqueues without waiting and signals
4//! [`UserOutboundEvent::Overflow`] when message-count or byte capacity is
5//! exhausted. [`UserOutboundReceiver::recv_event`] prioritizes that signal over
6//! queued output. User-session loops must stop normal draining after overflow.
7
8use std::{
9    collections::BTreeMap,
10    future::pending,
11    io,
12    sync::{
13        Arc, Mutex,
14        atomic::{AtomicUsize, Ordering},
15    },
16};
17
18use tokio::sync::{
19    mpsc::{self, error::TrySendError},
20    watch,
21};
22
23use super::UserCloseReason;
24use crate::engine::{
25    JsonPayload, RecordingStateUpdate, UserId, UserInfo, metrics::RuntimeMetrics,
26    source_model::UserStreamId, sync::lock_unpoisoned,
27};
28
29pub const MAX_BROADCAST_PAYLOAD_BYTES: usize = 16 * 1024;
30
31const ROOM_EVENT_QUEUE_BYTES: usize = 1024;
32const BROADCAST_QUEUE_OVERHEAD_BYTES: usize = 256;
33const TRACK_PROJECTION_QUEUE_OVERHEAD_BYTES: usize = 256;
34
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct BroadcastPayload {
37    message: Arc<JsonPayload>,
38    byte_len: usize,
39}
40
41impl BroadcastPayload {
42    /// Creates a broadcast payload and records its serialized byte length.
43    ///
44    /// # Errors
45    ///
46    /// Returns [`BroadcastPayloadError::TooLarge`] when the serialized JSON
47    /// exceeds [`MAX_BROADCAST_PAYLOAD_BYTES`]. Returns
48    /// [`BroadcastPayloadError::JsonSerialization`] when JSON serialization
49    /// fails.
50    pub fn try_new(message: JsonPayload) -> Result<Self, BroadcastPayloadError> {
51        let byte_len = serialized_json_len(&message)?;
52        if byte_len > MAX_BROADCAST_PAYLOAD_BYTES {
53            return Err(BroadcastPayloadError::TooLarge {
54                actual: byte_len,
55                limit: MAX_BROADCAST_PAYLOAD_BYTES,
56            });
57        }
58        Ok(Self {
59            message: Arc::new(message),
60            byte_len,
61        })
62    }
63
64    #[must_use]
65    pub const fn byte_len(&self) -> usize {
66        self.byte_len
67    }
68
69    #[must_use]
70    pub fn to_json(&self) -> JsonPayload {
71        self.message.as_ref().clone()
72    }
73}
74
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
76pub enum BroadcastPayloadError {
77    TooLarge { actual: usize, limit: usize },
78    JsonSerialization,
79}
80
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub enum RoomEventMessage {
83    Broadcast {
84        sender_id: UserId,
85        message: BroadcastPayload,
86    },
87    UserJoined {
88        user_id: UserId,
89        info: UserInfo,
90    },
91    UserDeparted {
92        user_id: UserId,
93    },
94    UserInfoChanged(BTreeMap<UserId, UserInfo>),
95    RecordingStateChanged(RecordingStateUpdate),
96}
97
98impl RoomEventMessage {
99    #[must_use]
100    pub(super) fn queued_bytes(&self) -> usize {
101        match self {
102            Self::Broadcast { message, .. } => message
103                .byte_len()
104                .saturating_add(BROADCAST_QUEUE_OVERHEAD_BYTES),
105            Self::UserInfoChanged(snapshot) => {
106                ROOM_EVENT_QUEUE_BYTES.saturating_mul(snapshot.len())
107            }
108            Self::UserJoined { .. }
109            | Self::UserDeparted { .. }
110            | Self::RecordingStateChanged(_) => ROOM_EVENT_QUEUE_BYTES,
111        }
112    }
113}
114
115#[derive(Debug, Default)]
116struct JsonByteCounter {
117    len: usize,
118}
119
120impl io::Write for JsonByteCounter {
121    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
122        self.len = self.len.saturating_add(buf.len());
123        Ok(buf.len())
124    }
125
126    fn flush(&mut self) -> io::Result<()> {
127        Ok(())
128    }
129}
130
131fn serialized_json_len(value: &JsonPayload) -> Result<usize, BroadcastPayloadError> {
132    let mut counter = JsonByteCounter::default();
133    serde_json::to_writer(&mut counter, value)
134        .map_err(|_error| BroadcastPayloadError::JsonSerialization)?;
135    Ok(counter.len)
136}
137
138pub const DEFAULT_USER_OUTBOUND_QUEUE_CAPACITY: usize = 128;
139pub const DEFAULT_USER_OUTBOUND_QUEUE_BYTE_CAPACITY: usize =
140    DEFAULT_USER_OUTBOUND_QUEUE_CAPACITY * MAX_BROADCAST_PAYLOAD_BYTES;
141
142pub(super) type OutboundSender = UserOutboundSender;
143
144#[derive(Debug, Clone, PartialEq, Eq)]
145pub struct RemoteTrackProjection {
146    pub consumer_mid: String,
147    pub user_id: UserId,
148    pub stream_id: UserStreamId,
149    pub producer_active: bool,
150}
151
152#[derive(Debug, Clone, PartialEq, Eq)]
153pub struct RemoteTrackSnapshot {
154    pub tracks: Vec<RemoteTrackProjection>,
155    pub requires_negotiation: bool,
156}
157
158impl RemoteTrackSnapshot {
159    fn queued_bytes(&self) -> usize {
160        self.tracks
161            .iter()
162            .fold(ROOM_EVENT_QUEUE_BYTES, |bytes, track| {
163                let user_id_bytes = match &track.user_id {
164                    UserId::Integer(_) => 0,
165                    UserId::String(value) => value.len(),
166                };
167                bytes
168                    .saturating_add(TRACK_PROJECTION_QUEUE_OVERHEAD_BYTES)
169                    .saturating_add(track.consumer_mid.len())
170                    .saturating_add(user_id_bytes)
171                    .saturating_add(track.stream_id.as_str().len())
172            })
173    }
174}
175
176#[derive(Debug, Clone)]
177pub(in crate::engine::room) struct VersionedRemoteTrackSnapshot {
178    pub(in crate::engine::room) snapshot: RemoteTrackSnapshot,
179    pub(in crate::engine::room) revision: u64,
180}
181
182/// room output that belongs to one connected user
183#[derive(Debug, Clone)]
184pub enum UserOutbound {
185    Message(RoomEventMessage),
186    RemoteTracks(RemoteTrackSnapshot),
187    Close(UserCloseReason),
188}
189
190impl UserOutbound {
191    #[must_use]
192    pub(super) fn queued_bytes(&self) -> usize {
193        match self {
194            Self::Message(message) => message.queued_bytes(),
195            Self::RemoteTracks(snapshot) => snapshot.queued_bytes(),
196            Self::Close(_) => ROOM_EVENT_QUEUE_BYTES,
197        }
198    }
199}
200
201/// queue overflow details captured when a user cannot accept more outbound work
202#[derive(Debug, Clone, Copy, PartialEq, Eq)]
203pub struct UserOutboundOverflow {
204    kind: UserOutboundOverflowKind,
205    message_capacity: usize,
206    byte_capacity: usize,
207    queued_bytes: usize,
208    message_bytes: usize,
209}
210
211/// outbound queue limit that rejected a message
212#[derive(Debug, Clone, Copy, PartialEq, Eq)]
213pub enum UserOutboundOverflowKind {
214    MessageCount,
215    QueuedBytes,
216}
217
218impl UserOutboundOverflow {
219    const fn new(
220        kind: UserOutboundOverflowKind,
221        message_capacity: usize,
222        byte_capacity: usize,
223        queued_bytes: usize,
224        message_bytes: usize,
225    ) -> Self {
226        Self {
227            kind,
228            message_capacity,
229            byte_capacity,
230            queued_bytes,
231            message_bytes,
232        }
233    }
234
235    #[must_use]
236    pub const fn capacity(self) -> usize {
237        self.message_capacity
238    }
239
240    #[must_use]
241    pub const fn kind(self) -> UserOutboundOverflowKind {
242        self.kind
243    }
244
245    #[must_use]
246    pub const fn byte_capacity(self) -> usize {
247        self.byte_capacity
248    }
249
250    #[must_use]
251    pub const fn queued_bytes(self) -> usize {
252        self.queued_bytes
253    }
254
255    #[must_use]
256    pub const fn message_bytes(self) -> usize {
257        self.message_bytes
258    }
259}
260
261/// non-blocking outbound send failure
262#[derive(Debug, Clone, Copy, PartialEq, Eq)]
263pub enum UserOutboundSendError {
264    Full(UserOutboundOverflow),
265    Closed,
266}
267
268/// receiver-side queue event for user-session loops
269#[derive(Debug)]
270pub enum UserOutboundEvent {
271    Message(UserOutbound),
272    Overflow(UserOutboundOverflow),
273    Closed,
274}
275
276/// message and byte capacity for one user outbound queue
277#[derive(Debug, Clone, Copy, PartialEq, Eq)]
278pub struct UserOutboundQueueLimits {
279    message_capacity: usize,
280    byte_capacity: usize,
281}
282
283impl UserOutboundQueueLimits {
284    #[must_use]
285    pub fn new(message_capacity: usize, byte_capacity: usize) -> Self {
286        Self {
287            message_capacity: message_capacity.max(1),
288            byte_capacity: byte_capacity.max(1),
289        }
290    }
291
292    #[must_use]
293    pub const fn message_capacity(self) -> usize {
294        self.message_capacity
295    }
296
297    #[must_use]
298    pub const fn byte_capacity(self) -> usize {
299        self.byte_capacity
300    }
301}
302
303impl Default for UserOutboundQueueLimits {
304    fn default() -> Self {
305        Self::new(
306            DEFAULT_USER_OUTBOUND_QUEUE_CAPACITY,
307            DEFAULT_USER_OUTBOUND_QUEUE_BYTE_CAPACITY,
308        )
309    }
310}
311
312#[derive(Debug)]
313struct QueuedUserOutbound {
314    outbound: UserOutbound,
315    bytes: usize,
316}
317
318#[derive(Debug, Clone)]
319pub struct UserOutboundSender {
320    messages: mpsc::Sender<QueuedUserOutbound>,
321    overflow: watch::Sender<Option<UserOutboundOverflow>>,
322    metrics: Arc<RuntimeMetrics>,
323    limits: UserOutboundQueueLimits,
324    queued_bytes: Arc<AtomicUsize>,
325    latest_track_snapshot: Arc<Mutex<Option<VersionedRemoteTrackSnapshot>>>,
326}
327
328#[derive(Debug)]
329pub struct UserOutboundReceiver {
330    messages: mpsc::Receiver<QueuedUserOutbound>,
331    overflow: watch::Receiver<Option<UserOutboundOverflow>>,
332    metrics: Arc<RuntimeMetrics>,
333    queued_bytes: Arc<AtomicUsize>,
334}
335
336impl UserOutboundSender {
337    #[must_use]
338    pub fn channel(capacity: usize, metrics: Arc<RuntimeMetrics>) -> (Self, UserOutboundReceiver) {
339        Self::channel_with_limits(
340            UserOutboundQueueLimits::new(capacity, DEFAULT_USER_OUTBOUND_QUEUE_BYTE_CAPACITY),
341            metrics,
342        )
343    }
344
345    #[must_use]
346    pub fn channel_with_limits(
347        limits: UserOutboundQueueLimits,
348        metrics: Arc<RuntimeMetrics>,
349    ) -> (Self, UserOutboundReceiver) {
350        let (messages_tx, messages_rx) = mpsc::channel(limits.message_capacity());
351        let (overflow_tx, overflow_rx) = watch::channel(None);
352        let queued_bytes = Arc::new(AtomicUsize::new(0));
353        (
354            Self {
355                messages: messages_tx,
356                overflow: overflow_tx,
357                metrics: Arc::clone(&metrics),
358                limits,
359                queued_bytes: Arc::clone(&queued_bytes),
360                latest_track_snapshot: Arc::new(Mutex::new(None)),
361            },
362            UserOutboundReceiver {
363                messages: messages_rx,
364                overflow: overflow_rx,
365                metrics,
366                queued_bytes,
367            },
368        )
369    }
370
371    /// Enqueues `outbound` without waiting for capacity.
372    ///
373    /// [`UserOutboundSendError::Full`] also signals an overflow event for
374    /// [`UserOutboundReceiver::recv_event`].
375    ///
376    /// # Errors
377    ///
378    /// Returns [`UserOutboundSendError::Full`] when message-count or byte
379    /// capacity is exhausted. Returns [`UserOutboundSendError::Closed`] when the
380    /// receiver has been dropped.
381    pub fn send(&self, outbound: UserOutbound) -> Result<(), UserOutboundSendError> {
382        self.enqueue(outbound)
383    }
384
385    /// Suppresses older track state while carrying every late negotiation edge
386    /// forward on the latest snapshot.
387    pub(in crate::engine::room) fn send_remote_tracks(
388        &self,
389        snapshot: VersionedRemoteTrackSnapshot,
390    ) -> Result<(), UserOutboundSendError> {
391        let revision = snapshot.revision;
392        {
393            let mut latest = lock_unpoisoned(&self.latest_track_snapshot);
394            if self.messages.is_closed() {
395                return Err(UserOutboundSendError::Closed);
396            }
397            if latest
398                .as_ref()
399                .is_none_or(|current| revision > current.revision)
400            {
401                self.enqueue(UserOutbound::RemoteTracks(snapshot.snapshot.clone()))?;
402                *latest = Some(snapshot);
403                return Ok(());
404            }
405            if let Some(current) = latest.as_mut()
406                && revision < current.revision
407                && snapshot.snapshot.requires_negotiation
408            {
409                current.snapshot.requires_negotiation = true;
410                self.enqueue(UserOutbound::RemoteTracks(current.snapshot.clone()))?;
411            }
412        }
413        Ok(())
414    }
415
416    fn enqueue(&self, outbound: UserOutbound) -> Result<(), UserOutboundSendError> {
417        let bytes = outbound.queued_bytes();
418        self.reserve_bytes(bytes)?;
419        match self
420            .messages
421            .try_send(QueuedUserOutbound { outbound, bytes })
422        {
423            Ok(()) => {
424                self.metrics.add_ws_outbound_queued_messages(1);
425                Ok(())
426            }
427            Err(TrySendError::Full(_outbound)) => {
428                self.release_bytes(bytes);
429                let overflow = self.mark_overflow(
430                    UserOutboundOverflowKind::MessageCount,
431                    self.queued_bytes.load(Ordering::Acquire),
432                    bytes,
433                );
434                Err(UserOutboundSendError::Full(overflow))
435            }
436            Err(TrySendError::Closed(_outbound)) => {
437                self.release_bytes(bytes);
438                Err(UserOutboundSendError::Closed)
439            }
440        }
441    }
442
443    fn reserve_bytes(&self, bytes: usize) -> Result<(), UserOutboundSendError> {
444        let byte_capacity = self.limits.byte_capacity();
445        let mut queued = self.queued_bytes.load(Ordering::Acquire);
446        loop {
447            let Some(next) = queued.checked_add(bytes) else {
448                let overflow =
449                    self.mark_overflow(UserOutboundOverflowKind::QueuedBytes, queued, bytes);
450                return Err(UserOutboundSendError::Full(overflow));
451            };
452            if next > byte_capacity {
453                let overflow =
454                    self.mark_overflow(UserOutboundOverflowKind::QueuedBytes, queued, bytes);
455                return Err(UserOutboundSendError::Full(overflow));
456            }
457            match self.queued_bytes.compare_exchange_weak(
458                queued,
459                next,
460                Ordering::AcqRel,
461                Ordering::Acquire,
462            ) {
463                Ok(_previous) => return Ok(()),
464                Err(current) => queued = current,
465            }
466        }
467    }
468
469    fn release_bytes(&self, bytes: usize) {
470        self.queued_bytes.fetch_sub(bytes, Ordering::AcqRel);
471    }
472
473    fn mark_overflow(
474        &self,
475        kind: UserOutboundOverflowKind,
476        queued_bytes: usize,
477        message_bytes: usize,
478    ) -> UserOutboundOverflow {
479        let overflow = UserOutboundOverflow::new(
480            kind,
481            self.limits.message_capacity(),
482            self.limits.byte_capacity(),
483            queued_bytes,
484            message_bytes,
485        );
486        self.metrics.record_ws_outbound_queue_overflow();
487        let _ = self.overflow.send(Some(overflow));
488        overflow
489    }
490}
491
492impl UserOutboundReceiver {
493    #[must_use]
494    pub fn has_overflowed(&self) -> bool {
495        self.overflow.borrow().is_some()
496    }
497
498    /// Receives queued output without observing overflow.
499    ///
500    /// User-session loops should use [`Self::recv_event`].
501    pub async fn recv(&mut self) -> Option<UserOutbound> {
502        self.messages
503            .recv()
504            .await
505            .map(|message| self.record_received(message))
506    }
507
508    /// Attempts to receive queued output without observing overflow.
509    ///
510    /// User-session loops should use [`Self::recv_event`].
511    ///
512    /// # Errors
513    ///
514    /// Returns [`mpsc::error::TryRecvError::Empty`] when no output is queued and
515    /// [`mpsc::error::TryRecvError::Disconnected`] when every sender is dropped.
516    pub fn try_recv(&mut self) -> Result<UserOutbound, mpsc::error::TryRecvError> {
517        self.messages
518            .try_recv()
519            .map(|message| self.record_received(message))
520    }
521
522    /// Receives the next queue event with overflow prioritized over queued output.
523    pub async fn recv_event(&mut self) -> UserOutboundEvent {
524        if let Some(overflow) = *self.overflow.borrow_and_update() {
525            return UserOutboundEvent::Overflow(overflow);
526        }
527        tokio::select! {
528            biased;
529            overflow = wait_for_overflow(&mut self.overflow) => {
530                UserOutboundEvent::Overflow(overflow)
531            }
532            message = self.messages.recv() => {
533                message.map_or(UserOutboundEvent::Closed, |message| {
534                    UserOutboundEvent::Message(self.record_received(message))
535                })
536            }
537        }
538    }
539
540    fn record_received(&self, message: QueuedUserOutbound) -> UserOutbound {
541        self.queued_bytes.fetch_sub(message.bytes, Ordering::AcqRel);
542        self.metrics.add_ws_outbound_queued_messages(-1);
543        message.outbound
544    }
545}
546
547impl Drop for UserOutboundReceiver {
548    fn drop(&mut self) {
549        let mut pending = 0_i64;
550        let mut bytes = 0_usize;
551        while let Ok(message) = self.messages.try_recv() {
552            pending = pending.saturating_add(1);
553            bytes = bytes.saturating_add(message.bytes);
554        }
555        if pending > 0 {
556            self.metrics.add_ws_outbound_queued_messages(-pending);
557        }
558        if bytes > 0 {
559            self.queued_bytes.fetch_sub(bytes, Ordering::AcqRel);
560        }
561    }
562}
563
564async fn wait_for_overflow(
565    overflow: &mut watch::Receiver<Option<UserOutboundOverflow>>,
566) -> UserOutboundOverflow {
567    loop {
568        if let Some(overflow) = *overflow.borrow_and_update() {
569            return overflow;
570        }
571        if overflow.changed().await.is_err() {
572            // A dropped overflow sender parks this arm forever, avoiding a busy
573            // loop and letting the message arm of the select resolve.
574            pending::<()>().await;
575        }
576    }
577}
578
579#[derive(Debug, Clone)]
580pub(super) struct MessageFanout {
581    recipients: Vec<OutboundSender>,
582    message: RoomEventMessage,
583}
584
585impl MessageFanout {
586    pub(super) fn emit(self) {
587        for recipient in self.recipients {
588            let _ = recipient.send(UserOutbound::Message(self.message.clone()));
589        }
590    }
591}
592
593pub(super) fn fanout_all(
594    recipients: impl IntoIterator<Item = OutboundSender>,
595    message: &RoomEventMessage,
596) -> MessageFanout {
597    MessageFanout {
598        recipients: recipients.into_iter().collect(),
599        message: message.clone(),
600    }
601}
o_sfu_core/engine/room/outbound.rs

o_sfu_core/engine/room/
outbound.rs
