Refactor fairness logic with injectable clock

examples/metadata_routing.rs

@@ -60,7 +60,7 @@ impl FrameMetadata for HeaderSerializer {
 struct Ping;
 
 #[derive(bincode::Decode, bincode::Encode)]
-#[expect(
+#[allow(
     dead_code,
     reason = "placeholder for demonstration of metadata routing"
 )]

examples/ping_pong.rs

@@ -40,7 +40,10 @@ const PING_ID: u32 = 1;
 ///
 /// The middleware chain generates the actual response, so this
 /// handler intentionally performs no work.
-#[allow(clippy::unused_async)]
+#[allow(
+    clippy::unused_async,
+    reason = "handler intentionally performs no work"
+)]
 async fn ping_handler() {}
 
 struct PongMiddleware;

src/connection.rs

@@ -37,13 +37,15 @@
         crate::metrics::dec_connections();
     }
 }
 
 /// Return the current number of active connections.
 #[must_use]
-pub fn active_connection_count() -> u64 { ACTIVE_CONNECTIONS.load(Ordering::Relaxed) }
+pub fn active_connection_count() -> u64 {
+    ACTIVE_CONNECTIONS.load(Ordering::Relaxed)
+}
 
 use crate::{
-    fairness::Fairness,
+    fairness::FairnessTracker,
     hooks::{ConnectionContext, ProtocolHooks},
     push::{FrameLike, PushHandle, PushQueues},
     response::{FrameStream, WireframeError},
@@ -108,7 +110,7 @@
     counter: Option<ActiveConnection>,
     hooks: ProtocolHooks<F, E>,
     ctx: ConnectionContext,
-    fairness: Fairness,
+    fairness: FairnessTracker,
     connection_id: Option<ConnectionId>,
     peer_addr: Option<SocketAddr>,
 }
@@ -166,7 +168,7 @@
             counter: Some(counter),
             hooks,
             ctx,
-            fairness: Fairness::new(FairnessConfig::default()),
+            fairness: FairnessTracker::new(FairnessConfig::default()),
             connection_id: None,
             peer_addr: None,
         };
@@ -179,17 +181,23 @@
         );
         actor.hooks.on_connection_setup(handle, &mut actor.ctx);
         actor
     }
 
     /// Replace the fairness configuration.
-    pub fn set_fairness(&mut self, fairness: FairnessConfig) { self.fairness.set_config(fairness); }
+    pub fn set_fairness(&mut self, fairness: FairnessConfig) {
+        self.fairness.set_config(fairness);
+    }
 
     /// Set or replace the current streaming response.
-    pub fn set_response(&mut self, stream: Option<FrameStream<F, E>>) { self.response = stream; }
+    pub fn set_response(&mut self, stream: Option<FrameStream<F, E>>) {
+        self.response = stream;
+    }
 
     /// Get a clone of the shutdown token used by the actor.
     #[must_use]
-    pub fn shutdown_token(&self) -> CancellationToken { self.shutdown.clone() }
+    pub fn shutdown_token(&self) -> CancellationToken {
+        self.shutdown.clone()
+    }
 
     /// Drive the actor until all sources are exhausted or shutdown is triggered.
     ///
@@ -393,26 +401,28 @@
     fn after_high(&mut self, out: &mut Vec<F>, state: &mut ActorState) {
         self.fairness.after_high();
 
-        if self.fairness.should_yield()
-            && let Some(rx) = &mut self.low_rx
-        {
-            match rx.try_recv() {
-                Ok(mut frame) => {
-                    self.hooks.before_send(&mut frame, &mut self.ctx);
-                    out.push(frame);
-                    self.after_low();
-                }
-                Err(mpsc::error::TryRecvError::Empty) => {}
-                Err(mpsc::error::TryRecvError::Disconnected) => {
-                    self.low_rx = None;
-                    state.mark_closed();
+        if self.fairness.should_yield_to_low_priority() {
+            if let Some(rx) = &mut self.low_rx {
+                match rx.try_recv() {
+                    Ok(mut frame) => {
+                        self.hooks.before_send(&mut frame, &mut self.ctx);
+                        out.push(frame);
+                        self.after_low();
+                    }
+                    Err(mpsc::error::TryRecvError::Empty) => {}
+                    Err(mpsc::error::TryRecvError::Disconnected) => {
+                        self.low_rx = None;
+                        state.mark_closed();
+                    }
                 }
             }
         }
     }
 
     /// Reset counters after processing a low-priority frame.
-    fn after_low(&mut self) { self.fairness.after_low(); }
+    fn after_low(&mut self) {
+        self.fairness.after_low();
+    }
 
     /// Push a frame from the response stream into `out` or handle completion.
     ///
@@ -446,14 +456,18 @@
 
         Ok(())
     }
 
     /// Await cancellation on the provided shutdown token.
     #[inline]
-    async fn wait_shutdown(token: CancellationToken) { token.cancelled_owned().await; }
+    async fn wait_shutdown(token: CancellationToken) {
+        token.cancelled_owned().await;
+    }
 
     /// Receive the next frame from a push queue.
     #[inline]
-    async fn recv_push(rx: &mut mpsc::Receiver<F>) -> Option<F> { rx.recv().await }
+    async fn recv_push(rx: &mut mpsc::Receiver<F>) -> Option<F> {
+        rx.recv().await
+    }
 
     /// Poll `f` if `opt` is `Some`, returning `None` otherwise.
     #[expect(
@@ -533,14 +547,20 @@
         if matches!(self.run_state, RunState::Active) {
             self.run_state = RunState::ShuttingDown;
         }
     }
 
     /// Returns `true` while the actor is actively processing sources.
-    fn is_active(&self) -> bool { matches!(self.run_state, RunState::Active) }
+    fn is_active(&self) -> bool {
+        matches!(self.run_state, RunState::Active)
+    }
 
     /// Returns `true` once shutdown has begun.
-    fn is_shutting_down(&self) -> bool { matches!(self.run_state, RunState::ShuttingDown) }
+    fn is_shutting_down(&self) -> bool {
+        matches!(self.run_state, RunState::ShuttingDown)
+    }
 
     /// Returns `true` when all sources have finished.
-    fn is_done(&self) -> bool { matches!(self.run_state, RunState::Finished) }
+    fn is_done(&self) -> bool {
+        matches!(self.run_state, RunState::Finished)
+    }
 }

