feat(pacing): add send pacing, reconnect backoff and log rate limiting

Pure scheduling logic with an injected `now`, so the rules of spec 4.3.6
(fps cap, keepalive, idle clear, heartbeat) are unit-testable without I/O.

src/lib.rs

@@ -3,4 +3,5 @@
 
 pub mod config;
 pub mod hyperhdr;
+pub mod pacing;
 pub mod slot;

src/pacing.rs

@@ -0,0 +1,380 @@
+//! Pure scheduling logic for the HyperHDR connection: send pacing, reconnect backoff
+//! and log rate limiting. Nothing here does I/O or reads the clock; `now` is injected.
+
+use std::time::{Duration, Instant};
+
+/// What the network thread should do next.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Action {
+    /// Send the newest frame, which has not been sent yet.
+    SendFrame,
+    /// Re-send the last frame so the server does not time out the connection.
+    SendKeepalive,
+    /// Clear our priority: no new frames for `idle_clear`.
+    Clear,
+    /// Keep the connection open while cleared, without showing an image.
+    Heartbeat,
+    /// Nothing to do until the given instant, or until the next event if `None`.
+    Wait(Option<Instant>),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PacingConfig {
+    /// `1 / max_fps`.
+    pub min_interval: Duration,
+    pub keepalive: Duration,
+    pub idle_clear: Option<Duration>,
+}
+
+/// Send rules (spec 4.3.6), evaluated as a function of `now` and the recorded events.
+#[derive(Debug)]
+pub struct Pacer {
+    cfg: PacingConfig,
+    has_frame: bool,
+    unsent: bool,
+    cleared: bool,
+    last_frame_at: Option<Instant>,
+    last_image_at: Option<Instant>,
+    last_message_at: Option<Instant>,
+}
+
+impl Pacer {
+    pub fn new(cfg: PacingConfig) -> Self {
+        Self {
+            cfg,
+            has_frame: false,
+            unsent: false,
+            cleared: false,
+            last_frame_at: None,
+            last_image_at: None,
+            last_message_at: None,
+        }
+    }
+
+    /// A new frame arrived. Returns `true` if it replaces a frame that was never sent.
+    pub fn on_new_frame(&mut self, now: Instant) -> bool {
+        let superseded = self.unsent;
+        self.has_frame = true;
+        self.unsent = true;
+        self.cleared = false;
+        self.last_frame_at = Some(now);
+        superseded
+    }
+
+    /// The source disappeared: forget the last frame.
+    pub fn on_source_gone(&mut self) {
+        self.has_frame = false;
+        self.unsent = false;
+        self.cleared = false;
+        self.last_frame_at = None;
+    }
+
+    /// A new connection was established (and registered).
+    pub fn on_connected(&mut self) {
+        self.last_image_at = None;
+        self.last_message_at = None;
+        if self.has_frame && !self.cleared {
+            self.unsent = true;
+        }
+    }
+
+    /// `action` was written successfully at `now`.
+    pub fn on_sent(&mut self, action: Action, now: Instant) {
+        match action {
+            Action::SendFrame | Action::SendKeepalive => {
+                self.unsent = false;
+                self.last_image_at = Some(now);
+                self.last_message_at = Some(now);
+            }
+            Action::Clear => {
+                self.cleared = true;
+                self.last_message_at = Some(now);
+            }
+            Action::Heartbeat => self.last_message_at = Some(now),
+            Action::Wait(_) => {}
+        }
+    }
+
+    /// Whether a connection to HyperHDR should exist while disconnected.
+    /// Marks the source as cleared if the idle timeout expired in the meantime.
+    pub fn wants_connection(&mut self, now: Instant) -> bool {
+        if self.idle_expired(now) {
+            self.cleared = true;
+            self.unsent = false;
+        }
+        self.has_frame && !self.cleared
+    }
+
+    pub fn is_cleared(&self) -> bool {
+        self.cleared
+    }
+
+    fn idle_expired(&self, now: Instant) -> bool {
+        match (self.cfg.idle_clear, self.last_frame_at) {
+            (Some(idle), Some(at)) => !self.unsent && now >= at + idle,
+            _ => false,
+        }
+    }
+
+    /// Next action for a connected client.
+    pub fn poll(&self, now: Instant) -> Action {
+        let image_allowed_at = self.last_image_at.map(|t| t + self.cfg.min_interval);
+
+        if self.unsent {
+            return match image_allowed_at {
+                Some(at) if now < at => Action::Wait(Some(at)),
+                _ => Action::SendFrame,
+            };
+        }
+        if !self.has_frame {
+            return Action::Wait(None);
+        }
+
+        let keepalive_at = self.last_message_at.map_or(now, |t| t + self.cfg.keepalive);
+        if self.cleared {
+            return if now >= keepalive_at {
+                Action::Heartbeat
+            } else {
+                Action::Wait(Some(keepalive_at))
+            };
+        }
+
+        let idle_at = match (self.cfg.idle_clear, self.last_frame_at) {
+            (Some(idle), Some(at)) => Some(at + idle),
+            _ => None,
+        };
+        if idle_at.is_some_and(|at| now >= at) {
+            return Action::Clear;
+        }
+
+        let keepalive_at = image_allowed_at.map_or(keepalive_at, |at| at.max(keepalive_at));
+        if now >= keepalive_at {
+            return Action::SendKeepalive;
+        }
+        Action::Wait(Some(
+            idle_at.map_or(keepalive_at, |at| at.min(keepalive_at)),
+        ))
+    }
+}
+
+/// Exponential reconnect backoff (spec 4.3.7).
+#[derive(Debug, Clone)]
+pub struct Backoff {
+    initial: Duration,
+    max: Duration,
+    reset_after: Duration,
+    next: Duration,
+}
+
+impl Backoff {
+    pub fn new(initial: Duration, max: Duration, reset_after: Duration) -> Self {
+        Self {
+            initial,
+            max,
+            reset_after,
+            next: initial,
+        }
+    }
+
+    /// Delay before the next attempt. `session` is how long the failed connection had been
+    /// up, or `None` if connecting itself failed.
+    pub fn on_failure(&mut self, session: Option<Duration>) -> Duration {
+        if session.is_some_and(|s| s >= self.reset_after) {
+            self.next = self.initial;
+        }
+        let delay = self.next;
+        self.next = (self.next * 2).min(self.max);
+        delay
+    }
+
+    pub fn reset(&mut self) {
+        self.next = self.initial;
+    }
+}
+
+/// Lets one event through per `interval` and counts the suppressed ones.
+#[derive(Debug, Clone)]
+pub struct RateLimit {
+    interval: Duration,
+    last: Option<Instant>,
+    suppressed: u64,
+}
+
+impl RateLimit {
+    pub fn new(interval: Duration) -> Self {
+        Self {
+            interval,
+            last: None,
+            suppressed: 0,
+        }
+    }
+
+    /// Returns `Some(suppressed since the last emitted event)` if this event should be logged.
+    pub fn check(&mut self, now: Instant) -> Option<u64> {
+        match self.last {
+            Some(last) if now < last + self.interval => {
+                self.suppressed += 1;
+                None
+            }
+            _ => {
+                self.last = Some(now);
+                Some(std::mem::take(&mut self.suppressed))
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const MS: Duration = Duration::from_millis(1);
+
+    fn pacer(idle_clear: Option<Duration>) -> (Pacer, Instant) {
+        let cfg = PacingConfig {
+            min_interval: 100 * MS,
+            keepalive: 1000 * MS,
+            idle_clear,
+        };
+        (Pacer::new(cfg), Instant::now())
+    }
+
+    #[test]
+    fn nothing_without_frames() {
+        let (p, t0) = pacer(None);
+        assert_eq!(p.poll(t0), Action::Wait(None));
+    }
+
+    #[test]
+    fn fps_ceiling_and_freshest_wins() {
+        let (mut p, t0) = pacer(None);
+        assert!(!p.on_new_frame(t0));
+        assert_eq!(p.poll(t0), Action::SendFrame);
+        p.on_sent(Action::SendFrame, t0);
+
+        // Too early: wait for the slot, newer frames replace the pending one.
+        assert!(!p.on_new_frame(t0 + 10 * MS));
+        assert_eq!(p.poll(t0 + 10 * MS), Action::Wait(Some(t0 + 100 * MS)));
+        assert!(p.on_new_frame(t0 + 20 * MS));
+        assert_eq!(p.poll(t0 + 99 * MS), Action::Wait(Some(t0 + 100 * MS)));
+        assert_eq!(p.poll(t0 + 100 * MS), Action::SendFrame);
+        p.on_sent(Action::SendFrame, t0 + 100 * MS);
+
+        // Late frame goes out immediately.
+        p.on_new_frame(t0 + 350 * MS);
+        assert_eq!(p.poll(t0 + 350 * MS), Action::SendFrame);
+    }
+
+    #[test]
+    fn keepalive_repeats_last_frame() {
+        let (mut p, t0) = pacer(None);
+        p.on_new_frame(t0);
+        p.on_sent(Action::SendFrame, t0);
+        assert_eq!(p.poll(t0 + 500 * MS), Action::Wait(Some(t0 + 1000 * MS)));
+        assert_eq!(p.poll(t0 + 1000 * MS), Action::SendKeepalive);
+        p.on_sent(Action::SendKeepalive, t0 + 1000 * MS);
+        assert_eq!(p.poll(t0 + 1001 * MS), Action::Wait(Some(t0 + 2000 * MS)));
+    }
+
+    #[test]
+    fn keepalive_respects_fps_ceiling() {
+        let cfg = PacingConfig {
+            min_interval: 500 * MS,
+            keepalive: 100 * MS,
+            idle_clear: None,
+        };
+        let mut p = Pacer::new(cfg);
+        let t0 = Instant::now();
+        p.on_new_frame(t0);
+        p.on_sent(Action::SendFrame, t0);
+        assert_eq!(p.poll(t0 + 100 * MS), Action::Wait(Some(t0 + 500 * MS)));
+        assert_eq!(p.poll(t0 + 500 * MS), Action::SendKeepalive);
+    }
+
+    #[test]
+    fn idle_clear_and_resume() {
+        let (mut p, t0) = pacer(Some(3000 * MS));
+        p.on_new_frame(t0);
+        p.on_sent(Action::SendFrame, t0);
+        for s in 1..3 {
+            let t = t0 + s * 1000 * MS;
+            assert_eq!(p.poll(t), Action::SendKeepalive);
+            p.on_sent(Action::SendKeepalive, t);
+        }
+        assert_eq!(p.poll(t0 + 2500 * MS), Action::Wait(Some(t0 + 3000 * MS)));
+        let t_clear = t0 + 3000 * MS;
+        assert_eq!(p.poll(t_clear), Action::Clear);
+        p.on_sent(Action::Clear, t_clear);
+        assert!(p.is_cleared());
+
+        // No more frame keepalives, only heartbeats.
+        assert_eq!(
+            p.poll(t_clear + 500 * MS),
+            Action::Wait(Some(t_clear + 1000 * MS))
+        );
+        assert_eq!(p.poll(t_clear + 1000 * MS), Action::Heartbeat);
+        p.on_sent(Action::Heartbeat, t_clear + 1000 * MS);
+
+        // A new frame resumes sending immediately.
+        let t_new = t_clear + 1200 * MS;
+        p.on_new_frame(t_new);
+        assert!(!p.is_cleared());
+        assert_eq!(p.poll(t_new), Action::SendFrame);
+    }
+
+    #[test]
+    fn pending_frame_is_not_idle() {
+        let (mut p, t0) = pacer(Some(1000 * MS));
+        p.on_new_frame(t0);
+        assert!(p.wants_connection(t0 + 5000 * MS));
+        assert_eq!(p.poll(t0 + 5000 * MS), Action::SendFrame);
+    }
+
+    #[test]
+    fn connection_wanted_only_with_fresh_frames() {
+        let (mut p, t0) = pacer(Some(2000 * MS));
+        assert!(!p.wants_connection(t0));
+        p.on_new_frame(t0);
+        p.on_sent(Action::SendFrame, t0);
+        assert!(p.wants_connection(t0 + 1000 * MS));
+        assert!(!p.wants_connection(t0 + 2000 * MS));
+        p.on_new_frame(t0 + 3000 * MS);
+        assert!(p.wants_connection(t0 + 3000 * MS));
+        p.on_source_gone();
+        assert!(!p.wants_connection(t0 + 3000 * MS));
+    }
+
+    #[test]
+    fn reconnect_resends_last_frame() {
+        let (mut p, t0) = pacer(None);
+        p.on_new_frame(t0);
+        p.on_sent(Action::SendFrame, t0);
+        p.on_connected();
+        assert_eq!(p.poll(t0 + MS), Action::SendFrame);
+    }
+
+    #[test]
+    fn backoff_sequence_and_reset() {
+        let mut b = Backoff::new(500 * MS, 10_000 * MS, 30_000 * MS);
+        let delays: Vec<_> = (0..7).map(|_| b.on_failure(None).as_millis()).collect();
+        assert_eq!(delays, [500, 1000, 2000, 4000, 8000, 10000, 10000]);
+        // Short session does not reset.
+        assert_eq!(b.on_failure(Some(29_000 * MS)).as_millis(), 10000);
+        // Long session does.
+        assert_eq!(b.on_failure(Some(31_000 * MS)).as_millis(), 500);
+        assert_eq!(b.on_failure(None).as_millis(), 1000);
+        b.reset();
+        assert_eq!(b.on_failure(None).as_millis(), 500);
+    }
+
+    #[test]
+    fn rate_limit() {
+        let mut r = RateLimit::new(10_000 * MS);
+        let t0 = Instant::now();
+        assert_eq!(r.check(t0), Some(0));
+        assert_eq!(r.check(t0 + 1000 * MS), None);
+        assert_eq!(r.check(t0 + 9999 * MS), None);
+        assert_eq!(r.check(t0 + 10_000 * MS), Some(2));
+        assert_eq!(r.check(t0 + 10_001 * MS), None);
+    }
+}