time: fix resetting expired timers causing panics (#2587)

* Add Unit Test demonstrating the issue

This test demonstrates a panic that occurs when the user inserts an
item with an instant in the past, and then tries to reset the timeout
using the returned key

* Guard reset_at against removals of expired items

Trying to remove an already expired Timer Wheel entry (called by
DelayQueue.reset()) causes panics in some cases as described in (#2573)

This prevents this panic by removing the item from the expired queue and
not the wheel in these cases

Fixes: #2473

tokio/src/time/delay_queue.rs

@@ -427,6 +427,22 @@ impl<T> DelayQueue<T> {
         }
     }
 
+    /// Removes the key fom the expired queue or the timer wheel
+    /// depending on its expiration status
+    ///
+    /// # Panics
+    /// Panics if the key is not contained in the expired queue or the wheel
+    fn remove_key(&mut self, key: &Key) {
+        use crate::time::wheel::Stack;
+
+        // Special case the `expired` queue
+        if self.slab[key.index].expired {
+            self.expired.remove(&key.index, &mut self.slab);
+        } else {
+            self.wheel.remove(&key.index, &mut self.slab);
+        }
+    }
+
     /// Removes the item associated with `key` from the queue.
     ///
     /// There must be an item associated with `key`. The function returns the
@@ -456,15 +472,7 @@ impl<T> DelayQueue<T> {
     /// # }
     /// ```
     pub fn remove(&mut self, key: &Key) -> Expired<T> {
-        use crate::time::wheel::Stack;
-
-        // Special case the `expired` queue
-        if self.slab[key.index].expired {
-            self.expired.remove(&key.index, &mut self.slab);
-        } else {
-            self.wheel.remove(&key.index, &mut self.slab);
-        }
-
+        self.remove_key(key);
         let data = self.slab.remove(key.index);
 
         Expired {
@@ -508,7 +516,7 @@ impl<T> DelayQueue<T> {
     /// # }
     /// ```
     pub fn reset_at(&mut self, key: &Key, when: Instant) {
-        self.wheel.remove(&key.index, &mut self.slab);
+        self.remove_key(key);
 
         // Normalize the deadline. Values cannot be set to expire in the past.
         let when = self.normalize_deadline(when);

tokio/tests/time_delay_queue.rs

@@ -481,6 +481,27 @@ async fn reset_later_after_slot_starts() {
     assert_eq!(entry, "foo");
 }
 
+#[tokio::test]
+async fn reset_inserted_expired() {
+    time::pause();
+    let mut queue = task::spawn(DelayQueue::new());
+    let now = Instant::now();
+
+    let key = queue.insert_at("foo", now - ms(100));
+
+    // this causes the panic described in #2473
+    queue.reset_at(&key, now + ms(100));
+
+    assert_eq!(1, queue.len());
+
+    delay_for(ms(200)).await;
+
+    let entry = assert_ready_ok!(poll!(queue)).into_inner();
+    assert_eq!(entry, "foo");
+
+    assert_eq!(queue.len(), 0);
+}
+
 #[tokio::test]
 async fn reset_earlier_after_slot_starts() {
     time::pause();