docs(roadmap): record the chunk-2 session candidates — no-io idle sleep gap, cross-thread unpark

The terminal-wake fix cannot reach the (deadline, no-io) idle branch:
thread::sleep, no pipe to write. Documented as a Later item with the
two candidate mechanisms (clamped sleep vs condvar/futex), plus the
cross-thread-unpark observation from the same session.

ROADMAP.md

@@ -113,6 +113,26 @@ shared instead" check. In scope only if the data demands it.
 
 ## Later
 
+### Unwakeable idle sleep when io is absent (terminal-wake residual)
+The `(Some(deadline), None)` idle branch — timers pending, io subsystem never
+initialized — blocks in `thread::sleep` with no wake mechanism at all. The
+terminal wake (writes the wake pipe at AllDone) cannot reach it: no io, no
+pipe. Same stall as the fixed bug, in any no-io runtime: a sibling that
+blocked on an orphaned deadline sleeps it out in full after everything else
+finished. Candidates, mutually exclusive: (a) clamp the sleep (cheap, but
+turns idle into periodic wakeups), or (b) park the branch on a condvar/futex
+the AllDone path signals — and at that point consider making the condvar the
+idle primitive for the no-io runtime generally (a cross-thread unpark could
+signal it too, see below). Decide before any no-io deployment.
+
+### Cross-thread unpark
+`RuntimeInner::enqueue` does not wake idle sibling schedulers — only io
+completions write the wake pipe. Mid-flight this is masked (the enqueuing
+thread is awake and eats the work itself), but it costs parallelism: work
+enqueued by a busy thread waits until the sibling's idle poll times out.
+Candidate from the urus chunk-2 session; needs bench evidence (does the
+shared-queue handoff latency actually show up?) before a mechanism is picked.
+
 ### Per-switch cost (context shims, epoch protocol)
 The shootout's residual: per-wake latency is 0.16–0.18 µs at N=1 and
 0.8–1.2 µs at N=8+, dominated by the context-switch shims and the epoch

src/io.rs

@@ -260,6 +260,15 @@ impl IoThread {
         self.wake_read
     }
 
+    /// Write the wake pipe directly: rouse every scheduler thread blocked in
+    /// its idle `poll_wake`. Used by the terminal (AllDone) path — an idle
+    /// sibling may be blocked on a snapshot that nothing will ever refresh
+    /// (an orphaned timer deadline, or `io_outstanding` from a waiter that
+    /// was stop-cancelled and so never produces a completion).
+    pub fn wake(&self) {
+        wake_scheduler(self.wake_write);
+    }
+
     /// Register interest in `fd` becoming readable/writable; record `pid`
     /// as the parked waiter. The epoll thread will push a `FdReady`
     /// completion when the kernel signals.

src/runtime.rs

@@ -1238,6 +1238,17 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                     // woken by them — e.g. a sleeper cancelled out of its sleep);
                     // they must not keep the runtime alive. Drop them on the way out.
                     inner.timers.lock().unwrap().clear();
+                    // Terminal wake: a sibling scheduler may be blocked in its
+                    // idle wait on a snapshot that is now terminally stale — an
+                    // orphaned long deadline (it would sleep it out in full) or
+                    // a stale `io_outstanding > 0` from a stop-cancelled waiter
+                    // (it would block in poll(-1) forever; cancellation produces
+                    // no completion, so nothing else writes the wake pipe).
+                    // One byte wakes every poller; each re-runs the verdict,
+                    // reaches AllDone itself, and re-wakes — idempotent.
+                    if let Some(io) = inner.io.lock().unwrap().as_ref() {
+                        io.wake();
+                    }
                     return;
                 }
                 Pop::Idle { io_outstanding, wake_fd } => {

src/scheduler.rs

@@ -212,6 +212,18 @@ pub fn yield_now() {
 }
 
 pub fn park_current() {
+    // Entry-side observation point: a stop flagged while we were QUEUED is
+    // otherwise lost — the stop's wildcard unpark no-ops on a Queued actor
+    // (the pending run "is" the wake), so if our first action on resume is
+    // this park, no wake is ever coming and the wake-side check below is
+    // unreachable. Checking here closes that hole; a flag that lands after
+    // this check is covered by the existing protocol (the stop's unpark
+    // finds Running / the prep-to-park window, sets Notified, and the
+    // park-return re-queues us into the wake-side check). Unwinding from
+    // here is the same unwind path as the wake side: leftover wait
+    // registrations are stale-epoch / dead-generation and self-clean at
+    // their wakers' failed CAS.
+    crate::preempt::check_cancelled();
     runtime::set_yield_intent(YieldIntent::Park);
     unsafe { crate::context::switch_to_scheduler() };
     // Observation point on the wakeup side of every blocking primitive

tests/cancel.rs

@@ -130,3 +130,61 @@ fn join_on_stopped_actor_returns_ok() {
         assert!(h.join().is_ok(), "join on a stopped actor returns Ok(())");
     });
 }
