fix(proxy-engine): fix inbound route browser ringing and provider-facing SDP advertisement while preventing RTP port exhaustion

rust/crates/proxy-engine/src/rtp.rs

@@ -1,17 +1,19 @@
-//! RTP port pool and media forwarding.
+//! RTP port pool for media sockets.
 //!
-//! Manages a pool of even-numbered UDP ports for RTP media.
-//! Each port gets a bound tokio UdpSocket. Supports:
-//! - Direct forwarding (SIP-to-SIP, no transcoding)
-//! - Transcoding forwarding (via codec-lib, e.g. G.722 ↔ Opus)
-//! - Silence generation
-//! - NAT priming
+//! Manages a pool of even-numbered UDP ports for RTP media. `allocate()`
+//! hands back an `Arc<UdpSocket>` to the caller (stored on the owning
+//! `LegInfo`), while the pool itself keeps only a `Weak<UdpSocket>`. When
+//! the call terminates and `LegInfo` is dropped, the strong refcount
+//! reaches zero, the socket is closed, and `allocate()` prunes the dead
+//! weak ref the next time it scans that slot — so the port automatically
+//! becomes available for reuse without any explicit `release()` plumbing.
 //!
-//! Ported from ts/call/rtp-port-pool.ts + sip-leg.ts RTP handling.
+//! This fixes the previous leak where the pool held `Arc<UdpSocket>` and
+//! `release()` was never called, eventually exhausting the port range and
+//! causing "503 Service Unavailable" on new calls.
 
 use std::collections::HashMap;
-use std::net::SocketAddr;
-use std::sync::Arc;
+use std::sync::{Arc, Weak};
 use tokio::net::UdpSocket;
 
 /// A single RTP port allocation.
@@ -24,7 +26,7 @@ pub struct RtpAllocation {
 pub struct RtpPortPool {
     min: u16,
     max: u16,
-    allocated: HashMap<u16, Arc<UdpSocket>>,
+    allocated: HashMap<u16, Weak<UdpSocket>>,
 }
 
 impl RtpPortPool {
@@ -41,11 +43,19 @@ impl RtpPortPool {
     pub async fn allocate(&mut self) -> Option<RtpAllocation> {
         let mut port = self.min;
         while port < self.max {
+            // Prune a dead weak ref at this slot: if the last strong Arc
+            // (held by the owning LegInfo) was dropped when the call ended,
+            // the socket is already closed and the slot is free again.
+            if let Some(weak) = self.allocated.get(&port) {
+                if weak.strong_count() == 0 {
+                    self.allocated.remove(&port);
+                }
+            }
             if !self.allocated.contains_key(&port) {
                 match UdpSocket::bind(format!("0.0.0.0:{port}")).await {
                     Ok(sock) => {
                         let sock = Arc::new(sock);
-                        self.allocated.insert(port, sock.clone());
+                        self.allocated.insert(port, Arc::downgrade(&sock));
                         return Some(RtpAllocation { port, socket: sock });
                     }
                     Err(_) => {
@@ -57,83 +67,6 @@ impl RtpPortPool {
         }
         None // Pool exhausted.
     }
-
-    /// Release a port back to the pool.
-    pub fn release(&mut self, port: u16) {
-        self.allocated.remove(&port);
-        // Socket is dropped when the last Arc reference goes away.
-    }
-
-    pub fn size(&self) -> usize {
-        self.allocated.len()
-    }
-
-    pub fn capacity(&self) -> usize {
-        ((self.max - self.min) / 2) as usize
-    }
-}
-
-/// An active RTP relay between two endpoints.
-/// Receives on `local_socket` and forwards to `remote_addr`.
-pub struct RtpRelay {
-    pub local_port: u16,
-    pub local_socket: Arc<UdpSocket>,
-    pub remote_addr: Option<SocketAddr>,
-    /// If set, transcode packets using this codec session before forwarding.
-    pub transcode: Option<TranscodeConfig>,
-    /// Packets received counter.
-    pub pkt_received: u64,
-    /// Packets sent counter.
-    pub pkt_sent: u64,
-}
-
-pub struct TranscodeConfig {
-    pub from_pt: u8,
-    pub to_pt: u8,
-    pub session_id: String,
-}
-
-impl RtpRelay {
-    pub fn new(port: u16, socket: Arc<UdpSocket>) -> Self {
-        Self {
-            local_port: port,
-            local_socket: socket,
-            remote_addr: None,
-            transcode: None,
-            pkt_received: 0,
-            pkt_sent: 0,
-        }
-    }
-
-    pub fn set_remote(&mut self, addr: SocketAddr) {
-        self.remote_addr = Some(addr);
-    }
-}
-
-/// Send a 1-byte NAT priming packet to open a pinhole.
-pub async fn prime_nat(socket: &UdpSocket, remote: SocketAddr) {
-    let _ = socket.send_to(&[0u8], remote).await;
-}
-
-/// Build an RTP silence frame for PCMU (payload type 0).
-pub fn silence_frame_pcmu() -> Vec<u8> {
-    // 12-byte RTP header + 160 bytes of µ-law silence (0xFF)
-    let mut frame = vec![0u8; 172];
-    frame[0] = 0x80; // V=2
-    frame[1] = 0; // PT=0 (PCMU)
-    // seq, timestamp, ssrc left as 0 — caller should set these
-    frame[12..].fill(0xFF); // µ-law silence
-    frame
-}
-
-/// Build an RTP silence frame for G.722 (payload type 9).
-pub fn silence_frame_g722() -> Vec<u8> {
-    // 12-byte RTP header + 160 bytes of G.722 silence
-    let mut frame = vec![0u8; 172];
-    frame[0] = 0x80; // V=2
-    frame[1] = 9; // PT=9 (G.722)
-    // G.722 silence: all zeros is valid silence
-    frame
 }
 
 /// Build an RTP header with the given parameters.