fix(runtime): consume wake-pipe bytes only under the drain lock

Lost-wakeup: schedule_loop's phase-1 drain uses drain_lock.try_lock(), and
try_lock losers skip the completion drain entirely. Both schedulers park on
one shared wake pipe and, until now, drained ALL its bytes right after their
idle poll_wake returned — outside the drain lock. A loser could therefore
eat the byte announcing a completion the winner had not seen (the winner was
already past drain_completions when the epoll thread pushed it), and both
threads would park with the completion stranded. Because the bridge eventfd
is registered EPOLLONESHOT, the kernel had already disarmed it at
epoll_wait, so no later write could re-fire it: the runtime slept until an
unrelated timer deadline forced another phase-1 pass.

Fix: drain_wake_pipe() moves inside the drain guard, immediately before
drain_completions(); the two post-poll drains in the Pop::Idle arms are
removed. Producers push their completion before writing the byte, so a byte
consumed under the guard always has its completion visible to the drain that
follows. An unconsumed byte keeps the (level-triggered) idle poll returning
instantly, so a try_lock loser spins briefly until the winner releases —
it can no longer sleep through stranded work.

Found via smarm_beam's ingress-cap drain barrier flaking under CPU load
(5/25 loaded suite runs wedged; mid-wedge stacks showed both schedulers in
poll_wake with an FdReady stranded and the eventfd disarmed). Post-fix:
60/60 loaded runs green, tight 5.8-6.8s timing band, no stall tail.
Root-cause notes: smarm_beam outputs/flake-rootcause-egress-overload.md.

src/io.rs

@@ -498,8 +498,12 @@ fn make_pipe() -> io::Result<(RawFd, RawFd)> {
     Ok((fds[0], fds[1]))
 }
 
-/// Drain pending bytes from the wake pipe. The scheduler calls this after
+/// Drain pending bytes from the wake pipe. Nonblocking (pipe is O_NONBLOCK).
-/// a `poll` wakeup so the next idle call sees an empty pipe.
+///
+/// DISCIPLINE: called only by the phase-1 drain-lock winner, immediately
+/// before `drain_completions`. Bytes are the notification channel for
+/// completions; consuming one anywhere else can strand the completion it
+/// announces (see the lost-wakeup note at the call site in `schedule_loop`).
 pub fn drain_wake_pipe(fd: RawFd) {
     let mut buf = [0u8; 64];
     loop {

src/runtime.rs

@@ -1342,7 +1342,20 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
             let completions = match inner.io.lock() {
                 Ok(mut io) => io
                     .as_mut()
-                    .map(|io| io.drain_completions())
+                    .map(|io| {
+                        // Consume wake-pipe bytes ONLY here, under the drain
+                        // lock and strictly before draining completions.
+                        // Producers push their completion before writing the
+                        // byte, so every byte consumed here has its completion
+                        // visible to the drain below. Consuming bytes anywhere
+                        // else — in particular after an idle poll, outside the
+                        // lock — loses wakeups: a try_lock loser can eat the
+                        // byte for a completion the winner never saw, leaving
+                        // it stranded (and its EPOLLONESHOT fd disarmed) until
+                        // an unrelated timer forces another drain pass.
+                        crate::io::drain_wake_pipe(io.wake_fd());
+                        io.drain_completions()
+                    })
                     .unwrap_or_default(),
                 Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
             };
@@ -1552,16 +1565,22 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                                 let timeout = deadline - now;
                                 match fd_opt {
                                     Some(fd) => {
+                                        // Wake only; the byte (if any) is
+                                        // consumed by the next drain-lock
+                                        // winner in phase 1. Level-triggered
+                                        // poll means an unconsumed byte makes
+                                        // this return immediately, so a loser
+                                        // spins briefly until the winner
+                                        // releases — never sleeps through it.
                                         crate::io::poll_wake(fd, Some(timeout));
-                                        crate::io::drain_wake_pipe(fd);
                                     }
                                     None => thread::sleep(timeout),
                                 }
                             }
                         }
                         (None, Some(fd)) if io_outstanding > 0 => {
+                            // See above: no byte consumption outside phase 1.
                             crate::io::poll_wake(fd, None);
-                            crate::io::drain_wake_pipe(fd);
                         }
                         _ => {
                             thread::sleep(std::time::Duration::from_micros(100));