feat: sliding window bitrate and start paused

Author: lherman-cs

pulsebeam/src/bitrate.rs

@@ -23,7 +23,7 @@ impl Default for BitrateControllerConfig {
             headroom_factor: 1.0,
             max_decay_factor: 0.95,
             emergency_drop_threshold: 0.50,
-            required_up_samples: 5,
+            required_up_samples: 1,
             quantization_step: Bitrate::kbps(10),
         }
     }

pulsebeam/src/participant/downstream/mod.rs

@@ -34,20 +34,23 @@ impl DownstreamAllocator {
             MediaKind::Audio => self.audio.add_track(track),
             MediaKind::Video => self.video.add_track(track),
         }
+        self.update_allocations();
     }
 
     pub fn remove_track(&mut self, track: &TrackReceiver) {
         match track.meta.kind {
             MediaKind::Audio => self.audio.remove_track(&track.meta.id),
             MediaKind::Video => self.video.remove_track(&track.meta.id),
         }
+        self.update_allocations();
     }
 
     pub fn add_slot(&mut self, mid: Mid, kind: MediaKind) {
         match kind {
             MediaKind::Audio => self.audio.add_slot(mid),
             MediaKind::Video => self.video.add_slot(mid),
         }
+        self.update_allocations();
     }
 
     /// Handle BWE and compute both current and desired bitrate in one pass.

pulsebeam/src/participant/downstream/video.rs

@@ -97,7 +97,7 @@ impl VideoAllocator {
             };
 
             next_track.1.assigned_mid.replace(slot.mid);
-            slot.switch_to(next_track.1.track.lowest_quality().clone());
+            slot.assign_to(next_track.1.track.lowest_quality().clone());
         }
     }
 
