perf(server): c10k connection-plane wave — memory ratchet fix, loud maxclients, striped registry, deadline-heap blocking (W1-W8+T3)

CHANGELOG.md

@@ -6,6 +6,49 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+- **c10k connection-plane wave (PR #TBD)** — from the 2026-07-29 empirical
+  review (`tmp/C10K-REVIEW.md`: 10k/25k live-connection ramps, idle-CPU
+  measurement, symbolized flames, pipeline-ratchet A/B on the Linux VM):
+  - **Pipeline memory ratchet fixed (W1).** One 1024-deep pipeline
+    permanently ratcheted a connection from 56.5 KB to ~217 KB RSS until
+    disconnect (measured: 5000 such conns pinned 1.09 GB). Batch scratch
+    vecs now shrink after oversized batches, I/O buffer shrink floor
+    lowered 64 KiB → 16 KiB, tokio bump arena capped, subscriber-mode
+    per-message 8 KiB alloc+zero removed.
+  - **maxclients rejection is loud (W3).** Over-cap connections now receive
+    `-ERR max number of clients reached` (Redis parity) instead of a silent
+    EOF on the monoio + TLS paths; the gate runs before per-connection
+    context construction; startup now checks `RLIMIT_NOFILE` against
+    maxclients, raises the soft limit when possible, and warns with the
+    real client ceiling otherwise.
+  - **Blocked-client sweep is deadline-heap driven (W6)** — O(due) instead
+    of walking every blocked waiter at 100 Hz (~2M entry visits/s/shard at
+    10k BLPOP clients); block-forever waiters are never visited.
+  - **SPSC drain starvation fixed (W7).** The cross-shard drain rotates its
+    start consumer, so the 256-message budget no longer lets a hot
+    low-index peer starve high-index peers indefinitely.
+  - **IP-affinity funnel load-gated (W8).** A shard above 2× the mean
+    connection count no longer receives affinity-funneled connections
+    (falls back to round-robin) — previously one migrated/subscribed client
+    pinned all same-IP connections to one shard with no load feedback.
+
+### Changed
+- **Client registry striped 16 ways (W5).** Accepts/closes no longer
+  serialize on one global lock; `CLIENT LIST` locks one stripe at a time
+  (output is no longer insertion-ordered — Redis makes no ordering
+  guarantee); `CLIENT KILL ID` is an O(1) lookup instead of a full scan.
+- **`active_cross_txn` boxed (W2)** — ~2.2 KB of inline transaction
+  SmallVecs no longer ride in every connection's task future;
+  `ConnectionState` is guarded ≤768 B by a regression test.
+- **Dead plumbing deleted (W4):** the zero-registrant `pending_wakers`
+  waker relay (per-conn Rc clone + twice-per-iteration sweep) and the
+  never-adopted duplicate `server/conn_state.rs` module (209 lines).
+- The experimental `MOON_URING=1` tokio bridge now logs a loud startup
+  warning that its connections bypass maxclients and CLIENT LIST/KILL
+  (T3; full accounting integration tracked in
+  `.planning/rfcs/c1m-connection-plane.md`).
+
 ## [0.8.2] — 2026-07-20
 
 ### Fixed

CLAUDE.md

@@ -145,7 +145,7 @@ orb run -m moon-dev bash -c 'sudo apt-get update -qq && sudo apt-get install -y
 - Never hold a lock across `.await` points.
 - Replace `.read().unwrap()` / `.write().unwrap()` with `.read()` / `.write()` (parking_lot doesn't poison).
 - Per-shard locks only — no global locks on the write path.
-- **monoio cross-thread wakers:** `monoio::spawn` creates `!Send` tasks; `Waker::wake()` from another OS thread does NOT reach them. The cross-shard **reply** path therefore has the connection **await a `flume` oneshot directly** (the target shard sends the reply on it after executing). A `pending_wakers: Rc<RefCell<Vec<Waker>>>` relay is still swept each event-loop iteration (pub/sub + backpressure paths thread it), but it is **no longer the reply-wake mechanism** — the old "register your waker, event loop drains it after SPSC" design was retired when test `swf0` disproved its premise (see `event_loop.rs` ~1730). For cross-thread signalling, prefer `flume::bounded(1)` over custom atomic oneshots.
+- **monoio cross-thread wakers:** `monoio::spawn` creates `!Send` tasks; `Waker::wake()` from another OS thread does NOT reach them. The cross-shard **reply** path therefore has the connection **await a `flume` oneshot directly** (the target shard sends the reply on it after executing). The old `pending_wakers` relay ("register your waker, event loop drains it after SPSC") was retired when test `swf0` disproved its premise, and the vestigial plumbing was **deleted entirely** in the c10k wave (it had zero registrants). For cross-thread signalling, prefer `flume::bounded(1)` over custom atomic oneshots.
 
 ### Error Handling
 - All command errors return `Frame::Error(Bytes)` — no `Result` types in dispatch paths.

src/blocking/mod.rs

@@ -66,6 +66,17 @@ pub struct BlockingRegistry {
     waiters: HashMap<(usize, Bytes), VecDeque<WaitEntry>>,
     /// wait_id -> list of (db_index, key) for cross-key cleanup on wakeup/timeout.
     wait_keys: HashMap<u64, Vec<(usize, Bytes)>>,
+    /// Deadline index (c10k W6): min-heap of (deadline, db_index, key) for
+    /// every registered waiter that HAS a deadline. `expire_timed_out` used
+    /// to walk EVERY queue of EVERY blocked waiter at its 100 Hz cadence
+    /// (~2M entry visits/s/shard at 10k blocked clients, tmp/C10K-REVIEW.md
+    /// defect #3); with the heap it touches only queues holding an actually
+    /// -due candidate. Entries are lazily invalidated: a waiter served or
+    /// cancelled before its deadline leaves a stale heap entry that pops to
+    /// a no-op at its original deadline — bounded by registration rate ×
+    /// timeout, the same envelope as the queues themselves. Zero-timeout
+    /// (block-forever) waiters never enter the heap.
+    deadlines: std::collections::BinaryHeap<std::cmp::Reverse<(std::time::Instant, usize, Bytes)>>,
     /// Monotonically increasing wait_id counter (lower 48 bits).
     next_id: u64,
     /// Shard ID encoded in upper 16 bits of wait_id for global uniqueness.
@@ -81,6 +92,7 @@ impl BlockingRegistry {
         BlockingRegistry {
             waiters: HashMap::new(),
             wait_keys: HashMap::new(),
+            deadlines: std::collections::BinaryHeap::new(),
             next_id: 0,
             shard_id,
         }
@@ -100,6 +112,11 @@ impl BlockingRegistry {
         let wait_id = entry.wait_id;
         let queue_key = (db_index, key.clone());
 
+        if let Some(deadline) = entry.deadline {
+            self.deadlines
+                .push(std::cmp::Reverse((deadline, db_index, key)));
+        }
+
         self.waiters
             .entry(queue_key.clone())
             .or_insert_with(VecDeque::new)
@@ -151,30 +168,48 @@ impl BlockingRegistry {
 
     /// Expire all timed-out waiters. Sends None through their reply channels.
     ///
-    /// Two-pass approach: first collect timed-out entries, then clean up.
-    pub fn expire_timed_out(&mut self, now: std::time::Instant) {
-        // Pass 1: collect timed-out (wait_id, reply_tx) pairs
+    /// Heap-driven (c10k W6): pops due deadline-index entries and scans ONLY
+    /// the queues they name; every other blocked waiter is untouched. Runs at
+    /// 100 Hz from the shard timer, so the no-expiry steady state must stay
+    /// O(1). Returns the number of queue entries visited (observability +
+    /// test surface — the old implementation visited every blocked waiter).
+    pub fn expire_timed_out(&mut self, now: std::time::Instant) -> usize {
+        let mut visited = 0usize;
         let mut timed_out: Vec<(u64, crate::runtime::channel::OneshotSender<Option<Frame>>)> =
             Vec::new();
         let mut timed_out_ids: Vec<u64> = Vec::new();
 
-        for queue in self.waiters.values_mut() {
+        while self
+            .deadlines
+            .peek()
+            .is_some_and(|std::cmp::Reverse((d, _, _))| *d <= now)
+        {
+            #[allow(clippy::unwrap_used)] // peek above just proved non-empty
+            let std::cmp::Reverse((_, db_index, key)) = self.deadlines.pop().unwrap();
+            let queue_key = (db_index, key);
+            // A missing queue is a STALE heap entry (waiter served/cancelled
+            // before its deadline) — the lazy-invalidation no-op.
+            let Some(queue) = self.waiters.get_mut(&queue_key) else {
+                continue;
+            };
             let mut i = 0;
             while i < queue.len() {
+                visited += 1;
                 let is_expired = queue[i].deadline.map_or(false, |d| d <= now);
                 if is_expired {
+                    #[allow(clippy::unwrap_used)] // i < queue.len() by loop guard
                     let entry = queue.remove(i).unwrap();
                     timed_out_ids.push(entry.wait_id);
                     timed_out.push((entry.wait_id, entry.reply_tx));
                 } else {
                     i += 1;
                 }
             }
+            if queue.is_empty() {
+                self.waiters.remove(&queue_key);
+            }
         }
 
-        // Remove empty queues
-        self.waiters.retain(|_, q| !q.is_empty());
-
         // Send None (timeout) to all timed-out waiters
         for (_wait_id, reply_tx) in timed_out {
             let _ = reply_tx.send(None);
@@ -187,6 +222,7 @@ impl BlockingRegistry {
         for id in timed_out_ids {
             self.wait_keys.remove(&id);
         }
+        visited
     }
 }
 
@@ -303,3 +339,111 @@ mod tests {
         assert!(!reg.has_waiters(0, &Bytes::from_static(b"nokey")));
     }
 }
+
+#[cfg(test)]
+mod deadline_heap_tests {
+    use super::*;
+    use std::time::{Duration, Instant};
+
+    fn entry(reg: &mut BlockingRegistry, deadline: Option<Instant>) -> (u64, WaitEntry) {
+        let id = reg.next_wait_id();
+        let (tx, _rx) = crate::runtime::channel::oneshot();
+        (
+            id,
+            WaitEntry {
+                wait_id: id,
+                cmd: BlockedCommand::BLPop,
+                reply_tx: tx,
+                deadline,
+            },
+        )
+    }
+
+    /// c10k W6: the 100 Hz sweep must not touch block-forever waiters. The
+    /// old implementation walked every queue of every blocked client.
+    #[test]
+    fn sweep_skips_undeadlined_waiters() {
+        let mut reg = BlockingRegistry::new(0);
+        let now = Instant::now();
+        for i in 0..100u32 {
+            let (_, e) = entry(&mut reg, None);
+            reg.register(0, Bytes::from(format!("k{i}")), e);
+        }
+        let (_, due) = entry(&mut reg, Some(now - Duration::from_millis(1)));
+        reg.register(0, Bytes::from_static(b"due-key"), due);
+
+        let visited = reg.expire_timed_out(now);
+        assert_eq!(visited, 1, "only the due queue's single entry is visited");
+        assert!(!reg.has_waiters(0, &Bytes::from_static(b"due-key")));
+        assert!(reg.has_waiters(0, &Bytes::from_static(b"k0")));
+        assert!(reg.has_waiters(0, &Bytes::from_static(b"k99")));
+
+        // Steady state after the sweep: nothing due, zero visits.
+        assert_eq!(reg.expire_timed_out(now), 0);
+    }
+
+    /// Shared key: expire only due entries, preserve FIFO of the rest.
+    #[test]
+    fn shared_key_partial_expiry_preserves_fifo() {
+        let mut reg = BlockingRegistry::new(0);
+        let now = Instant::now();
+        let key = Bytes::from_static(b"shared");
+        let (_, e1) = entry(&mut reg, Some(now - Duration::from_millis(5)));
+        reg.register(0, key.clone(), e1);
+        let (id2, e2) = entry(&mut reg, Some(now + Duration::from_secs(60)));
+        reg.register(0, key.clone(), e2);
+        let (id3, e3) = entry(&mut reg, None);
+        reg.register(0, key.clone(), e3);
+
+        reg.expire_timed_out(now);
+        assert_eq!(reg.pop_front(0, &key).unwrap().wait_id, id2);
+        assert_eq!(reg.pop_front(0, &key).unwrap().wait_id, id3);
+        assert!(reg.pop_front(0, &key).is_none());
+    }
+
+    /// A waiter served before its deadline leaves a stale heap entry — the
+    /// sweep at its original deadline must be a no-op, not a panic or a
+    /// spurious timeout reply.
+    #[test]
+    fn served_waiter_stale_heap_entry_is_noop() {
+        let mut reg = BlockingRegistry::new(0);
+        let now = Instant::now();
+        let key = Bytes::from_static(b"served");
+        let (id, e) = entry(&mut reg, Some(now + Duration::from_millis(10)));
+        reg.register(0, key.clone(), e);
+
+        let popped = reg.pop_front(0, &key).expect("served");
+        assert_eq!(popped.wait_id, id);
+        reg.remove_wait(id);
+
+        let visited = reg.expire_timed_out(now + Duration::from_secs(1));
+        assert_eq!(visited, 0, "stale heap entry must no-op");
+    }
+
+    /// Multi-key waiter (BLPOP k1 k2) with one shared deadline: both queue
+    /// entries removed in one sweep, one visit each.
+    #[test]
+    fn multikey_waiter_expires_from_all_queues() {
+        let mut reg = BlockingRegistry::new(0);
+        let now = Instant::now();
+        let id = reg.next_wait_id();
+        let deadline = Some(now - Duration::from_millis(1));
+        for k in [&b"mk1"[..], &b"mk2"[..]] {
+            let (tx, _rx) = crate::runtime::channel::oneshot();
+            reg.register(
+                0,
+                Bytes::copy_from_slice(k),
+                WaitEntry {
+                    wait_id: id,
+                    cmd: BlockedCommand::BLPop,
+                    reply_tx: tx,
+                    deadline,
+                },
+            );
+        }
+        let visited = reg.expire_timed_out(now);
+        assert_eq!(visited, 2);
+        assert!(!reg.has_waiters(0, &Bytes::from_static(b"mk1")));
+        assert!(!reg.has_waiters(0, &Bytes::from_static(b"mk2")));
+    }
+}