fix(security): cap built-in relay fan-out per poll tick

src/server/mod.rs

@@ -60,6 +60,13 @@ const MAX_RECV_BYTES_PER_CONN_PER_POLL: usize = 256 * 1024;
 /// connections in this one.
 const MAX_BUDGET_DRAIN_PASSES_PER_CONN_PER_POLL: usize = 3;
 
+/// Cap total `send_frame` calls issued while fanning publisher media out to
+/// players in one `process_connections` pass. Without this, a full
+/// `pending_relay` batch multiplied by every playing connection can
+/// monopolize the single-threaded poll loop (e.g. 1024 frames × 255 players
+/// ≈ 261k chunk encodes per poll tick).
+const MAX_RELAY_SENDS_PER_POLL: usize = 4096;
+
 /// Cached codec headers and last keyframe for a (app, stream_name) pair.
 /// Replayed to players that join after the publisher has already sent headers.
 struct StreamCache {
@@ -585,7 +592,7 @@ impl Server {
 
         // Collect all frames queued by publishers, then relay them to players
         // on the same (app, stream_name) pair.
-        let relay_frames: Vec<_> = self
+        let mut relay_frames: Vec<_> = self
             .connections
             .iter_mut()
             .flat_map(|c| c.pending_relay.drain(..))
@@ -634,7 +641,14 @@ impl Server {
 
         // Update per-stream cache and relay each frame in order so players
         // receive frames in the same sequence the publisher sent them.
+        let mut relay_sends = 0usize;
+        let mut relay_processed = 0usize;
         for frame in &relay_frames {
+            let player_count = self.count_relay_players(frame);
+            if player_count > 0 && relay_sends + player_count > MAX_RELAY_SENDS_PER_POLL {
+                break;
+            }
+
             let abandon_key = (
                 frame.app.clone(),
                 frame.stream_name.clone(),
@@ -644,14 +658,7 @@ impl Server {
                 self.cache_relay_frame(frame);
             }
             for (i, conn) in self.connections.iter_mut().enumerate() {
-                let is_player = conn.relay_enabled
-                    && conn.transport.is_some()
-                    && conn
-                        .current_stream
-                        .as_ref()
-                        .map(|s| s.is_playing && conn.relay_route_key() == frame.stream_name)
-                        .unwrap_or(false);
-                if !is_player || conn.app != frame.app {
+                if !Self::conn_is_relay_player(conn, frame) {
                     continue;
                 }
                 if conn
@@ -667,6 +674,11 @@ impl Server {
                     closed.push(i);
                 }
             }
+            relay_sends += player_count;
+            relay_processed += 1;
+        }
+        for frame in relay_frames.drain(relay_processed..) {
+            self.requeue_relay_frame(frame);
         }
 
         // Flush all connections.
@@ -741,6 +753,34 @@ impl Server {
         }
     }
 
+    fn conn_is_relay_player(conn: &Conn, frame: &crate::session::conn::RelayFrame) -> bool {
+        conn.relay_enabled
+            && conn.transport.is_some()
+            && conn.app == frame.app
+            && conn
+                .current_stream
+                .as_ref()
+                .map(|s| s.is_playing && conn.relay_route_key() == frame.stream_name)
+                .unwrap_or(false)
+    }
+
+    fn count_relay_players(&self, frame: &crate::session::conn::RelayFrame) -> usize {
+        self.connections
+            .iter()
+            .filter(|conn| Self::conn_is_relay_player(conn, frame))
+            .count()
+    }
+
+    fn requeue_relay_frame(&mut self, frame: crate::session::conn::RelayFrame) {
+        if let Some(conn) = self
+            .connections
+            .iter_mut()
+            .find(|c| c.conn_id == frame.publisher_conn_id)
+        {
+            conn.pending_relay.push(frame);
+        }
+    }
+
     fn cache_relay_frame(&mut self, frame: &crate::session::conn::RelayFrame) {
         let is_avc_header = frame.frame_type == FrameType::Video
             && frame.payload.len() >= 2
@@ -941,6 +981,17 @@ mod tests {
         assert!(MAX_RECV_BYTES_PER_CONN_PER_POLL <= 1024 * 1024);
     }
 
+    #[test]
+    fn relay_send_budget_limits_worst_case_player_fan_out() {
+        // One publisher can queue 1024 frames; with 256 connections that could
+        // be 261_120 send_frame calls per poll without a relay budget.
+        let worst_case = 1024 * 256;
+        assert!(
+            MAX_RELAY_SENDS_PER_POLL < worst_case / 10,
+            "relay budget should be well below unbounded fan-out"
+        );
+    }
+
     fn test_server() -> Server {
         Server::new(ServerConfig {
             max_connections: 4,