fix(memleak): release call-lifecycle resources + bump rustrtc to 0.3.86

Fixes several memory leaks that accumulated across calls:

- policy: track acquired concurrency slots in PolicyCheckOutcome and release
  them in SipSession::cleanup()/Drop (release_concurrency was never called,
  permanently exhausting max_total after N calls). Abort path releases
  immediately; success path threads holds via DialplanHints/Application.
- bridge: N-peer forwarder task now honors a child cancel token (select!) so it
  stops when the bridge tears down instead of pinning the peers map + extra
  PeerConnections forever. prune_sub_tasks caps live sub-tasks (renegotiation
  churn) and replaces racy try_lock drops with awaited lock + push.
- mixer: MediaMixer gains a Drop that cancels + aborts its 1Hz task;
  MixerRegistry auto-removes mixers once participant count hits 0 so orphaned
  supervisor mixers (hang-up without supervisor_stop) no longer leak.
- conference_mixer: prune route_gains referencing a leaving leg; guard
  participant_count against underflow on duplicate remove.
- engine: store SipFlow capture JoinHandle on MediaSession; stop/restart/
  DestroySession abort it (previously 'stop' was a no-op logging only).
- controller: set_timeout/cancel_timeout now track JoinHandles in a map and
  abort on cancel/re-register/Drop so pending timers don't outlive the call.
- Bump rustrtc to 0.3.86 (SRTP SSRC-context eviction, SCTP flow-control hang,
  notify_one race fix, MediaRelay task exit, PeerConnection task registry).

Author: shenjinti

Cargo.toml

@@ -147,7 +147,7 @@ hex = { version = "0.4.3", optional = true }
 audio-codec = "0.3.39"
 #udio-codec = { path = "../audio-codec", default-features = false }
 #rustrtc = { path = "../rustrtc" }
