Add egress loss stat

src/rtp/rtcp/twcc.rs

@@ -1097,6 +1097,53 @@ impl TwccSendRegister {
         Some(&self.queue[index])
     }
 
+    /// Calculate the egress loss for given time window.
+    ///
+    /// **Note:** The register only keeps a limited number of records and using `duration` values
+    /// larger than ~1-2 seconds is liable to be inaccurate since some packets sent might have already
+    /// been evicted from the register.
+    ///
+    /// Packets sent very recently, within one RTT, are not considered because it's not likely that
+    /// a TWCC report could've been generated by the sender and received yet.
+    pub fn loss(&self, duration: Duration, now: Instant) -> Option<f32> {
+        let recent_rtt = self
+            .queue
+            .iter()
+            .rev()
+            .find_map(|r| r.rtt())
+            .map(|rtt| rtt.clamp(Duration::from_millis(5), Duration::from_millis(150)))
+            .unwrap_or(Duration::from_millis(50));
+
+        // Consider only packets that were sent at least one RTT ago. For more recent sends we will
+        // likely not have received the TWCC report in any case.
+        let upper_bound = now - recent_rtt;
+        // Consider only packets in the span specified by the caller
+        let lower_bound = now - duration;
+
+        let packets = self
+            .queue
+            .iter()
+            .rev()
+            .skip_while(|s| s.local_send_time >= upper_bound)
+            .take_while(|s| s.local_send_time >= lower_bound);
+
+        let (total, lost) = packets.fold((0, 0), |(total, lost), s| {
+            let was_lost = s
+                .recv_report
+                .as_ref()
+                .map(|rr| rr.remote_recv_time.is_none())
+                .unwrap_or(true);
+
+            (total + 1, lost + u64::from(was_lost))
+        });
+
+        if total == 0 {
+            return None;
+        }
+
+        Some((lost as f32) / (total as f32))
+    }
+
     /// Get all send records in a range.
     pub fn send_records(
         &self,
@@ -1811,4 +1858,56 @@ mod test {
             vec![0, 1, 2, 3, 4, 5, 6, 7]
         );
     }
+
+    #[test]
+    fn test_twcc_register_loss() {
+        let mut reg = TwccSendRegister::new(25);
+        let mut now = Instant::now();
+        for i in 0..9 {
+            reg.register_seq(i.into(), now, 0);
+            now = now + Duration::from_millis(15);
+        }
+
+        // This packet was sent too recently to be included when calculating loss
+        now = now + Duration::from_millis(35);
+        reg.register_seq(10.into(), now, 0);
+
+        now = now + Duration::from_millis(5);
+        reg.apply_report(
+            Twcc {
+                sender_ssrc: Ssrc::new(),
+                ssrc: Ssrc::new(),
+                base_seq: 0,
+                status_count: 9,
+                reference_time: 35,
+                feedback_count: 0,
+                chunks: [
+                    PacketChunk::VectorDouble(0b11_01_01_01_00_01_00_01, 7),
+                    PacketChunk::Run(PacketStatus::ReceivedSmallDelta, 2),
+                ]
+                .into(),
+                delta: [
+                    Delta::Small(10),
+                    Delta::Small(10),
+                    Delta::Small(10),
+                    Delta::Small(10),
+                    Delta::Small(10),
+                    Delta::Small(10),
+                    Delta::Small(10),
+                ]
+                .into(),
+            },
+            now,
+        )
+        .expect("apply_report to return Some(_)");
+
+        now = now + Duration::from_millis(20);
+        let loss = reg
+            .loss(Duration::from_millis(150), now)
+            .expect("Should be able to calcualte loss");
+
+        let pct = (loss * 100.0).floor() as u32;
+
+        assert_eq!(pct, 33, "The loss percentage should be 33");
+    }
 }

src/session.rs

@@ -867,6 +867,8 @@ impl Session {
         snapshot.tx = snapshot.egress.values().map(|s| s.bytes).sum();
         snapshot.rx = snapshot.ingress.values().map(|s| s.bytes).sum();
         snapshot.bwe_tx = self.bwe.as_ref().and_then(|bwe| bwe.last_estimate());
+
+        snapshot.egress_loss_ratio = self.twcc_tx_register.loss(Duration::from_secs(1), now);
     }
 
     pub fn set_bwe_current_bitrate(&mut self, current_bitrate: Bitrate) {

src/stats.rs

@@ -21,6 +21,7 @@ pub(crate) struct StatsSnapshot {
     pub peer_rx: u64,
     pub tx: u64,
     pub rx: u64,
+    pub egress_loss_ratio: Option<f32>,
     pub ingress: HashMap<(Mid, Option<Rid>), MediaIngressStats>,
     pub egress: HashMap<(Mid, Option<Rid>), MediaEgressStats>,
     pub bwe_tx: Option<Bitrate>,
@@ -34,6 +35,7 @@ impl StatsSnapshot {
             peer_tx: 0,
             tx: 0,
             rx: 0,
+            egress_loss_ratio: None,
             ingress: HashMap::new(),
             egress: HashMap::new(),
             bwe_tx: None,
@@ -68,6 +70,8 @@ pub struct PeerStats {
     pub timestamp: Instant,
     /// The last egress bandwidth estimate from the BWE subsystem, if enabled.
     pub bwe_tx: Option<Bitrate>,
+    /// The egress loss over the last second.
+    pub egress_loss_fraction: Option<f32>,
 }
 
 /// An event carrying stats for every (mid, rid) in egress direction
@@ -184,6 +188,7 @@ impl Stats {
             bytes_tx: snapshot.tx,
             timestamp: snapshot.timestamp,
             bwe_tx: snapshot.bwe_tx,
+            egress_loss_fraction: snapshot.egress_loss_ratio,
         };
 
         self.events.push_back(StatsEvent::Peer(event));