feat(park): fenced producer fast path + earliest-deadline snapshot

Two integration-driven amendments ahead of the runtime swap:

wake_one_if_idle() realizes RFC 018's "empty-mask fast path is one
relaxed load" soundly: a bare relaxed load is a lost-wake in the Dekker
shape for the lock-free ring queues, so the producer publishes work,
fences (SeqCst), then reads the mask Relaxed — paired with a matching
fence between the consumer's bit-publish and its re-check in park().
The pure-compute hot path (mask 0) never takes the shared mask line
exclusive; the RMW read stays on the rare chain-rule path only. Loom
models 1/2 now drive the fenced pattern end to end.

next_deadline is the earliest KNOWN timer deadline, independent of
whether anyone is parked — which tk_armed cannot give: under saturation
nobody parks, nobody arms, yet due timers must still fire (ratified
design point (a): the busy-path due-check). Maintained under the timers
mutex (note_deadline on insert — which also carries the timekeeper
re-arm wake — refresh_deadline after pop/clear); read lock-free.
deadline_due() costs one Relaxed load and a branch when no timer exists;
the clock is read only when one does.

src/park.rs

@@ -22,18 +22,20 @@
 //!   producer that published work before observing our bit has left us
 //!   work the re-check finds; one that observes the bit wakes us.
 //!
-//!   The publish/re-check pair is a store-buffer (Dekker) shape, and it is
+//!   The publish/re-check pair is a store-buffer (Dekker) shape. Two sound
-//!   resolved by the **same-location-RMW handshake**, not by `SeqCst`
+//!   resolutions coexist here, chosen per call-site cost profile: the
-//!   loads: every mask read on the producer side ([`Coordinator::wake_one`],
+//!   **fence handshake** on the hot producer path
-//!   [`Coordinator::idle_mask`]) is an atomic RMW (`fetch_or(0)`). An RMW
+//!   ([`Coordinator::wake_one_if_idle`]: publish work; `fence(SeqCst)`;
-//!   always reads the *latest* value in the mask's modification order, so
+//!   one *Relaxed* mask load — pairing with the consumer's `fetch_or(bit)`;
-//!   a producer can never miss a bit that a consumer's `fetch_or(bit)`
+//!   `fence(SeqCst)`; re-check inside [`Coordinator::park`]), so the
-//!   has already written; conversely, a consumer whose `fetch_or(bit)`
+//!   pure-compute hot path (everyone busy, mask 0) never takes the shared
-//!   lands *after* the producer's release-RMW reads-from it, acquires the
+//!   mask line exclusive — the RFC's "one relaxed load" fast path, made
-//!   producer's prior work publication, and its re-check finds the work.
+//!   sound; and the **same-location-RMW read** (`fetch_or(0)`, in
-//!   Both directions hold under plain acquire/release — the loom models
+//!   [`Coordinator::wake_one`] / [`Coordinator::idle_mask`]) for the rare
-//!   verify exactly this (loom models `SeqCst` as `AcqRel`, so a
+//!   paths (chain rule — at most one per wake) where reading the latest
-//!   plain-load design would — and did — fail model 1/2 with a lost wake).
+//!   mask by modification-order coherence is worth an RMW. The loom models
+//!   drive the fence pattern end to end (a fence-less plain-load draft
+//!   would — and did — fail model 1/2 with a lost wake, as it must).
 //! - **Timekeeper**: at most one parked scheduler holds the timer
 //!   deadline (RFC 018 §timers) so a timer expiry wakes one scheduler,
 //!   not a herd. The role is an atomic `(holder id, armed deadline)`
@@ -53,7 +55,7 @@
 //! Standalone until the runtime swap (RFC 018 commit 2): nothing outside
 //! tests constructs a [`Coordinator`] yet.
 
-use crate::sync_shim::{AtomicU64, Ordering};
+use crate::sync_shim::{fence, AtomicU64, Ordering};
 use std::time::Instant;
 
 /// Sentinel for "no timekeeper" in the holder word.
@@ -286,9 +288,17 @@ pub(crate) struct Coordinator {
     /// caller's timer serialization (arm/disarm/insert-check); read
     /// lock-free by the insert wake path.
     tk_holder: AtomicU64,
-    /// Armed deadline as nanos since `origin`, or `NO_DEADLINE`. The
+    /// Armed deadline as nanos since `origin`, or `NO_DEADLINE`. Written
-    /// busy-path due-check reads this Relaxed.
+    /// only by the timekeeper arm/disarm protocol.
     tk_armed: AtomicU64,
+    /// Earliest KNOWN timer deadline (nanos since `origin`), or
+    /// `NO_DEADLINE` — independent of whether any scheduler is parked,
+    /// which is what the timekeeper's `tk_armed` cannot give: under
+    /// saturation nobody parks and nobody arms, yet due timers must still
+    /// fire (ratified design point (a)). Maintained under the caller's
+    /// timers mutex (`note_deadline` on insert, `refresh_deadline` after a
+    /// pop/peek); read lock-free by the busy-path due-check.
+    next_deadline: AtomicU64,
     /// Time origin for the nanos encoding.
     origin: Instant,
 }
