fix(quinn): handle overdue timers without polling the async timer

drive_timer() used AsyncTimer::poll() to determine whether a protocol
deadline had elapsed. Under Tokio's cooperative task budget,
Sleep::poll() may return Poll::Pending for an already-expired deadline
once the task's budget is exhausted, which can happen when
process_conn_events() drains a busy channel.

As a result, handle_timeout() is not called even though the deadline has
already elapsed. For QUIC, timers such as PTO, loss detection, and idle
timeouts are correctness-critical and should not be deferred to a later
scheduling round.

Fix this by checking runtime.now() >= deadline before consulting the
async timer. The clock is not subject to cooperative budgeting. The
timer remains responsible only for registering a wakeup when the
deadline lies in the future.

Author: syszery

quinn/src/connection.rs

@@ -1157,51 +1157,44 @@ impl State {
         }
     }
 
-    fn drive_timer(&mut self, cx: &mut Context) -> bool {
-        // Check whether we need to (re)set the timer. If so, we must poll again to ensure the
-        // timer is registered with the runtime (and check whether it's already
-        // expired).
-        match self.inner.poll_timeout() {
-            Some(deadline) => {
-                if let Some(delay) = &mut self.timer {
-                    // There is no need to reset the tokio timer if the deadline
-                    // did not change
-                    if self
-                        .timer_deadline
-                        .map(|current_deadline| current_deadline != deadline)
-                        .unwrap_or(true)
-                    {
-                        delay.as_mut().reset(deadline);
-                    }
-                } else {
-                    self.timer = Some(self.runtime.new_timer(deadline));
-                }
-                // Store the actual expiration time of the timer
-                self.timer_deadline = Some(deadline);
+    fn drive_timer(&mut self, cx: &mut Context<'_>) -> bool {
+        let Some(deadline) = self.inner.poll_timeout() else {
+            self.timer_deadline = None;
+            return false;
+        };
+
+        // Use the clock rather than the async timer to detect expiry: Sleep::poll
+        // respects Tokio's cooperative budget and can return Pending for elapsed
+        // deadlines.
+        let now = self.runtime.now();
+        if now >= deadline {
+            self.inner.handle_timeout(now);
+            self.timer_deadline = None;
+            return true;
+        }
+
+        match &mut self.timer {
+            // Avoid resetting the timer when the deadline is unchanged.
+            Some(delay) if self.timer_deadline != Some(deadline) => {
+                delay.as_mut().reset(deadline);
             }
             None => {
-                self.timer_deadline = None;
-                return false;
+                self.timer = Some(self.runtime.new_timer(deadline));
             }
+            _ => {}
         }
-
-        if self.timer_deadline.is_none() {
-            return false;
-        }
+        self.timer_deadline = Some(deadline);
 
         let delay = self
             .timer
             .as_mut()
             .expect("timer must exist in this state")
             .as_mut();
         if delay.poll(cx).is_pending() {
-            // Since there wasn't a timeout event, there is nothing new
-            // for the connection to do
             return false;
         }
 
-        // A timer expired, so the caller needs to check for
-        // new transmits, which might cause new timers to be set.
+        // The deadline elapsed in the window between the clock check and poll.
         self.inner.handle_timeout(self.runtime.now());
         self.timer_deadline = None;
         true