@@ -301,7 +301,7 @@ impl VideoAllocator {
         for (idx, plan) in committed.drain(..).enumerate() {
             let slot = &mut self.slots[idx];
             if plan.paused {
-                slot.pause(plan.receiver);
+                slot.assign_to(plan.receiver);
             } else {
                 slot.switch_to(plan.receiver);
             }
@@ -310,16 +310,16 @@ impl VideoAllocator {
         let total_allocated = Bitrate::from(total_allocated.max(MIN_BANDWIDTH));
         let total_desired = Bitrate::from(total_desired.max(MIN_BANDWIDTH));
 
-        if self.ticks >= 30 {
-            tracing::debug!(
-                available = %available_bandwidth,
-                budget = %Bitrate::from(budget),
-                allocated = %total_allocated,
-                desired = %total_desired,
-                "allocation summary"
-            );
-            self.ticks = 0;
-        }
+        // if self.ticks >= 30 {
+        tracing::debug!(
+            available = %available_bandwidth,
+            budget = %Bitrate::from(budget),
+            allocated = %total_allocated,
+            desired = %total_desired,
+            "allocation summary"
+        );
+        self.ticks = 0;
+        // }
         self.ticks += 1;
 
         Some((total_allocated, total_desired))
@@ -453,6 +453,11 @@ impl Slot {
         }
     }
 
+    // similar to switch_to but it will start as paused
+    pub fn assign_to(&mut self, receiver: SimulcastReceiver) {
+        self.transition_to(SlotState::Paused { active: receiver });
+    }
+
     pub fn switch_to(&mut self, mut receiver: SimulcastReceiver) {
         if let Some(current) = self.target_receiver()
             && current.rid == receiver.rid
@@ -527,12 +532,6 @@ impl Slot {
         self.transition_to(SlotState::Idle);
     }
 
-    pub fn pause(&mut self, current_receiver: SimulcastReceiver) {
-        self.transition_to(SlotState::Paused {
-            active: current_receiver,
-        });
-    }
-
     fn request_keyframe(&self, kind: KeyframeRequestKind) {
         match self.state() {
             SlotState::Resuming { staging } => {

pulsebeam/src/rtp/monitor.rs

@@ -332,80 +332,85 @@ impl StreamMonitor {
     }
 }
 
+#[derive(Debug)]
+struct Snapshot {
+    ts: Instant,
+    bytes: usize,
+}
+
 #[derive(Debug)]
 pub struct BitrateEstimate {
     controller: BitrateController,
     last_update: Instant,
     accumulated_bytes: usize,
 
-    history: VecDeque<f64>,
-    scratch: Vec<f64>,
-    window_size: usize,
+    history: VecDeque<Snapshot>,
+    window_sum: usize,
+    max_window_duration: Duration,
 }
 
 impl BitrateEstimate {
     pub fn new(now: Instant) -> Self {
         let config = BitrateControllerConfig {
             min_bitrate: Bitrate::kbps(10),
             max_bitrate: Bitrate::mbps(5),
-            default_bitrate: Bitrate::kbps(100),
-            headroom_factor: 1.0,
-            required_up_samples: 1,
-            quantization_step: Bitrate::kbps(10),
+            default_bitrate: Bitrate::kbps(300),
             ..Default::default()
         };
 
-        let controller = BitrateController::new(config);
-        let window_size = 35;
-
         Self {
+            controller: BitrateController::new(config),
             last_update: now,
             accumulated_bytes: 0,
-            history: VecDeque::with_capacity(window_size),
-            scratch: Vec::with_capacity(window_size),
-            window_size,
-            controller,
+            history: VecDeque::new(),
+            window_sum: 0,
+            max_window_duration: Duration::from_secs(1),
         }
     }
 
     pub fn record(&mut self, packet_len: usize) {
-        self.accumulated_bytes = self.accumulated_bytes.saturating_add(packet_len);
+        self.accumulated_bytes += packet_len;
     }
 
     pub fn poll(&mut self, now: Instant) {
         let elapsed = now.saturating_duration_since(self.last_update);
-
         if elapsed < Duration::from_millis(200) {
             return;
         }
 
-        let elapsed_secs = elapsed.as_secs_f64();
-        if elapsed_secs == 0.0 {
-            return;
-        }
-
-        let raw_bps = (self.accumulated_bytes as f64 * 8.0) / elapsed_secs;
+        let snapshot = Snapshot {
+            ts: now,
+            bytes: self.accumulated_bytes,
+        };
+        self.window_sum += snapshot.bytes;
+        self.history.push_back(snapshot);
 
-        if self.history.len() >= self.window_size {
-            self.history.pop_front();
+        while let Some(front) = self.history.front() {
+            if now.saturating_duration_since(front.ts) > self.max_window_duration {
+                let removed = self.history.pop_front().unwrap();
+                self.window_sum = self.window_sum.saturating_sub(removed.bytes);
+            } else {
+                break;
+            }
         }
-        self.history.push_back(raw_bps);
 
-        self.scratch.clear();
-        self.scratch.extend(self.history.iter());
+        let actual_window_duration = if let Some(front) = self.history.front() {
+            now.saturating_duration_since(front.ts)
+        } else {
+            Duration::ZERO
+        };
 
-        let robust_bps = if self.scratch.is_empty() {
-            raw_bps
+        // Safety: ensure we don't divide by zero at the very start
+        let valid_duration = actual_window_duration.max(elapsed).as_secs_f64();
+
+        // sliding window average
+        let bps = if valid_duration > 0.001 {
+            (self.window_sum as f64 * 8.0) / valid_duration
         } else {
-            let target_idx = (self.scratch.len() as f64 * 0.5) as usize;
-            let target_idx = target_idx.min(self.scratch.len().saturating_sub(1));
-            let (_, &mut val, _) = self
-                .scratch
-                .select_nth_unstable_by(target_idx, |a, b| a.total_cmp(b));
-            val
+            0.0
         };
 
-        self.controller.update(Bitrate::from(robust_bps));
+        self.controller.update(Bitrate::from(bps));
         self.last_update = now;
         self.accumulated_bytes = 0;
     }