refactor(wakes): epoch-stamp every registration-based wake; retire per-primitive wait seqs

The slot epoch is THE wait identity, so the hand-rolled per-primitive copies
go away: channel loses cur_wait/next_wait_seq/timed_out, mutex loses Wait.seq
and next_seq, TimerTarget::on_timeout takes the epoch. Registrations become
(pid, epoch) — channel parked_receiver, mutex waiters, io fd waiters,
Blocking io completions, sleep timers, joiner lists — and their wakers move
to unpark_at. begin_wait is lock-free, so each primitive opens the wait
inside the same critical section that publishes the registration.

recv_timeout's wake-classification loop collapses: wakes are precise, so
queue → Ok, senders==0 → Disconnected, else Timeout — the 'Defensive'
re-register branch is now unreachable by protocol, not by audit. Same for
Mutex::lock_timeout's one-shot park.

End-state invariant, auditable in one sentence: the only wildcard wake is
request_stop, which is terminal.

src/runtime.rs

@@ -17,7 +17,7 @@
 //!  }
 //!
 //!  Slot {
-//!    word:    AtomicU64           ← (generation << 32) | state — THE state machine
+//!    word:    AtomicU64           ← (gen << 32) | (epoch << 8) | state — THE state machine
 //!    sp:      AtomicUsize         ← saved stack pointer
 //!    stop_ptr:AtomicPtr<...>      ← into the actor's Arc<AtomicBool>
 //!    closure: AtomicPtr<...>      ← first-resume closure, swap-to-take
@@ -320,7 +320,9 @@ pub(crate) type Closure = Box<dyn FnOnce() + Send>;
 /// link / finalize), never on the yield/park/unpark hot path.
 pub(crate) struct SlotCold {
     pub(crate) actor: Option<Actor>,
-    pub(crate) waiters: Vec<Pid>,
+    /// Parked joiners as `(pid, park-epoch)`; finalize wakes each via the
+    /// epoch-matched unpark.
+    pub(crate) waiters: Vec<(Pid, u32)>,
     pub(crate) outcome: Option<Outcome>,
     pub(crate) supervisor_channel: Option<Sender<Signal>>,
     /// Watchers registered via `monitor()`, each tagged with its
@@ -543,6 +545,14 @@ impl RuntimeInner {
         self.unpark_inner(pid, Some(epoch));
     }
 
+    /// Open a new wait for `pid` (the calling actor itself): bump its
+    /// park-epoch and return it. See slot_state.rs for the rules.
+    #[must_use]
+    pub(crate) fn begin_wait(&self, pid: Pid) -> u32 {
+        let slot = self.slot_at(pid).expect("begin_wait: own slot vanished");
+        slot.word.begin_wait(pid.generation())
+    }
+
     fn unpark_inner(&self, pid: Pid, want: Option<u32>) {
         if let Some(slot) = self.slot_at(pid) {
             match slot.word.unpark(pid.generation(), want) {
@@ -937,9 +947,9 @@ fn finalize_actor(inner: &Arc<RuntimeInner>, pid: Pid, outcome: Outcome) {
         }
     }
 
-    // Unpark joiners.
-    for joiner in waiters {
-        inner.unpark(joiner);
+    // Unpark joiners (epoch-matched: each registered under the cold lock).
+    for (joiner, epoch) in waiters {
+        inner.unpark_at(joiner, epoch);
     }
 
     // Reclaim if no outstanding handles (re-verified inside).
@@ -990,17 +1000,19 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                     // actor is between `timers.insert_sleep` and
                     // `park_current`; RunningNotified makes the upcoming park
                     // re-queue), or gone (no-op).
-                    crate::timer::Reason::Sleep => inner.unpark(entry.pid),
-                    crate::timer::Reason::WaitTimeout { target, wait_seq } => {
-                        // The callback may call unpark itself.
-                        target.on_timeout(entry.pid, wait_seq);
+                    crate::timer::Reason::Sleep { epoch } => {
+                        inner.unpark_at(entry.pid, epoch)
+                    }
+                    crate::timer::Reason::WaitTimeout { target, epoch } => {
+                        // The callback may call unpark_at itself.
+                        target.on_timeout(entry.pid, epoch);
                     }
                 }
             }
 
             for completion in completions {
                 match completion {
-                    crate::io::Completion::Blocking { pid, result } => {
+                    crate::io::Completion::Blocking { pid, epoch, result } => {
                         if let Some(io) = inner.io.lock().unwrap().as_mut() {
                             io.outstanding = io.outstanding.saturating_sub(1);
                         }
@@ -1018,19 +1030,19 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                                     // flight; discard the result.
                                 }
                             }
-                            inner.unpark(pid);
+                            inner.unpark_at(pid, epoch);
                         }
                     }
                     crate::io::Completion::FdReady { fd, events: _ } => {
                         // Resolve the parked pid under the io lock, then wake
                         // through the protocol. Lock order: io before all.
-                        let parked_pid = inner.io.lock().unwrap().as_mut().and_then(|io| {
-                            let pid = io.waiters.remove(&fd);
+                        let parked = inner.io.lock().unwrap().as_mut().and_then(|io| {
+                            let entry = io.waiters.remove(&fd);
                             io.epoll_deregister(fd);
-                            pid
+                            entry
                         });
-                        if let Some(pid) = parked_pid {
-                            inner.unpark(pid);
+                        if let Some((pid, epoch)) = parked {
+                            inner.unpark_at(pid, epoch);
                         }
                     }
                 }