+
+/// Regression: `request_stop` against a QUEUED actor must not be lossy.
+///
+/// The stop flag is set, but the wildcard unpark no-ops on a Queued actor
+/// (the pending run "is" the wake). If the actor's first action on resume
+/// is a blocking park — no allocation, no `check!()` on the way — a
+/// wake-side-only check in `park_current` never runs: the actor parks with
+/// the stop flag already set, and nothing will ever wake it. The runtime
+/// then idles forever (the root is parked on the monitor channel).
+///
+/// Fix: an entry-side `check_cancelled` in `park_current`. The remaining
+/// window (flag set after the entry check, before the park lands) is closed
+/// by the existing protocol: the stop's unpark then finds Running /
+/// the prep-to-park window, sets Notified, and the park-return re-queues
+/// into the wake-side check.
+///
+/// Watchdog harness: without the fix this deadlocks, so the runtime runs on
+/// a side thread and the test fails on a timeout instead of hanging cargo.
+#[test]
+fn stop_flagged_while_queued_lands_at_first_park() {
+    use std::sync::mpsc;
+    use std::time::Duration;
+
+    let dropped = Arc::new(AtomicBool::new(false));
+    let saw_stopped = Arc::new(AtomicBool::new(false));
+    let (d, s) = (dropped.clone(), saw_stopped.clone());
+
+    let (done_tx, done_rx) = mpsc::channel::<()>();
+    std::thread::spawn(move || {
+        let rt = smarm::init(smarm::Config::exact(1));
+        rt.run(move || {
+            let h = spawn(move || {
+                let _g = DropFlag(d);
+                let (tx, rx) = channel::<u8>();
+                let _keep = tx; // keep the channel open: recv() parks
+                let _ = rx.recv(); // first observation point is this park
+            });
+            let pid = h.pid();
+            let down = monitor(pid);
+            // Stop while the child is still QUEUED — before it ever runs.
+            // The unpark no-ops; only the flag is left behind.
+            request_stop(pid);
+            let dn = down.rx.recv().expect("monitor channel closed before Down");
+            assert_eq!(dn.pid, pid);
+            if matches!(dn.reason, DownReason::Stopped) {
+                s.store(true, Ordering::SeqCst);
+            }
+            let _ = h.join();
+        });
+        let _ = done_tx.send(());
+    });
+    done_rx
+        .recv_timeout(Duration::from_secs(10))
+        .expect("runtime deadlocked: stop against a QUEUED actor was lost at its first park");
+
+    assert!(saw_stopped.load(Ordering::SeqCst), "expected DownReason::Stopped");
+    assert!(dropped.load(Ordering::SeqCst), "Drop guard must run during the cancellation unwind");
+}

tests/terminal_wake.rs

