fix(executor): keep the hot-queue walk alive across nested ticks

compio-executor/src/lib.rs

@@ -212,6 +212,12 @@ impl Executor {
         self.shared().drain_sync(queue);
 
         for id in queue.iter_hot().take(self.config.max_interval as _) {
+            // A nested `tick` (a task's poll driving the same executor) may
+            // have already taken this task off the hot list and processed it;
+            // skip instead of asserting or polling it a second time.
+            if !queue.is_hot(id) {
+                continue;
+            }
             queue.make_cold(id);
             let task = queue.take(id).expect("Task was not reset back");
             let res = unsafe { task.run() };

compio-executor/src/queue.rs

@@ -165,12 +165,32 @@ impl TaskQueue {
         unsafe { self.with_inner(|inner| inner.make_cold(key)) }
     }
 
+    /// Whether `key` currently sits on the hot queue.
+    pub fn is_hot(&self, key: TaskId) -> bool {
+        unsafe { self.with_inner(|inner| inner.map.get(key).is_some_and(|item| item.is_hot)) }
+    }
+
+    /// Returns the hot successor of `key`, or the current hot head when
+    /// `key` is no longer hot.
+    ///
+    /// The iterator over the hot queue holds a cursor across `tick`'s task
+    /// processing, but a task's poll may re-enter `tick` (nested driving,
+    /// e.g. a synchronous harvest inside a thread-per-core runtime) and
+    /// reshuffle the list, moving the cursor's target off the hot queue
+    /// (`make_cold` also clears its `next`). The old
+    /// `debug_assert!(item.is_hot)` panicked on debug builds in that case,
+    /// and `item.next` (already `None`) silently truncated the iteration on
+    /// release builds. Restarting from the current hot head keeps the walk
+    /// alive; tasks the nested tick already processed are simply skipped.
     pub fn next_hot(&self, key: TaskId) -> Option<TaskId> {
         unsafe {
             self.with_inner(|inner| {
                 inner.map.get(key).and_then(|item| {
-                    debug_assert!(item.is_hot);
-                    item.next
+                    if item.is_hot {
+                        item.next
+                    } else {
+                        inner.hot.head
+                    }
                 })
             })
         }

compio-executor/tests/executor.rs

@@ -257,3 +257,59 @@ fn test_join_result_resume_unwind() {
         assert_eq!(*msg, "resume_unwind panic");
     });
 }
+/// Regression: `tick` used to walk the hot queue with a lazily prefetching
+/// iterator. A task re-entering `tick` from inside its poll (nested driving,
+/// e.g. a synchronous harvest on a thread-per-core runtime) could transition
+/// the prefetched task from hot to cold, tripping `debug_assert!(item.is_hot)`
+/// in `next_hot` on debug builds and silently truncating the iteration on
+/// release builds.
+#[test]
+fn test_tick_is_reentrant_safe() {
+    use std::{
+        future::Future,
+        pin::Pin,
+        rc::Rc,
+        task::{Context, Poll},
+    };
+
+    let exe = Rc::new(Executor::new());
+
+    // Task A re-enters the executor from within its poll while B is still
+    // queued: the nested tick transitions B (pending) from hot to cold, which
+    // is exactly the stale state the old prefetched cursor dereferenced on
+    // its next step.
+    struct Reenter(Rc<Executor>);
+
+    impl Future for Reenter {
+        type Output = ();
+
+        fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> {
+            self.0.tick();
+            Poll::Ready(())
+        }
+    }
+
+    // Task B never completes: the nested tick moves it to the cold queue and
+    // resets it there.
+    struct Never;
+
+    impl Future for Never {
+        type Output = ();
+
+        fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> {
+            Poll::Pending
+        }
+    }
+
+    // A first so it holds the hot head and B is the prefetched successor.
+    exe.spawn(Reenter(Rc::clone(&exe))).detach();
+    exe.spawn(Never).detach();
+
+    // With the regression this panics on debug builds via
+    // `debug_assert!(item.is_hot)` in `next_hot`.
+    let still_hot = exe.tick();
+
+    // A completed and the inner tick drained the queue behind it; B rests in
+    // the cold queue, so no hot work remains.
+    assert!(!still_hot);
+}