Cap per-connection recv drain to prevent poll-loop starvation (fixes #76)

src/server/mod.rs

@@ -32,6 +32,12 @@ const MAX_PENDING_TLS_HANDSHAKES: usize = 128;
 #[cfg(feature = "tls")]
 const TLS_HANDSHAKE_TIMEOUT_SECS: u64 = 10;
 
+/// Maximum inbound bytes drained from one connection per `process_connections`
+/// pass. Without this cap a peer that keeps the kernel recv buffer full can
+/// monopolize the single-threaded poll loop and starve every other session
+/// until its socket buffer is empty.
+const MAX_RECV_BYTES_PER_CONN_PER_POLL: usize = 256 * 1024;
+
 /// 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 {
@@ -455,18 +461,24 @@ impl Server {
         // Drive recv/processing for every connection.
         self.drain_pending_cache_evictions();
         for (i, conn) in self.connections.iter_mut().enumerate() {
+            let mut bytes_drained = 0usize;
             loop {
+                if bytes_drained >= MAX_RECV_BYTES_PER_CONN_PER_POLL {
+                    break;
+                }
                 let Some(transport) = conn.transport.as_mut() else {
                     closed.push(i);
                     break;
                 };
                 let mut again = 0i32;
                 let n = transport.recv(&mut buf, &mut again);
                 if n > 0 {
-                    if conn.recv(&buf[..n as usize]).is_err() {
+                    let chunk_len = n as usize;
+                    if conn.recv(&buf[..chunk_len]).is_err() {
                         closed.push(i);
                         break;
                     }
+                    bytes_drained += chunk_len;
                 } else if n == 0 {
                     closed.push(i);
                     break;
@@ -807,3 +819,18 @@ impl Drop for Server {
         self.running = false;
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn recv_budget_is_at_least_one_socket_read() {
+        assert!(MAX_RECV_BYTES_PER_CONN_PER_POLL >= 65536);
+    }
+
+    #[test]
+    fn recv_budget_is_small_enough_for_fairness_across_connections() {
+        assert!(MAX_RECV_BYTES_PER_CONN_PER_POLL <= 1024 * 1024);
+    }
+}