@@ -304,6 +314,7 @@ impl Coordinator {
             idle: AtomicU64::new(0),
             tk_holder: AtomicU64::new(NO_TIMEKEEPER),
             tk_armed: AtomicU64::new(NO_DEADLINE),
+            next_deadline: AtomicU64::new(NO_DEADLINE),
             origin: Instant::now(),
         }
     }
@@ -342,6 +353,11 @@ impl Coordinator {
         // visible to the re-check below (see module docs).
         let prev = self.idle.fetch_or(bit, Ordering::AcqRel);
         debug_assert_eq!(prev & bit, 0, "park: idle bit already set for this id");
+        // Fence half of the producer handshake (see `wake_one_if_idle`):
+        // orders our bit-publish before the re-check's loads, so it pairs
+        // with the producer's publish→fence→mask-load — at least one side
+        // must see the other's store, whichever queue backend is in play.
+        fence(Ordering::SeqCst);
         // (2) the mandatory post-publish re-check.
         if recheck() {
             self.idle.fetch_and(!bit, Ordering::AcqRel);
@@ -394,6 +410,23 @@ impl Coordinator {
         }
     }
 
+    /// The producer-side wake tail (`enqueue`'s fast path, RFC 018 "enqueue
+    /// wakes"). The caller has just published work (queue push); we fence,
+    /// then read the mask with ONE Relaxed load — 0 means every scheduler
+    /// is awake and the pure-compute hot path pays no RMW on the shared
+    /// mask line. Soundness is the fence handshake (module docs): our
+    /// fence orders the caller's push before the mask load; the consumer's
+    /// fence (in `park`) orders its bit-publish before its re-check — at
+    /// least one side must observe the other's store, so a consumer we
+    /// miss here is a consumer whose re-check finds the caller's work.
+    pub(crate) fn wake_one_if_idle(&self) -> bool {
+        fence(Ordering::SeqCst);
+        if self.idle.load(Ordering::Relaxed) == 0 {
+            return false;
+        }
+        self.wake_one()
+    }
+
     /// Terminal wake (replaces the AllDone wake-pipe byte): clear the mask
     /// and deliver a permit to *every* parker, parked or not. A permit set
     /// on a busy scheduler costs one spurious park return — nothing at the
@@ -470,11 +503,53 @@ impl Coordinator {
     }
 
     /// The armed-deadline snapshot (nanos since origin; `NO_DEADLINE` =
-    /// none). One Relaxed load — the busy-path due-check (RFC 018 design
+    /// none). One Relaxed load.
-    /// point (a)) reads this every scheduler-loop iteration.
     pub(crate) fn armed_deadline_nanos(&self) -> u64 {
         self.tk_armed.load(Ordering::Relaxed)
     }
+
+    // ----- earliest-deadline snapshot (busy-path due-check) -----
+
+    /// Record a newly inserted timer deadline. MUST be called under the
+    /// timers mutex (same serialization rule as `try_arm_timer`), which is
+    /// why plain compare+store suffices for the min-maintenance. Also runs
+    /// the timekeeper re-arm check (`timer_inserted`) — one call site for
+    /// both consequences of an insert.
+    pub(crate) fn note_deadline(&self, deadline: Instant) {
+        let n = self.deadline_nanos(deadline);
+        if n < self.next_deadline.load(Ordering::Relaxed) {
+            self.next_deadline.store(n, Ordering::Release);
+        }
+        self.timer_inserted(deadline);
+    }
+
+    /// Re-anchor the snapshot to the heap minimum (`None` = heap empty)
+    /// after a `pop_due` / `clear`. MUST be called under the timers mutex.
+    pub(crate) fn refresh_deadline(&self, next: Option<Instant>) {
+        let n = next.map_or(NO_DEADLINE, |d| self.deadline_nanos(d));
+        self.next_deadline.store(n, Ordering::Release);
+    }
+
+    /// Busy-path due-check: is the earliest known deadline at or past
+    /// `now`? One Relaxed load when no deadline is armed — the clock is
+    /// read only when a timer actually exists (matching the old drain
+    /// phase's is_empty guard), so the pure-compute hot path pays a load
+    /// and a branch.
+    pub(crate) fn deadline_due(&self) -> bool {
+        let n = self.next_deadline.load(Ordering::Relaxed);
+        n != NO_DEADLINE && self.deadline_nanos(Instant::now()) >= n
+    }
+
+    /// The earliest-deadline snapshot as an `Instant`, for the idle path's
+    /// park timeout (`None` = no timer pending).
+    pub(crate) fn next_deadline_instant(&self) -> Option<Instant> {
+        let n = self.next_deadline.load(Ordering::Acquire);
+        if n == NO_DEADLINE {
+            None
+        } else {
+            self.origin.checked_add(std::time::Duration::from_nanos(n))
+        }
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -662,6 +737,63 @@ mod tests {
     fn more_than_64_schedulers_asserts() {
         let _ = Coordinator::new(65);
     }
+
+    #[test]
+    fn wake_one_if_idle_noop_on_empty_and_wakes_on_parked() {
+        let c = Arc::new(Coordinator::new(1));
+        assert!(!c.wake_one_if_idle(), "empty mask must be a no-op");
+        let c2 = c.clone();
+        let t = std::thread::spawn(move || {
+            assert_eq!(c2.park(0, None, || false), ParkResult::Woken);
+        });
+        let t0 = Instant::now();
+        while c.idle_mask() == 0 {
+            assert!(t0.elapsed() < Duration::from_secs(5), "never parked");
+            std::thread::yield_now();
+        }
+        assert!(c.wake_one_if_idle());
+        match t.join() {
+            Ok(()) => {}
+            Err(p) => std::panic::resume_unwind(p),
+        }
+    }
+
+    #[test]
+    fn deadline_snapshot_min_maintenance_and_due_check() {
+        let c = Coordinator::new(1);
+        assert!(!c.deadline_due(), "no deadline: never due");
+        assert_eq!(c.next_deadline_instant(), None);
+        let far = Instant::now() + Duration::from_secs(60);
+        let near = Instant::now() + Duration::from_millis(1);
+        c.note_deadline(far);
+        assert!(!c.deadline_due());
+        c.note_deadline(near); // min wins
+        assert!(c.next_deadline_instant().is_some_and(|d| d <= near));
+        c.note_deadline(far); // later insert must NOT raise the snapshot
+        assert!(c.next_deadline_instant().is_some_and(|d| d <= near));
+        std::thread::sleep(Duration::from_millis(2));
+        assert!(c.deadline_due(), "past deadline not reported due");
+        c.refresh_deadline(Some(far));
+        assert!(!c.deadline_due(), "refresh did not re-anchor");
+        c.refresh_deadline(None);
+        assert_eq!(c.next_deadline_instant(), None);
+        // A deadline at/before origin encodes as 0: always due.
+        c.note_deadline(Instant::now() - Duration::from_secs(1));
+        assert!(c.deadline_due());
+    }
+
+    #[test]
+    fn note_deadline_earlier_wakes_timekeeper_via_snapshot_path() {
+        // note_deadline must carry the timer_inserted re-arm wake too.
+        let c = Coordinator::new(2);
+        let far = Instant::now() + Duration::from_secs(60);
+        let near = Instant::now() + Duration::from_millis(1);
+        assert!(c.try_arm_timer(0, far));
+        c.note_deadline(near);
+        let r = c.park(0, Some(far), || false);
+        assert_eq!(r, ParkResult::Woken, "re-arm wake lost through note_deadline");
+        c.disarm_timer(0);
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -691,8 +823,10 @@ mod loom_tests {
                 let c = c.clone();
                 let item = item.clone();
                 thread::spawn(move || {
+                    // The enqueue shape: publish, then the fenced fast-path
+                    // wake tail (this is what the runtime's enqueue calls).
                     item.store(1, O::SeqCst);
-                    c.wake_one();
+                    c.wake_one_if_idle();
                 })
             };
 
@@ -753,9 +887,10 @@ mod loom_tests {
                 }));
             }
 
-            // Producer (main): two items, ONE wake.
+            // Producer (main): two items, ONE wake, via the enqueue-shaped
+            // fenced fast path.
             items.store(2, O::SeqCst);
-            c.wake_one();
+            c.wake_one_if_idle();
 
             for h in hs {
                 h.join().unwrap();

src/sync_shim.rs

@@ -6,10 +6,10 @@
 //! Build the loom models with: `RUSTFLAGS="--cfg loom" cargo test --lib --release`
 
 #[cfg(loom)]
-pub(crate) use loom::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+pub(crate) use loom::sync::atomic::{fence, AtomicU64, AtomicUsize, Ordering};
 
 #[cfg(not(loom))]
-pub(crate) use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+pub(crate) use std::sync::atomic::{fence, AtomicU64, AtomicUsize, Ordering};
 
 // park.rs condvar-parker (loom + non-Linux builds only; the Linux non-loom
 // build parks on a futex and never touches these — gating them identically