@@ -0,0 +1,115 @@
+//! Terminal wake: a scheduler thread blocked in the idle wait must be woken
+//! when a sibling reaches the all-done verdict, or `rt.run` stalls in its
+//! worker join.
+//!
+//! Mechanism (runtime.rs, `Pop::Idle`): an idle scheduler snapshots
+//! `peek_deadline()` / `io_outstanding` and blocks in `poll_wake` (or
+//! `thread::sleep`) on that snapshot. `enqueue` does not write the wake
+//! pipe — only IO completions do — so the snapshot can go terminally stale:
+//!
+//! - An actor parked in `wait_readable` that is `request_stop`ped produces
+//!   NO completion (cancellation deregisters the waiter); a sibling that
+//!   blocked on `io_outstanding > 0` with no timers is in `poll(-1)` forever.
+//! - An actor cancelled out of a long `sleep` leaves its timer entry
+//!   orphaned; a sibling that blocked on that deadline sleeps it out in full.
+//!
+//! In both cases the remaining work completes on the *other* scheduler
+//! thread, which hits AllDone and returns — and nothing wakes the blocked
+//! one. `Runtime::run` joins it: a permanent hang in the first case, a
+//! full-deadline stall in the second.
+//!
+//! Both tests force the window deterministically: the root busy-spins
+//! (creating no timer entries and occupying one scheduler thread) so the
+//! other thread settles into the stale idle wait before the stop is issued.
+
+use std::sync::mpsc;
+use std::time::{Duration, Instant};
+
+struct PipePair {
+    read: libc::c_int,
+    write: libc::c_int,
+}
+
+impl PipePair {
+    fn new() -> Self {
+        let mut fds: [libc::c_int; 2] = [0; 2];
+        let r = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
+        assert_eq!(r, 0, "pipe2 failed");
+        PipePair { read: fds[0], write: fds[1] }
+    }
+}
+
+impl Drop for PipePair {
+    fn drop(&mut self) {
+        unsafe {
+            libc::close(self.read);
+            libc::close(self.write);
+        }
+    }
+}
+
+/// Occupy the current scheduler thread without creating timer entries or
+/// parking. A plain spin has no observation points, so the root stays
+/// on-CPU and the sibling scheduler is left alone with the idle branch.
+fn spin_for(d: Duration) {
+    let t0 = Instant::now();
+    while t0.elapsed() < d {
+        std::hint::spin_loop();
+    }
+}
+
+/// Run `body` under a 2-scheduler runtime on a watchdog thread; fail if
+/// `Runtime::run` has not returned within `limit`.
+fn run_with_watchdog(limit: Duration, body: impl FnOnce() + Send + 'static) {
+    let (done_tx, done_rx) = mpsc::channel::<()>();
+    std::thread::spawn(move || {
+        let rt = smarm::init(smarm::Config::exact(2));
+        rt.run(body);
+        let _ = done_tx.send(());
+    });
+    done_rx
+        .recv_timeout(limit)
+        .expect("Runtime::run did not return: idle scheduler thread was never woken at termination");
+}
+
+/// Permanent-hang variant: sibling blocked in `poll_wake(wake_fd, None)`
+/// because `io_outstanding > 0` (one actor parked in `wait_readable`) and no
+/// timers are pending. The waiter is then stop-cancelled — no IO completion
+/// ever writes the wake pipe — and everything else finishes on the root's
+/// thread. Without a terminal wake, `rt.run` never returns.
+#[test]
+fn run_returns_after_io_waiter_is_stop_cancelled() {
+    run_with_watchdog(Duration::from_secs(10), || {
+        let pipe = PipePair::new();
+        let rfd = pipe.read;
+        let h = smarm::spawn(move || {
+            // Never-readable fd (write end open, nothing written).
+            let _ = smarm::wait_readable(rfd);
+        });
+        // Let the waiter park and the sibling scheduler settle into the
+        // io_outstanding>0 / no-timers idle wait: poll(wake_fd, -1).
+        spin_for(Duration::from_millis(200));
+        smarm::request_stop(h.pid());
+        let _ = h.join();
+        drop(pipe);
+    });
+}
+
+/// Finite-stall variant: sibling blocked on an orphaned long timer
+/// deadline. An actor cancelled out of `sleep(60s)` leaves its timer entry
+/// behind (documented as harmless at AllDone — but a sibling that already
+/// blocked on that deadline sleeps it out in full, stalling `rt.run` for
+/// the better part of a minute).
+#[test]
+fn run_returns_after_long_sleeper_is_stop_cancelled() {
+    run_with_watchdog(Duration::from_secs(10), || {
+        let h = smarm::spawn(|| {
+            smarm::sleep(Duration::from_secs(60));
+        });
+        // Let the sleeper park and the sibling scheduler block on the 60s
+        // deadline: poll(wake_fd, ~60_000ms).
+        spin_for(Duration::from_millis(200));
+        smarm::request_stop(h.pid());
+        let _ = h.join();
+    });
+}