refactor: replace unbounded control snapshot channel with tokio watch

src/baseliner.rs

@@ -79,7 +79,7 @@ pub struct Baseliner {
     pub reselect_trigger: Sender<bool>,
     pub start_time: Instant,
     pub stats_rx: Receiver<MeasurementEvent>,
-    pub control_tx: Sender<ControlSnapshot>,
+    pub control_tx: tokio::sync::watch::Sender<Option<ControlSnapshot>>,
     pub selection_tx: Option<Sender<ControlSnapshot>>,
     pub baseline_metrics: MetricsSender,
     pub event_metrics: MetricsSender,
@@ -236,7 +236,7 @@ impl Baseliner {
             if let Some(selection_tx) = &self.selection_tx {
                 let _ = selection_tx.send(snapshot.clone());
             }
-            self.control_tx.send(snapshot)?;
+            self.control_tx.send(Some(snapshot))?;
         }
     }
 }

src/main.rs

@@ -40,6 +40,7 @@ use std::sync::{Arc, RwLock};
 use std::thread;
 use std::thread::{JoinHandle, sleep};
 use std::time::{Duration, Instant};
+use tokio::sync::watch;
 
 struct ReflectorSetup {
     peers: Arc<RwLock<Vec<SocketAddr>>>,
@@ -345,7 +346,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         reselection_enabled,
     } = setup_reflectors(settings)?;
 
-    let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded();
+    let (control_snapshot_tx, control_snapshot_rx) = watch::channel(None);
     let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
     let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded();
     #[cfg(feature = "irtt")]

src/ratecontroller.rs

@@ -16,7 +16,6 @@ use crate::platform::{
 use crate::settings::Settings;
 use crate::time::{ClockSample, Realtime};
 use crate::util::{ArcRwLock, RwLockExt};
-use flume::{Receiver, Sender};
 use log::{debug, info, warn};
 use std::fs::File;
 use std::io::Write;
@@ -84,20 +83,6 @@ impl<H> State<H> {
     }
 }
 
-fn drain_latest_snapshot(
-    snapshot_rx: &Receiver<ControlSnapshot>,
-    latest: &mut Option<ControlSnapshot>,
-) {
-    while let Ok(snapshot) = snapshot_rx.try_recv() {
-        if latest
-            .as_ref()
-            .is_none_or(|current| snapshot.generated_at >= current.generated_at)
-        {
-            *latest = Some(snapshot);
-        }
-    }
-}
-
 fn deltas_from_snapshot(
     snapshot: Option<&ControlSnapshot>,
     reflectors: &[MeasurementStream],
@@ -137,10 +122,9 @@ fn deltas_from_snapshot(
 
 pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
     settings: Settings,
-    control_snapshot: Option<ControlSnapshot>,
-    control_snapshot_rx: Receiver<ControlSnapshot>,
+    control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
     reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
-    reselect_trigger: Sender<bool>,
+    reselect_trigger: flume::Sender<bool>,
     state_dl: State<T::Handle>,
     state_ul: State<T::Handle>,
     rng: fastrand::Rng,
@@ -152,9 +136,9 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
 impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
     fn new_with_backends(
         settings: Settings,
-        control_snapshot_rx: Receiver<ControlSnapshot>,
+        control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
         reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
-        reselect_trigger: Sender<bool>,
+        reselect_trigger: flume::Sender<bool>,
         metrics: MetricsSender,
         mut stats_provider: S,
         mut traffic_control: T,
@@ -179,7 +163,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
 
         Ok(Self {
             settings,
-            control_snapshot: None,
             control_snapshot_rx,
             reflectors_lock,
             reselect_trigger,
@@ -479,15 +462,17 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
     }
 
     fn update_deltas(&mut self) -> anyhow::Result<()> {
-        drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot);
         let now_t = Instant::now();
         let reflectors = self.reflectors_lock.read_anyhow()?;
-        let (down_deltas, up_deltas) = deltas_from_snapshot(
-            self.control_snapshot.as_ref(),
-            &reflectors,
-            now_t,
-            self.settings.advanced_settings.tick_interval,
-        );
+        let (down_deltas, up_deltas) = {
+            let snapshot = self.control_snapshot_rx.borrow();
+            deltas_from_snapshot(
+                snapshot.as_ref(),
+                &reflectors,
+                now_t,
+                self.settings.advanced_settings.tick_interval,
+            )
+        };
         self.state_dl.deltas = down_deltas;
         self.state_ul.deltas = up_deltas;
 
@@ -513,9 +498,9 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
 impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
     pub fn new(
         settings: Settings,
-        control_snapshot_rx: Receiver<ControlSnapshot>,
+        control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
         reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
-        reselect_trigger: Sender<bool>,
+        reselect_trigger: flume::Sender<bool>,
         metrics: MetricsSender,
     ) -> anyhow::Result<Self> {
         Self::new_with_backends(
@@ -589,29 +574,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn drains_queued_snapshots_and_keeps_the_newest() {
-        let start = Instant::now();
-        let newest_at = start + Duration::from_secs(2);
-        let (tx, rx) = flume::unbounded();
-        tx.send(snapshot(start + Duration::from_secs(1), 1.0))
-            .unwrap();
-        tx.send(snapshot(newest_at, 2.0)).unwrap();
-        let mut latest = None;
-
-        drain_latest_snapshot(&rx, &mut latest);
-
-        let latest = latest.unwrap();
-        assert!(rx.is_empty());
-        assert_eq!(latest.generated_at, newest_at);
-        assert_eq!(
-            latest.reflectors[&stream(0, "192.0.2.1".parse().unwrap())]
-                .recent
-                .down,
-            2.0
-        );
-    }
-
     #[test]
     fn calculates_sorted_deltas_from_active_fresh_snapshot_state() {
         let now = Instant::now();