time: wake DelayQueue after resetting to expired (#8274)

Resetting an item into the expired stack bypassed the existing Sleep reset
path, leaving a pending consumer asleep until an unrelated deadline.

tokio-util/src/time/delay_queue.rs

@@ -864,11 +864,19 @@ impl<T> DelayQueue<T> {
         self.slab[*key].expired = false;
 
         self.insert_idx(when, *key);
+        let inserted_expired = self.slab[*key].expired;
 
         let next_deadline = self.next_deadline();
-        if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) {
-            // This should awaken us if necessary (ie, if already expired)
-            delay.as_mut().reset(deadline);
+        match (next_deadline, &mut self.delay) {
+            (None, _) => self.delay = None,
+            (Some(deadline), Some(delay)) => delay.as_mut().reset(deadline),
+            (Some(deadline), None) => self.delay = Some(Box::pin(sleep_until(deadline))),
+        }
+
+        if inserted_expired {
+            if let Some(waker) = self.waker.take() {
+                waker.wake();
+            }
         }
     }
 

tokio-util/tests/time_delay_queue.rs

@@ -202,6 +202,27 @@ async fn reset_entry() {
     assert!(entry.is_none())
 }
 
+#[tokio::test]
+async fn reset_to_past_wakes_pending_queue() {
+    time::pause();
+
+    let mut queue = task::spawn(DelayQueue::new());
+    let key = queue.insert("foo", ms(10_000));
+
+    assert_pending!(poll!(queue));
+    assert!(!queue.is_woken());
+
+    queue.reset_at(&key, Instant::now() - ms(100));
+
+    assert!(queue.is_woken());
+
+    let entry = assert_ready_some!(poll!(queue));
+    assert_eq!(*entry.get_ref(), "foo");
+
+    let entry = assert_ready!(poll!(queue));
+    assert!(entry.is_none());
+}
+
 // Reproduces tokio-rs/tokio#849.
 #[tokio::test]
 async fn reset_much_later() {