Fix Controller: pending messages are stuck in the scheduled map

kube-runtime/src/controller/runner.rs

@@ -444,6 +444,18 @@ mod tests {
         advance(Duration::from_secs(3)).await;
         assert!(poll!(runner.as_mut()).is_pending());
 
+        // Send the third message again and check it's ran
+        sched_tx
+            .send(ScheduleRequest {
+                message: 3,
+                run_at: Instant::now(),
+            })
+            .await
+            .unwrap();
+        advance(Duration::from_secs(3)).await;
+        assert!(poll!(runner.as_mut()).is_pending());
+        assert_eq!(*count.lock().unwrap(), 4);
+
         let (mut sched_tx, sched_rx) = mpsc::unbounded();
         let mut runner = Box::pin(
             Runner::new(scheduler(sched_rx), 1, |_| {

kube-runtime/src/scheduler.rs

@@ -131,6 +131,9 @@ impl<'a, T: Hash + Eq + Clone, R> SchedulerProj<'a, T, R> {
     pub fn pop_queue_message_into_pending(&mut self, cx: &mut Context<'_>) {
         while let Poll::Ready(Some(msg)) = self.queue.poll_expired(cx) {
             let msg = msg.into_inner();
+            self.scheduled.remove_entry(&msg).expect(
+                "Expired message was popped from the Scheduler queue, but was not in the metadata map",
+            );
             self.pending.insert(msg);
         }
     }