fix(server): accept-loop HOL block, migration fd ownership, park auth gates (#438 F3–F6 + conn-secondary)

CHANGELOG.md

@@ -25,6 +25,51 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `--experimental-per-shard-rewrite` is deprecated (warns, no-op).
 
 ### Fixed
+- **Central accept loop no longer head-of-line blocks on one wedged shard
+  (#438 F3).** Every central-listener delivery (tokio plain/TLS, monoio
+  plain/TLS — the monoio sends were *synchronous*, stalling the whole
+  listener thread) previously blocked on the routed shard's bounded conn
+  channel; one shard wedged at its 4096-conn cap froze accepts for every
+  other shard. Deliveries now `try_send` and rotate to the next shard with
+  room; only when every shard's channel is full does the loop fall back to
+  the blocking send (server-wide saturation, where back-pressure is
+  correct).
+- **Connection-migration fds can no longer leak (#438 F4).**
+  `MigrateConnectionPayload` carried the socket as a raw `i32` with no drop
+  semantics: a migration message still queued when a shard shut down — or
+  drained but never spawned — leaked the fd and stranded the client on a
+  connection no task would ever serve. The payload now owns the socket as an
+  `OwnedFd` end to end (producer → SPSC ring → pending-migrations queue →
+  target-shard spawn), so every undelivered path closes the socket and the
+  client sees a FIN. Three `unsafe from_raw_fd` blocks became safe
+  ownership conversions in the process. (Resumed-parked connections remain
+  deliberately `can_migrate:false`; rationale documented at the spawn site.)
+- **Migrated connections now carry the real `requirepass` (#438 F5, sec
+  L3).** The target-shard `ConnectionContext` was built with
+  `requirepass: None`; the session's auth state was unaffected (it travels
+  in `MigratedConnectionState`), but a later `AUTH` on a migrated
+  connection wrongly answered "no password is set", and any future code
+  deriving auth from the context would have failed open.
+- **Unauthenticated connections never task-park (#438 F6, sec L2).** On an
+  auth-enabled server, a client could open sockets and never authenticate
+  nor speak; each one downshifted and task-parked into the ~3.3 KB watcher
+  state, letting an attacker hold a maxclients-worth of silent connections
+  indefinitely at near-zero cost. Pre-AUTH connections now stay un-parked
+  (full handler task — visible in monitoring, still reaped by `timeout N`);
+  no-auth servers are unaffected.
+- **Idle-park cancel provenance + registry counter hygiene (#438
+  conn-secondary).** A bare `ECANCELED` (errno 125) from any non-sweep
+  source was indistinguishable from the idle sweep's cancel and re-parked
+  the connection — for a dead fd that is a permanent park→wake→park spin at
+  100% CPU. Park arms now require the sweep/drain to have marked the slot
+  (`was_swept_cancel`, consume-once) before treating 125 as a park signal.
+  Separately, a re-register over a still-present client id double-counted
+  `TOTAL_CLIENTS`/shard gauges permanently (skewing `shard_overloaded`
+  routing); `register` now balances against the replaced entry and the
+  `kept_registration` miss-arm logs the invariant violation loudly. (The
+  third secondary finding — `timeout` read once at connection setup — was
+  already fixed by the D1 chore-sweep rework, which re-reads the config
+  every second.)
 - **Graceful shutdown now drains connections (#438 F1, conn#8).** On
   SIGTERM/SIGINT each shard previously tore its runtime down the moment its
   event loop observed the cancellation, dropping every pending connection

src/client_registry.rs

@@ -312,8 +312,25 @@ pub fn register(
         shard,
         live: Arc::clone(&live),
     };
-    stripe(id).write().insert(id, entry);
-    TOTAL_CLIENTS.fetch_add(1, Ordering::Relaxed);
+    // #438 conn-secondary: a re-register over a still-present id (the
+    // `kept_registration` miss-arm fail-safe firing while the entry somehow
+    // survived, or any future double-register bug) must not double-count —
+    // the single eventual deregister would then leave TOTAL_CLIENTS and the
+    // shard gauges permanently high, skewing `shard_overloaded` routing.
+    // Balance against whatever the insert replaced.
+    let replaced = stripe(id).write().insert(id, entry);
+    if let Some(old) = replaced {
+        tracing::warn!(
+            "client registry: id {} re-registered while present (shard {} -> {})",
+            id,
+            old.shard,
+            shard
+        );
+        shard_conns_delta(old.shard, false);
+        crate::admin::metrics_setup::record_shard_connection_delta(old.shard, -1.0);
+    } else {
+        TOTAL_CLIENTS.fetch_add(1, Ordering::Relaxed);
+    }
     shard_conns_delta(shard, true);
     crate::admin::metrics_setup::record_shard_connection_delta(shard, 1.0);
     live
@@ -1282,4 +1299,40 @@ mod idle_timeout_tests {
         parked_delta(false);
         assert_eq!(parked_clients(), before);
     }
+
+    /// #438 conn-secondary: a re-register of a still-present id (the
+    /// kept_registration miss-arm fail-safe racing an entry back into
+    /// existence) must NOT increment TOTAL_CLIENTS again — the single
+    /// eventual deregister would then leave the count permanently +1,
+    /// skewing `shard_overloaded` routing.
+    ///
+    /// TOTAL_CLIENTS is process-global and other tests mutate it
+    /// concurrently, so the bracket retries with a fresh id when an
+    /// unrelated registration lands inside the two-load window (#446
+    /// review): with the fix, one undisturbed attempt reads t2 == t1; with
+    /// the pre-fix unconditional fetch_add, EVERY attempt reads t1+1.
+    /// Cleanup runs before the verdict so a failure cannot leak entries
+    /// into later tests.
+    #[test]
+    fn replacing_register_does_not_double_count() {
+        use std::sync::atomic::Ordering;
+        let mut clean_attempt_seen = false;
+        for attempt in 0..5u64 {
+            let id = 9_060 + attempt;
+            let _a = register(id, "t:1".into(), "default".into(), 908, -1);
+            let t1 = TOTAL_CLIENTS.load(Ordering::Relaxed);
+            let _b = register(id, "t:1".into(), "default".into(), 909, -1);
+            let t2 = TOTAL_CLIENTS.load(Ordering::Relaxed);
+            deregister(id);
+            assert!(live_handle(id).is_none(), "single deregister must clear");
+            if t2 == t1 {
+                clean_attempt_seen = true;
+                break;
+            }
+        }
+        assert!(
+            clean_attempt_seen,
+            "replacing insert added to TOTAL_CLIENTS on every attempt — double-count"
+        );
+    }
 }

src/server/conn/handler_monoio/idle_park.rs

@@ -66,6 +66,12 @@ pub(super) struct IdleSlot {
     /// read — the sweep then applies [`park_after_ms`] instead of
     /// [`IDLE_DOWNSHIFT_MS`], and the woken handler exits its task.
     stage2: Cell<bool>,
+    /// #438 conn-secondary: provenance for the cancel errno. Set by
+    /// [`sweep`]/[`cancel_all_parked`] when THEY fire this slot's canceller,
+    /// consumed by the handler's error check ([`IdleSlot::was_swept_cancel`]).
+    /// A bare `ECANCELED` (125) that no sweep produced is a REAL error and
+    /// must terminate the connection, not park it.
+    swept: Cell<bool>,
 }
 
 impl IdleSlot {
@@ -78,20 +84,34 @@ impl IdleSlot {
     /// cached clock; clamped to ≥1 so 0 stays the "not parked" sentinel.
     pub(super) fn mark_parked(&self, now_ms: u64) {
         self.stage2.set(false);
+        self.swept.set(false);
         self.parked_since_ms.set(now_ms.max(1));
     }
 
     /// Mark the connection parked in stage 2 (c1M P1): downshifted, session
     /// verified parkable, cancel = task exit.
     pub(super) fn mark_parked_stage2(&self, now_ms: u64) {
         self.stage2.set(true);
+        self.swept.set(false);
         self.parked_since_ms.set(now_ms.max(1));
     }
 
     /// Mark the connection no longer parked (woke or is processing).
     pub(super) fn mark_unparked(&self) {
         self.parked_since_ms.set(0);
     }
+
+    /// #438 conn-secondary: was this error produced by a sweep/drain cancel
+    /// of THIS slot? Checks the errno shape AND consumes the provenance flag
+    /// the cancelling chore set. A raw 125 without the flag (some non-sweep
+    /// `ECANCELED` source) reads as a real error — the caller terminates
+    /// instead of parking a connection nothing will ever cancel-wake again.
+    /// The flag is re-armed `false` at every park, so a raced data delivery
+    /// (cancel fired, bytes won) cannot leave a stale `true` behind past the
+    /// next park.
+    pub(super) fn was_swept_cancel(&self, e: &std::io::Error) -> bool {
+        is_sweep_cancel(e) && self.swept.replace(false)
+    }
 }
 
 thread_local! {
@@ -118,6 +138,7 @@ pub(super) fn register(client_id: u64) -> IdleParkRegistration {
         canceller: RefCell::new(Canceller::new()),
         parked_since_ms: Cell::new(0),
         stage2: Cell::new(false),
+        swept: Cell::new(false),
     });
     REGISTRY.with(|r| r.borrow_mut().insert(client_id, slot.clone()));
     IdleParkRegistration { slot, client_id }
@@ -148,6 +169,9 @@ pub(crate) fn sweep(now_ms: u64) -> usize {
                 IDLE_DOWNSHIFT_MS
             };
             if parked != 0 && now_ms.saturating_sub(parked) >= threshold {
+                // Provenance BEFORE the cancel fires: the woken handler may
+                // check `was_swept_cancel` on this same thread's next poll.
+                slot.swept.set(true);
                 // Consume-and-replace: cancel() fires every associated op and
                 // returns a fresh canceller for the connection's next park.
                 let old = slot.canceller.replace(Canceller::new());
@@ -179,6 +203,10 @@ pub(crate) fn cancel_all_parked() -> usize {
         let mut cancelled = 0usize;
         for slot in r.borrow().values() {
             if slot.parked_since_ms.get() != 0 {
+                // Same provenance mark as `sweep` — the drained handler's
+                // shutdown-token arm exits either way, but the errno check
+                // runs first and must classify this 125 as sweep-produced.
+                slot.swept.set(true);
                 let old = slot.canceller.replace(Canceller::new());
                 let fresh = old.cancel();
                 slot.canceller.replace(fresh);
@@ -396,6 +424,47 @@ mod idle_park_tests {
         drop(reg_idle);
     }
 
+    /// #438 conn-secondary: a bare errno 125 the sweep did not produce must
+    /// read as a REAL error; a swept cancel reads as sweep exactly once.
+    #[test]
+    fn was_swept_cancel_requires_sweep_provenance() {
+        let e125 = std::io::Error::from_raw_os_error(125);
+        let reg = register(90_020);
+        reg.slot.mark_parked(10_000);
+        // No sweep ran: a raw ECANCELED is NOT a sweep cancel.
+        assert!(!reg.slot.was_swept_cancel(&e125));
+        // Sweep fires → provenance set → consumed exactly once.
+        reg.slot.mark_parked(20_000);
+        assert_eq!(sweep(20_000 + IDLE_DOWNSHIFT_MS), 1);
+        assert!(reg.slot.was_swept_cancel(&e125));
+        assert!(!reg.slot.was_swept_cancel(&e125), "flag is consume-once");
+        // Non-125 errors never match even right after a sweep.
+        reg.slot.mark_parked(40_000);
+        assert_eq!(sweep(40_000 + IDLE_DOWNSHIFT_MS), 1);
+        let reset = std::io::Error::from(std::io::ErrorKind::ConnectionReset);
+        assert!(!reg.slot.was_swept_cancel(&reset));
+    }
+
+    /// A sweep whose cancel raced a data delivery leaves the flag set; the
+    /// next park must re-arm it to false so it cannot mask a later real 125.
+    #[test]
+    fn stale_sweep_flag_cleared_on_next_park() {
+        let e125 = std::io::Error::from_raw_os_error(125);
+        let reg = register(90_021);
+        reg.slot.mark_parked_stage2(10_000);
+        assert_eq!(sweep(10_000 + park_after_ms()), 1);
+        // Handler woke with DATA (raced) — flag never consumed. Re-park:
+        reg.slot.mark_parked(50_000);
+        assert!(
+            !reg.slot.was_swept_cancel(&e125),
+            "re-park must clear stale provenance"
+        );
+        // Drain cancel marks provenance too.
+        reg.slot.mark_parked(60_000);
+        assert_eq!(cancel_all_parked(), 1);
+        assert!(reg.slot.was_swept_cancel(&e125));
+    }
+
     #[test]
     fn registration_drop_deregisters() {
         {

src/server/conn/handler_monoio/mod.rs

@@ -382,6 +382,15 @@ pub(crate) async fn handle_connection_sharded_monoio<
     let (client_live, registry_guard) = match park.kept_registration {
         Some(guard) => {
             let live = crate::client_registry::live_handle(client_id).unwrap_or_else(|| {
+                // #438 conn-secondary: reaching here means the entry vanished
+                // while its guard was alive — an invariant violation, not a
+                // code path. Fail open (re-register so the conn stays served
+                // and killable) but LOUDLY; register() itself now balances
+                // the counters if the entry raced back into existence.
+                tracing::warn!(
+                    "client {}: registry entry missing on park resume despite live guard; re-registering",
+                    client_id
+                );
                 crate::client_registry::register(
                     client_id,
                     peer_addr.clone(),
@@ -919,6 +928,16 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 && !conn.in_multi
                 && conn.command_queue.is_empty()
                 && conn.active_cross_txn.is_none()
+                // F6 (#438, sec L2): unauthenticated conns never task-park.
+                // On an auth-enabled server, parking a pre-AUTH conn would
+                // let an attacker hold a maxclients-worth of silent sockets
+                // at ~3.3 KB each, indefinitely and invisibly cheap. Keeping
+                // them un-parked leaves each one pinned to a full handler
+                // task — costly enough to surface in CPU/RSS monitoring —
+                // and `timeout N` (the slowloris knob) still reaps them. On
+                // no-auth servers `authenticated` is true from accept, so
+                // this changes nothing there.
+                && conn.authenticated
                 // Replica-handshake conns (sent REPLCONF) never park: PSYNC
                 // on a resumed parked conn is unsupported (warn+close).
                 && !conn.saw_replconf
@@ -941,7 +960,10 @@ pub(crate) async fn handle_connection_sharded_monoio<
                     // Real socket/TLS error (EOF without close_notify,
                     // ECONNRESET, …): terminate. Parking instead would spin
                     // park→wake→park forever — the dead fd stays readable.
-                    Err(ref e) if !idle_park::is_sweep_cancel(e) => break,
+                    // #438 conn-secondary: `was_swept_cancel` also demands
+                    // sweep provenance — a bare errno 125 that no sweep of
+                    // ours produced is a real error too.
+                    Err(ref e) if !reg.slot.was_swept_cancel(e) => break,
                     // F1 (#438): cancelled by the shutdown drain, not the
                     // stage-2 sweep — exit through the flush+FIN epilogue
                     // instead of task-parking into a watcher that would just
@@ -1006,7 +1028,8 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 // always performed a read that re-surfaced them; the
                 // task-park path parks WITHOUT reading, so a mistaken
                 // downshift here would feed the park→wake→park spin.)
-                Err(ref e) if !idle_park::is_sweep_cancel(e) => break,
+                // #438 conn-secondary: provenance-checked — see stage 2.
+                Err(ref e) if !reg.slot.was_swept_cancel(e) => break,
                 // F1 (#438): cancelled by the shutdown drain, not the idle
                 // sweep — exit through the flush+FIN epilogue instead of
                 // re-parking a read nothing will ever complete.

src/server/conn/handler_sharded/mod.rs

@@ -192,12 +192,17 @@ pub(crate) async fn handle_connection_sharded(
         Some(stream),
     ) = (result.0, result.1)
     {
-        use std::os::unix::io::IntoRawFd;
         match stream.into_std() {
             Ok(std_stream) => {
-                let raw_fd = std_stream.into_raw_fd();
+                // F4 (#438): the fd rides the channel as an OwnedFd — an
+                // undelivered message (ring torn down at shutdown) closes the
+                // socket on drop instead of leaking it with a stranded client.
+                let owned_fd = std::os::fd::OwnedFd::from(std_stream);
                 let msg = ShardMessage::MigrateConnection(Box::new(
-                    crate::shard::dispatch::MigrateConnectionPayload { fd: raw_fd, state },
+                    crate::shard::dispatch::MigrateConnectionPayload {
+                        fd: owned_fd,
+                        state,
+                    },
                 ));
                 let target_idx = ChannelMesh::target_index(ctx.shard_id, target_shard);
                 let push_result = {
@@ -250,12 +255,12 @@ pub(crate) async fn handle_connection_sharded(
                             // serving the client on this shard with the state
                             // recovered from the undelivered message, instead of
                             // dropping the connection. Migration disabled so the
-                            // affinity tracker cannot loop.
-                            use std::os::unix::io::FromRawFd;
-                            // SAFETY: fd is a valid, uniquely-owned descriptor from
-                            // into_raw_fd(); the failed push left ownership with us.
-                            let std_stream =
-                                unsafe { std::net::TcpStream::from_raw_fd(payload.fd) };
+                            // affinity tracker cannot loop. F4 (#438): the fd
+                            // comes back as an OwnedFd; the safe From conversion
+                            // replaces the old unsafe from_raw_fd.
+                            use std::os::fd::AsRawFd;
+                            let raw_kill_fd = payload.fd.as_raw_fd();
+                            let std_stream = std::net::TcpStream::from(payload.fd);
                             match tokio::net::TcpStream::from_std(std_stream) {
                                 Ok(tcp_stream) => {
                                     let _ = handle_connection_sharded_inner(
@@ -267,7 +272,7 @@ pub(crate) async fn handle_connection_sharded(
                                         false, // can_migrate
                                         BytesMut::new(),
                                         Some(&payload.state),
-                                        payload.fd,
+                                        raw_kill_fd,
                                     )
                                     .await;
                                 }