src/fairness.rs

@@ -2,54 +2,94 @@
 //!
 //! This module encapsulates the logic for deciding when high-priority
 //! processing should yield to low-priority traffic based on configured
-//! thresholds and optional time slices.
+//! thresholds and optional time slices. A pluggable clock decouples the
+//! implementation from the runtime and allows deterministic tests.
 
-use tokio::time::Instant;
+use std::time::Duration;
 
 use crate::connection::FairnessConfig;
 
+/// Abstraction over a time source returning [`Instant`]s.
+pub(crate) trait Clock {
+    type Instant: Copy;
+    fn now(&self) -> Self::Instant;
+    fn elapsed(&self, start: Self::Instant) -> Duration;
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct TokioClock;
+
+impl Clock for TokioClock {
+    type Instant = tokio::time::Instant;
+    fn now(&self) -> Self::Instant {
+        tokio::time::Instant::now()
+    }
+    fn elapsed(&self, start: Self::Instant) -> Duration {
+        start.elapsed()
+    }
+}
+
 #[derive(Debug)]
-pub(crate) struct Fairness {
+pub(crate) struct FairnessTracker<C: Clock = TokioClock> {
     config: FairnessConfig,
     high_counter: usize,
-    high_start: Option<Instant>,
+    high_start: Option<C::Instant>,
+    clock: C,
 }
 
-impl Fairness {
+impl FairnessTracker<TokioClock> {
     pub(crate) fn new(config: FairnessConfig) -> Self {
+        Self::with_clock(config, TokioClock)
+    }
+}
+
+impl<C: Clock> FairnessTracker<C> {
+    pub(crate) fn with_clock(config: FairnessConfig, clock: C) -> Self {
         Self {
             config,
             high_counter: 0,
             high_start: None,
+            clock,
         }
     }
 
     pub(crate) fn set_config(&mut self, config: FairnessConfig) {
         self.config = config;
-        self.reset();
+        self.clear();
     }
 
     pub(crate) fn after_high(&mut self) {
         self.high_counter += 1;
         if self.high_counter == 1 {
-            self.high_start = Some(Instant::now());
+            self.high_start = Some(self.clock.now());
         }
     }
 
-    pub(crate) fn should_yield(&self) -> bool {
-        let threshold_hit = self.config.max_high_before_low > 0
-            && self.high_counter >= self.config.max_high_before_low;
-        let time_hit = self
-            .config
-            .time_slice
-            .zip(self.high_start)
-            .is_some_and(|(slice, start)| start.elapsed() >= slice);
-        threshold_hit || time_hit
+    pub(crate) fn should_yield_to_low_priority(&self) -> bool {
+        if self.config.max_high_before_low > 0
+            && self.high_counter >= self.config.max_high_before_low
+        {
+            return true;
+        }
+
+        if let (Some(slice), Some(start)) = (self.config.time_slice, self.high_start) {
+            if self.clock.elapsed(start) >= slice {
+                return true;
+            }
+        }
+
+        false
     }
 
-    pub(crate) fn after_low(&mut self) { self.reset(); }
+    pub(crate) fn after_low(&mut self) {
+        self.clear();
+    }
 
     pub(crate) fn reset(&mut self) {
+        self.clear();
+    }
+
+    fn clear(&mut self) {
         self.high_counter = 0;
         self.high_start = None;
     }
@@ -69,11 +109,11 @@
             max_high_before_low: 2,
             time_slice: None,
         };
-        let mut fairness = Fairness::new(cfg);
+        let mut fairness = FairnessTracker::new(cfg);
         fairness.after_high();
-        assert!(!fairness.should_yield());
+        assert!(!fairness.should_yield_to_low_priority());
         fairness.after_high();
-        assert!(fairness.should_yield());
+        assert!(fairness.should_yield_to_low_priority());
     }
 
     #[rstest]
@@ -83,11 +123,11 @@
             max_high_before_low: 1,
             time_slice: None,
         };
-        let mut fairness = Fairness::new(cfg);
+        let mut fairness = FairnessTracker::new(cfg);
         fairness.after_high();
-        assert!(fairness.should_yield());
+        assert!(fairness.should_yield_to_low_priority());
         fairness.after_low();
-        assert!(!fairness.should_yield());
+        assert!(!fairness.should_yield_to_low_priority());
     }
 
     #[rstest]
@@ -98,9 +138,9 @@
             max_high_before_low: 0,
             time_slice: Some(Duration::from_millis(5)),
         };