-rustrtc = "0.3.83"
+rustrtc = "0.3.86"
 reqwest = { version = "0.13.4", default-features = false, features = [
     "json",
     "stream",

src/call/app/controller.rs

@@ -3,11 +3,12 @@ use crate::call::domain::{CallCommand, HangupCommand, LegId, MediaSource};
 use crate::callrecord::CallRecordHangupReason;
 use crate::proxy::proxy_call::sip_session::SipSessionHandle;
 use parking_lot::Mutex;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::time::Duration;
 use thiserror::Error;
 use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
 use tokio::time::Instant;
 use tracing::{info, warn};
 
@@ -57,6 +58,11 @@ pub struct CallController {
     pub(crate) fired_timer_tx: mpsc::UnboundedSender<String>,
     /// Set of timer IDs that have been cancelled and should be suppressed.
     pub(crate) cancelled_timers: Arc<Mutex<HashSet<String>>>,
+    /// JoinHandles of pending timer tasks, keyed by timer id, so they can be
+    /// aborted on cancel/re-register instead of sleeping for the full delay
+    /// (which previously kept Arc clones + the channel sender alive past call
+    /// end and skewed task metrics).
+    pub(crate) timer_tasks: Arc<Mutex<HashMap<String, JoinHandle<()>>>>,
 }
 
 /// Error returned when the remote party hangs up during `collect_dtmf`.
@@ -122,6 +128,7 @@ impl CallController {
             event_rx,
             fired_timer_tx,
             cancelled_timers: Arc::new(Mutex::new(HashSet::new())),
+            timer_tasks: Arc::new(Mutex::new(HashMap::new())),
         };
         (ctrl, fired_timer_rx)
     }
@@ -218,33 +225,44 @@ impl CallController {
     /// After `delay`, [`CallApp::on_timeout`] will be invoked with `id`.
     ///
     /// Calling `set_timeout` with the same `id` before it fires **re-registers**
-    /// the timer (the previous one is cancelled). Use [`cancel_timeout`](Self::cancel_timeout)
+    /// the timer (the previous task is aborted). Use [`cancel_timeout`](Self::cancel_timeout)
     /// to suppress a pending fire without re-registering.
-    ///
-    /// # Panics
-    /// Will not panic; timer tasks are fire-and-forget on a Tokio runtime.
     pub fn set_timeout(&self, id: impl Into<String>, delay: Duration) {
         let id = id.into();
-        // If re-registering, un-cancel any previous suppression.
-            self.cancelled_timers.lock().remove(&id);
+        // Re-registering: abort the previous timer task (if any) and clear any
+        // cancellation flag so the new timer is armed fresh.
+        if let Some(handle) = self.timer_tasks.lock().remove(&id) {
+            handle.abort();
+        }
+        self.cancelled_timers.lock().remove(&id);
+
         let tx = self.fired_timer_tx.clone();
         let cancelled = self.cancelled_timers.clone();
+        let tasks = self.timer_tasks.clone();
         let id_task = id.clone();
-        crate::utils::spawn(async move {
+        let handle = crate::utils::spawn(async move {
             tokio::time::sleep(delay).await;
+            // Self-remove so the handle map does not retain finished tasks.
+            tasks.lock().remove(&id_task);
             // Only fire if not cancelled in the meantime.
             let was_cancelled = cancelled.lock().remove(&id_task);
             if !was_cancelled {
                 let _ = tx.send(id_task);
             }
         });
+        self.timer_tasks.lock().insert(id, handle);
     }
 
     /// Cancel a pending timer previously registered with [`set_timeout`](Self::set_timeout).
     ///
     /// If the timer has already fired, this is a no-op.
     pub fn cancel_timeout(&self, id: &str) {
         self.cancelled_timers.lock().insert(id.to_string());
+        // Abort the sleeping task immediately rather than letting it run for the
+        // remainder of its delay.
+        if let Some(handle) = self.timer_tasks.lock().remove(id) {
+            handle.abort();
+        }
     }
 
     /// Start recording the call audio.
@@ -424,6 +442,17 @@ impl CallController {
     }
 }
 
+impl Drop for CallController {
+    fn drop(&mut self) {
+        // Abort every still-pending timer task so they don't keep sleeping (and
+        // holding their captured Arc/channel clones) after the call has ended.
+        let tasks = std::mem::take(&mut *self.timer_tasks.lock());
+        for (_, handle) in tasks {
+            handle.abort();
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/call/mod.rs

@@ -913,6 +913,11 @@ pub struct Dialplan {
     /// When present, these take priority over the original SIP request headers
     /// when building CallInfo for application flows (IVR, voicemail, etc.).
     pub routed_headers: Option<Vec<rsipstack::sip::Header>>,
+
+    /// Concurrency slots acquired by routing policy checks. Released by the
+    /// session on hangup/teardown. Held in an `Arc<Mutex<..>>` so the shared
+    /// `Arc<Dialplan>` (inside `CallContext`) can still drain them on cleanup.
+    pub concurrency_holds: Arc<Mutex<Vec<crate::call::policy::ConcurrencyHold>>>,
 }
 
 impl std::fmt::Debug for Dialplan {
@@ -973,6 +978,7 @@ impl Dialplan {
             passthrough_failure: false,
             audio_profile: None,
             routed_headers: None,
+            concurrency_holds: Arc::new(Mutex::new(Vec::new())),
         }
     }
 

src/call/policy.rs

@@ -97,6 +97,25 @@ pub enum PolicyCheckStatus {
     Rejected(PolicyRejection),
 }
 
+/// A concurrency slot acquired by a call via policy enforcement. The slot must
+/// be released when the call ends (regardless of how it ends) to avoid
+/// permanently exhausting the configured `max_total` concurrency budget.
+#[derive(Debug, Clone)]
+pub struct ConcurrencyHold {
+    pub policy_id: String,
+    pub scope: String,
+    pub scope_value: String,
+}
+
+/// Outcome of [`PolicyGuard::check_policy`]. Carries the decision plus any
+/// concurrency slots that were acquired during the check (only present when the
+/// call was allowed AND a concurrency limit was enforced).
+#[derive(Debug)]
+pub struct PolicyCheckOutcome {
+    pub status: PolicyCheckStatus,
+    pub concurrency_holds: Vec<ConcurrencyHold>,
+}
+
 pub struct PolicyGuard {
     limiter: Arc<dyn FrequencyLimiter>,
 }
@@ -112,13 +131,61 @@ impl PolicyGuard {
         Self { limiter }
     }
 
+    /// Access the underlying limiter (used to release concurrency slots).
+    pub fn limiter(&self) -> &Arc<dyn FrequencyLimiter> {
+        &self.limiter
+    }
+
+    /// Best-effort release of all concurrency slots held by a call. Called from
+    /// the session cleanup path. Errors are logged and swallowed because a
+    /// failed release must not abort call teardown.
+    pub async fn release_concurrency_holds(holds: &[ConcurrencyHold], limiter: &dyn FrequencyLimiter) {
+        for hold in holds {
+            if let Err(e) = limiter
+                .release_concurrency(&hold.policy_id, &hold.scope, &hold.scope_value)
+                .await
+            {
+                error!(
+                    "failed to release concurrency slot for policy={} scope={}: {}",
+                    hold.policy_id, hold.scope, e
+                );
+            }
+        }
+    }
+
     pub async fn check_policy(
         &self,
         policy_id: &str,
         policy: &PolicySpec,
         caller: &str,
         callee: &str,
         origin_country: Option<&str>,
+    ) -> Result<PolicyCheckOutcome> {
+        let mut holds: Vec<ConcurrencyHold> = Vec::new();
+        let status = self
+            .check_policy_inner(policy_id, policy, caller, callee, origin_country, &mut holds)
+            .await?;
+        // Only carry holds forward when the call was allowed. On rejection the
+        // concurrency check returned false, so nothing was acquired.
+        let concurrency_holds = if status == PolicyCheckStatus::Allowed {
+            holds
+        } else {
+            Vec::new()
+        };
+        Ok(PolicyCheckOutcome {
+            status,
+            concurrency_holds,
+        })
+    }
+
+    async fn check_policy_inner(
+        &self,
+        policy_id: &str,
+        policy: &PolicySpec,
+        caller: &str,
+        callee: &str,
+        origin_country: Option<&str>,
+        holds: &mut Vec<ConcurrencyHold>,
     ) -> Result<PolicyCheckStatus> {
         // 1. Static Checks
         if let Some(prefix) = &policy.called_prefix
@@ -318,17 +385,26 @@ impl PolicyGuard {
             ));
         }
 
-        if let Some(limit) = &policy.concurrency
-            && !self
+        if let Some(limit) = &policy.concurrency {
+            let allowed = self
                 .limiter
                 .check_concurrency(policy_id, "caller", caller, limit.max_total)
-                .await?
-        {
-            let reason = format!("Concurrency limit reached for caller {}", caller);
-            warn!("{}", reason);
-            return Ok(PolicyCheckStatus::Rejected(
-                PolicyRejection::ConcurrencyLimitExceeded(reason),
-            ));
+                .await?;
+            if !allowed {
+                let reason = format!("Concurrency limit reached for caller {}", caller);
+                warn!("{}", reason);
+                return Ok(PolicyCheckStatus::Rejected(
+                    PolicyRejection::ConcurrencyLimitExceeded(reason),
+                ));
+            }
+            // Record the acquired slot so the caller can release it when the
+            // call ends. This is the only limit type that needs explicit release
+            // (frequency/daily limits are window-based and self-expire).
+            holds.push(ConcurrencyHold {
+                policy_id: policy_id.to_string(),
+                scope: "caller".to_string(),
+                scope_value: caller.to_string(),
+            });
         }
 
         Ok(PolicyCheckStatus::Allowed)
@@ -1150,6 +1226,46 @@ mod tests {
             .await;
 
         assert!(result.is_ok());
-        assert_eq!(result.unwrap(), PolicyCheckStatus::Allowed);
+        assert_eq!(result.unwrap().status, PolicyCheckStatus::Allowed);
+    }
+
+    #[tokio::test]
+    async fn test_check_policy_records_concurrency_hold() {
+        let limiter = InMemoryFrequencyLimiter::new();
+        let guard = PolicyGuard::new(limiter.clone());
+
+        let policy = PolicySpec {
+            concurrency: Some(ConcurrencyLimit {
+                max_total: 5,
+                max_per_account: HashMap::new(),
+            }),
+            ..Default::default()
+        };
+
+        let outcome = guard
+            .check_policy("hold-policy", &policy, "caller9", "callee9", None)
+            .await
+            .unwrap();
+
+        assert_eq!(outcome.status, PolicyCheckStatus::Allowed);
+        assert_eq!(outcome.concurrency_holds.len(), 1);
+        let hold = &outcome.concurrency_holds[0];
+        assert_eq!(hold.policy_id, "hold-policy");
+        assert_eq!(hold.scope, "caller");
+        assert_eq!(hold.scope_value, "caller9");
+
+        // Releasing must decrement the counter so subsequent calls are allowed.
+        limiter
+            .release_concurrency(&hold.policy_id, &hold.scope, &hold.scope_value)
+            .await
+            .unwrap();
+
+        // Now a second check for the same caller should still be allowed (slot
+        // was released), proving the release path works.
+        let outcome2 = guard
+            .check_policy("hold-policy", &policy, "caller9", "callee9", None)
+            .await
+            .unwrap();
+        assert_eq!(outcome2.status, PolicyCheckStatus::Allowed);
     }
 }

src/config.rs

@@ -973,6 +973,9 @@ pub struct DialplanHints {
     pub video_policy: Option<crate::proxy::routing::VideoPolicy>,
     /// Per-trunk ringback/early-media audio configuration
     pub ringback: Option<crate::proxy::routing::RingbackAudio>,
+    /// Concurrency slots acquired during routing policy enforcement. The
+    /// session releases them on hangup to avoid leaking the concurrency budget.
+    pub concurrency_holds: Vec<crate::call::policy::ConcurrencyHold>,
 }
 
 impl std::fmt::Debug for DialplanHints {
@@ -1003,6 +1006,7 @@ pub enum RouteResult {
         app_name: String,
         app_params: Option<serde_json::Value>,
         auto_answer: bool,
+        hints: Option<DialplanHints>,
     },
     NotHandled(InviteOption, Option<DialplanHints>),
     Abort(StatusCode, Option<String>),