Don't retain RR loss info when stats are disabled

This prevents unbounded memory growth of the Vec that holds onto packet
loss records derived from RTCP Receiver Reports, if the method that would
drain that Vec will never get called.

Author: OxleyS

Diff for: src/session.rs

@@ -131,7 +131,7 @@ impl Session {
         Session {
             id,
             medias: vec![],
-            streams: Streams::default(),
+            streams: Streams::new(config.stats_interval.is_some()),
             app: None,
             reordering_size_audio: config.reordering_size_audio,
             reordering_size_video: config.reordering_size_video,

Diff for: src/streams/mod.rs

@@ -149,6 +149,10 @@ pub(crate) struct Streams {
     /// Whether nack reports are enabled. This is an optimization to avoid too frequent
     /// Session::nack_at() when we don't need to send nacks.
     any_nack_active: Option<bool>,
+
+    /// Whether periodic statistics reports are expected to be generated. This informs us on
+    /// whether we should be holding onto data needed for those reports or not.
+    needs_stat_reports: bool,
 }
 
 /// Delay between cleaning up the RxLookup.
@@ -164,8 +168,8 @@ struct RxLookup {
     last_used: Instant,
 }
 
-impl Default for Streams {
-    fn default() -> Self {
+impl Streams {
+    pub(crate) fn new(needs_stat_reports: bool) -> Self {
         Self {
             streams_rx: Default::default(),
             rx_lookup: Default::default(),
@@ -174,11 +178,10 @@ impl Default for Streams {
             default_ssrc_tx: 0.into(), // this will be changed
             mids_to_report: Vec::with_capacity(10),
             any_nack_active: None,
+            needs_stat_reports,
         }
     }
-}
 
-impl Streams {
     pub(crate) fn map_dynamic_by_rid(
         &mut self,
         ssrc: Ssrc,
@@ -328,7 +331,7 @@ impl Streams {
     ) -> &mut StreamTx {
         self.streams_tx
             .entry(ssrc)
-            .or_insert_with(|| StreamTx::new(ssrc, rtx, midrid))
+            .or_insert_with(|| StreamTx::new(ssrc, rtx, midrid, self.needs_stat_reports))
     }
 
     pub fn remove_stream_tx(&mut self, ssrc: Ssrc) -> bool {

Diff for: src/streams/send.rs

@@ -153,7 +153,7 @@ pub(crate) struct StreamTxStats {
     /// Can be null in case of missing or bad reports
     rtt: Option<f32>,
     /// losses collecter from RR (known packets, lost ratio)
-    losses: Vec<(u64, f32)>,
+    losses: Option<Vec<(u64, f32)>>,
 
     /// `None` if `rtx_ratio_cap` is `None`.
     bytes_transmitted: Option<ValueHistory<u64>>,
@@ -162,26 +162,13 @@ pub(crate) struct StreamTxStats {
     bytes_retransmitted: Option<ValueHistory<u64>>,
 }
 
-impl Default for StreamTxStats {
-    fn default() -> Self {
-        Self {
-            bytes: 0,
-            bytes_resent: 0,
-            packets: 0,
-            packets_resent: 0,
-            firs: 0,
-            plis: 0,
-            nacks: 0,
-            rtt: None,
-            losses: Vec::default(),
-            bytes_transmitted: Some(Default::default()),
-            bytes_retransmitted: Some(Default::default()),
-        }
-    }
-}
-
 impl StreamTx {
-    pub(crate) fn new(ssrc: Ssrc, rtx: Option<Ssrc>, midrid: MidRid) -> Self {
+    pub(crate) fn new(
+        ssrc: Ssrc,
+        rtx: Option<Ssrc>,
+        midrid: MidRid,
+        needs_stat_reports: bool,
+    ) -> Self {
         debug!("Create StreamTx for SSRC: {}", ssrc);
 
         StreamTx {
@@ -205,7 +192,7 @@ impl StreamTx {
             last_sender_report: already_happened(),
             pending_request_keyframe: None,
             pending_request_remb: None,
-            stats: StreamTxStats::default(),
+            stats: StreamTxStats::new(needs_stat_reports),
             rtx_ratio: (0.0, already_happened()),
             pt_for_padding: None,
             remote_acked_ssrc: false,
@@ -1071,6 +1058,25 @@ impl StreamTx {
 }
 
 impl StreamTxStats {
+    fn new(needs_stat_reports: bool) -> Self {
+        Self {
+            bytes: 0,
+            bytes_resent: 0,
+            packets: 0,
+            packets_resent: 0,
+            firs: 0,
+            plis: 0,
+            nacks: 0,
+            rtt: None,
+
+            // This Vec is unbounded, don't collect if we're never going to drain it
+            losses: needs_stat_reports.then_some(Vec::default()),
+
+            bytes_transmitted: Some(Default::default()),
+            bytes_retransmitted: Some(Default::default()),
+        }
+    }
+
     fn update_packet_counts(&mut self, bytes: u64, is_resend: bool) {
         self.packets += 1;
         self.bytes += bytes;
@@ -1097,41 +1103,42 @@ impl StreamTxStats {
         let rtt = calculate_rtt_ms(ntp_time, r.last_sr_delay, r.last_sr_time);
         self.rtt = rtt;
 
-        let ext_seq = {
-            let prev = self.losses.last().map(|s| s.0).unwrap_or(r.max_seq as u64);
-            let next = (r.max_seq & 0xffff) as u16;
-            extend_u16(Some(prev), next)
-        };
+        if let Some(losses) = self.losses.as_mut() {
+            let ext_seq = {
+                let prev = losses.last().map(|s| s.0).unwrap_or(r.max_seq as u64);
+                let next = (r.max_seq & 0xffff) as u16;
+                extend_u16(Some(prev), next)
+            };
 
-        self.losses
-            .push((ext_seq, r.fraction_lost as f32 / u8::MAX as f32));
+            losses.push((ext_seq, r.fraction_lost as f32 / u8::MAX as f32));
+        }
     }
 
     pub(crate) fn fill(&mut self, snapshot: &mut StatsSnapshot, midrid: MidRid, now: Instant) {
         if self.bytes == 0 {
             return;
         }
 
-        let loss = {
+        let loss = self.losses.as_mut().and_then(|losses| {
             let mut value = 0_f32;
             let mut total_weight = 0_u64;
 
             // just in case we received RRs out of order
-            self.losses.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap());
+            losses.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap());
 
             // average known RR losses weighted by their number of packets
-            for it in self.losses.windows(2) {
+            for it in losses.windows(2) {
                 let [prev, next] = it else { continue };
                 let weight = next.0.saturating_sub(prev.0);
                 value += next.1 * weight as f32;
                 total_weight += weight;
             }
 
+            losses.drain(..losses.len().saturating_sub(1));
+
             let result = value / total_weight as f32;
             result.is_finite().then_some(result)
-        };
-
-        self.losses.drain(..self.losses.len().saturating_sub(1));
+        });
 
         snapshot.egress.insert(
             midrid,