-        let mut fairness = Fairness::new(cfg);
+        let mut fairness = FairnessTracker::new(cfg);
         fairness.after_high();
         time::advance(Duration::from_millis(6)).await;
-        assert!(fairness.should_yield());
+        assert!(fairness.should_yield_to_low_priority());
     }
 }

src/push.rs

@@ -97,7 +97,9 @@ pub(crate) struct PushHandleInner<F> {
 pub struct PushHandle<F>(Arc<PushHandleInner<F>>);
 
 impl<F: FrameLike> PushHandle<F> {
-    pub(crate) fn from_arc(arc: Arc<PushHandleInner<F>>) -> Self { Self(arc) }
+    pub(crate) fn from_arc(arc: Arc<PushHandleInner<F>>) -> Self {
+        Self(arc)
+    }
 
     /// Internal helper to push a frame with the requested priority.
     ///
@@ -253,7 +255,9 @@ impl<F: FrameLike> PushHandle<F> {
     }
 
     /// Downgrade to a `Weak` reference for storage in a registry.
-    pub(crate) fn downgrade(&self) -> Weak<PushHandleInner<F>> { Arc::downgrade(&self.0) }
+    pub(crate) fn downgrade(&self) -> Weak<PushHandleInner<F>> {
+        Arc::downgrade(&self.0)
+    }
 }
 
 /// Receiver ends of the push queues stored by the connection actor.
@@ -382,10 +386,10 @@ impl<F: FrameLike> PushQueues<F> {
         rate: Option<usize>,
         dlq: Option<mpsc::Sender<F>>,
     ) -> Result<(Self, PushHandle<F>), PushConfigError> {
-        if let Some(r) = rate
-            && (r == 0 || r > MAX_PUSH_RATE)
-        {
-            return Err(PushConfigError::InvalidRate(r));
+        if let Some(r) = rate {
+            if r == 0 || r > MAX_PUSH_RATE {
+                return Err(PushConfigError::InvalidRate(r));
+            }
         }
         let (high_tx, high_rx) = mpsc::channel(high_capacity);
         let (low_tx, low_rx) = mpsc::channel(low_capacity);

src/server.rs

@@ -229,7 +229,9 @@ where
     /// ```
     #[inline]
     #[must_use]
-    pub const fn worker_count(&self) -> usize { self.workers }
+    pub const fn worker_count(&self) -> usize {
+        self.workers
+    }
 
     /// Get the socket address the server is bound to, if available.
     #[must_use]
@@ -507,10 +509,10 @@ async fn process_stream<F, T>(
     let peer_addr = stream.peer_addr().ok();
     match read_preamble::<_, T>(&mut stream).await {
         Ok((preamble, leftover)) => {
-            if let Some(handler) = on_success.as_ref()
-                && let Err(e) = handler(&preamble, &mut stream).await
-            {
-                tracing::error!(error = ?e, ?peer_addr, "preamble callback error");
+            if let Some(handler) = on_success.as_ref() {
+                if let Err(e) = handler(&preamble, &mut stream).await {
+                    tracing::error!(error = ?e, ?peer_addr, "preamble callback error");
+                }
             }
             let stream = RewindStream::new(leftover, stream);
             // Hand the connection to the application for processing.
@@ -558,7 +560,7 @@ mod tests {
 
     /// Test helper preamble carrying no data.
     #[derive(Debug, Clone, PartialEq, Encode, Decode)]
-    #[expect(
+    #[allow(
         dead_code,
         reason = "used only in doctest to illustrate an empty preamble"
     )]