feat(server): c1M P1 follow-ups — TLS task-parking (47→26 KB), registration handoff, ECANCELED spin fix

CHANGELOG.md

@@ -6,6 +6,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **TLS connections task-park too (c1M P1-TLS).** Idle TLS connections now
+  exit their handler task after `--conn-park-secs`, like plain TCP: the
+  parked watcher awaits readability on the raw fd via a vendored
+  `io_ref()` passthrough, and a vendored `task_park_safe()` check vetoes
+  any park while the TLS stack holds state the fd cannot signal (wrapper
+  buffer bytes, pending EOF/error status, decrypted plaintext, a received
+  close_notify, or unsent output). Parked footprint keeps the rustls
+  session; the task future and working set are freed.
+
+### Fixed
+- **Read errors on a parked-capable connection terminate instead of
+  re-parking (park→wake→park spin).** The idle-park arms treated every
+  read error as the sweep's cancel; with task-exit parking that spun a
+  dead connection through park→wake→park at 100 % CPU forever (a TLS
+  client vanishing with a raw FIN — no close_notify — surfaces as a read
+  ERROR, and the dead fd stays readable; found live in E11 with 3 000
+  CLOSE_WAIT connections pinning a shard; an RST'd plain-TCP conn hits
+  the same loop). The arms now match monoio's exact ECANCELED (raw os
+  error 125, both drivers): only the sweep cancel downshifts/parks; real
+  errors tear down promptly. Regression tests: FIN-while-parked (TLS) and
+  RST-while-parked (plain).
+- **Resumed parked connections keep their registry identity (c1M P1
+  follow-up).** Waking from task-exit parking previously deregistered and
+  re-registered the client across a task-scheduling boundary: a racing
+  `CLIENT LIST` could briefly miss the connection, `CLIENT KILL ID` could
+  return 0, and — worse — the fresh registration silently reset the
+  connection's `CLIENT SETNAME` and `age` in `CLIENT LIST`. The wake now
+  hands the held registration through to the resumed handler
+  (registration handoff), so the entry — name, connected-at, kill state,
+  client counters — persists unbroken across any number of park/wake
+  cycles.
+
+### Changed
+- **Migrated connections task-park too (c1M P1 follow-up).** The
+  migrated-connection spawn path (Linux connection migration) now routes
+  `ParkIdle` like the primary accept path, so a connection that migrated
+  shards and then went idle parks out of the task model instead of holding
+  its full handler task forever.
+
 ## [0.8.3] — 2026-07-29
 
 ### Added

src/client_registry.rs

@@ -223,6 +223,14 @@ pub fn register(
     live
 }
 
+/// Fetch the live-state handle for an already-registered client. Used by
+/// resumed parked connections (c1M P1) to reuse their registration across a
+/// park/wake cycle instead of deregister + re-register — keeping the entry
+/// (name, connected_at, counters) continuously intact.
+pub fn live_handle(id: u64) -> Option<Arc<ClientLiveState>> {
+    stripe(id).read().get(&id).map(|e| Arc::clone(&e.live))
+}
+
 /// Deregister a client connection.
 pub fn deregister(id: u64) {
     if let Some(entry) = stripe(id).write().remove(&id) {

src/server/conn/core.rs

@@ -264,6 +264,15 @@ pub(crate) struct ConnectionState {
     pub workspace_id: Option<WorkspaceId>,
     pub command_queue: Vec<Frame>,
 
+    /// c1M P1: this connection has issued REPLCONF — it is (almost
+    /// certainly) a replica mid-handshake that will send PSYNC next, and
+    /// the resumed-parked path does not support the PSYNC hijack. Such
+    /// connections never task-park (sticky for the connection's lifetime;
+    /// health-checker probes that send a bare REPLCONF are short-lived, so
+    /// keeping them unparked costs nothing).
+    #[allow(dead_code)] // Read only by the monoio handler's park predicate
+    pub saw_replconf: bool,
+
     // Tracking
     pub tracking_state: TrackingState,
     pub tracking_rx: Option<channel::MpscReceiver<Frame>>,
@@ -323,6 +332,11 @@ impl ConnectionState {
             active_cross_txn: None,
             workspace_id: migrated.and_then(|s| s.workspace_id),
             command_queue: Vec::new(),
+            // Not carried through MigratedConnectionState: a conn that sent
+            // REPLCONF never parks (so never rehydrates through this path),
+            // and migration of a mid-handshake replica is not a supported
+            // flow (PSYNC hijacks before migration sampling matters).
+            saw_replconf: false,
             tracking_state: TrackingState::default(),
             tracking_rx: None,
             watched_keys: HashMap::new(),

src/server/conn/handler_monoio/idle_park.rs

@@ -182,6 +182,21 @@ pub(super) fn downshift_idle_buffers(
     }
 }
 
+/// The sweep's cancelled-op error, as monoio constructs it on BOTH drivers:
+/// io_uring surfaces the kernel's `-ECANCELED` (125) and the legacy driver
+/// hardcodes `from_raw_os_error(125)` in its cancel path — including on
+/// macOS, so a raw 125 check is exact everywhere (macOS errno stops at
+/// ~106; no real socket error collides).
+///
+/// The park arms MUST distinguish this from real errors: a real error (TLS
+/// EOF-without-close_notify, ECONNRESET) leaves the fd permanently
+/// readable, so treating it as "cancelled → park" spins the connection
+/// through an infinite park→wake→park loop at 100% CPU (found live in E11:
+/// 3k CLOSE_WAIT TLS conns pinned a shard).
+pub(crate) fn is_sweep_cancel(e: &std::io::Error) -> bool {
+    e.raw_os_error() == Some(125)
+}
+
 /// Stream capability trait for the two-stage idle park.
 ///
 /// The default is a plain read that ignores the cancel handle and reports
@@ -192,11 +207,21 @@ pub(crate) trait IdleParkRead: AsyncReadRent {
     const SUPPORTS_IDLE_PARK: bool = false;
 
     /// c1M P1: streams whose task can exit while the connection stays open
-    /// (requires a race-free standalone readiness await — plain TCP only;
-    /// TLS keeps W11+P4b behavior because rustls session state lives in the
-    /// stream and the wrapper exposes no readiness API).
+    /// (requires a race-free standalone readiness await on the raw fd —
+    /// plain TCP directly; TLS via the vendored wrapper's `io_ref()`
+    /// passthrough).
     const SUPPORTS_TASK_PARK: bool = false;
 
+    /// c1M P1: per-park veto for streams with internal buffering. Raw-fd
+    /// readability is only a complete wake signal when NOTHING is buffered
+    /// inside the stream itself — TLS overrides this with the vendored
+    /// `task_park_safe()` (wrapper buffers drained, no decrypted plaintext,
+    /// no pending close_notify/output). Plain TCP has no stream-owned
+    /// buffers: always true.
+    fn task_park_safe(&mut self) -> bool {
+        true
+    }
+
     fn idle_park_read(
         &mut self,
         buf: Vec<u8>,
@@ -233,6 +258,14 @@ impl IdleParkRead for monoio::net::TcpStream {
 /// two 16 KiB stream buffers.
 impl IdleParkRead for monoio_rustls::ServerTlsStream<monoio::net::TcpStream> {
     const SUPPORTS_IDLE_PARK: bool = true;
+    /// c1M P1-TLS: the parked watcher awaits readability on the raw fd via
+    /// the vendored `io_ref()` passthrough; `task_park_safe` below vetoes
+    /// any park while the TLS stack has buffered state the fd can't signal.
+    const SUPPORTS_TASK_PARK: bool = true;
+
+    fn task_park_safe(&mut self) -> bool {
+        monoio_rustls::ServerTlsStream::task_park_safe(self)
+    }
 
     fn idle_park_read(
         &mut self,

src/server/conn/handler_monoio/mod.rs

@@ -61,6 +61,37 @@ impl Drop for RegistryGuard {
     }
 }
 
+/// Park plumbing for `handle_connection_sharded_monoio` (c1M P1).
+#[cfg(feature = "runtime-monoio")]
+pub(crate) struct ParkArgs {
+    /// Only call sites that ROUTE `ParkIdle` (spawning the readiness
+    /// watcher) may pass true — anywhere else a park return would drop the
+    /// stream and silently close a healthy idle connection.
+    pub can_park: bool,
+    /// Registration carried across a park/wake cycle. A resumed parked
+    /// connection reuses this guard instead of re-registering, so the
+    /// CLIENT LIST entry (name, connected_at, kill state) stays
+    /// continuously intact and TOTAL_CLIENTS/shard counters don't churn —
+    /// there is no deregistered window a racing CLIENT LIST/KILL could
+    /// observe. `None` everywhere except `spawn_resumed_parked_conn`.
+    pub kept_registration: Option<RegistryGuard>,
+}
+
+#[cfg(feature = "runtime-monoio")]
+impl ParkArgs {
+    // Only referenced by the Linux-gated R-6 fail-open re-serve sites in
+    // conn_accept.rs; dead on non-Linux builds.
+    #[allow(dead_code)]
+    pub(crate) const NO_PARK: ParkArgs = ParkArgs {
+        can_park: false,
+        kept_registration: None,
+    };
+    pub(crate) const PARK: ParkArgs = ParkArgs {
+        can_park: true,
+        kept_registration: None,
+    };
+}
+
 /// Result of `handle_connection_sharded_monoio` execution.
 ///
 /// Same purpose as the Tokio handler's `HandlerResult`: the generic handler cannot
@@ -78,11 +109,12 @@ pub enum MonoioHandlerResult {
     /// downshifted state; the handler task exits and the caller parks the
     /// stream behind a tiny readiness watcher
     /// (`conn_accept::spawn_parked_idle_watcher`). Only returned when the
-    /// caller opted in via `can_park` — every other call site would drop the
-    /// stream and silently close a healthy connection. `registry_guard`
-    /// keeps the CLIENT LIST/KILL entry (and its maxclients slot) alive for
-    /// the parked lifetime; the watcher drops it just before re-registering
-    /// on wake.
+    /// caller opted in via `park.can_park` — every other call site would
+    /// drop the stream and silently close a healthy connection.
+    /// `registry_guard` keeps the CLIENT LIST/KILL entry (and its maxclients
+    /// slot) alive for the parked lifetime; on wake the watcher hands it
+    /// back to the resumed handler (`ParkArgs::kept_registration`), so the
+    /// registry entry is never dropped across a park/wake cycle.
     ParkIdle {
         state: Box<MigratedConnectionState>,
         registry_guard: RegistryGuard,
@@ -210,10 +242,9 @@ pub(crate) async fn handle_connection_sharded_monoio<
     // (non-unix). Threaded from the concrete spawn site; the generic `S` here
     // has no `AsRawFd` bound.
     kill_fd: i32,
-    // c1M P1: only call sites that ROUTE `ParkIdle` (spawning the readiness
-    // watcher) may pass true — anywhere else a park return would drop the
-    // stream and silently close a healthy idle connection.
-    can_park: bool,
+    // c1M P1 park plumbing: opt-in flag + optional registration carried
+    // across a park/wake cycle (see [`ParkArgs`]).
+    park: ParkArgs,
 ) -> (MonoioHandlerResult, Option<S>) {
     use monoio::io::AsyncWriteRentExt;
 
@@ -248,15 +279,37 @@ pub(crate) async fn handle_connection_sharded_monoio<
     conn.refresh_acl_cache(&ctx.acl_table);
     let db_count = ctx.shard_databases.db_count();
 
-    // Register in global client registry for CLIENT LIST/INFO/KILL.
-    let client_live = crate::client_registry::register(
-        client_id,
-        peer_addr.clone(),
-        conn.current_user.clone(),
-        ctx.shard_id,
-        kill_fd,
-    );
-    let registry_guard = RegistryGuard(client_id);
+    // Register in global client registry for CLIENT LIST/INFO/KILL. A
+    // resumed parked connection arrives with its registration still held
+    // (`kept_registration`) and reuses it — the entry was never dropped, so
+    // name/connected_at/kill state persist and counters don't churn. The
+    // `live_handle` miss arm is unreachable while the guard is alive (the
+    // guard is the only deregistration path); registering fresh there is a
+    // fail-safe, not a code path.
+    let (client_live, registry_guard) = match park.kept_registration {
+        Some(guard) => {
+            let live = crate::client_registry::live_handle(client_id).unwrap_or_else(|| {
+                crate::client_registry::register(
+                    client_id,
+                    peer_addr.clone(),
+                    conn.current_user.clone(),
+                    ctx.shard_id,
+                    kill_fd,
+                )
+            });
+            (live, guard)
+        }
+        None => (
+            crate::client_registry::register(
+                client_id,
+                peer_addr.clone(),
+                conn.current_user.clone(),
+                ctx.shard_id,
+                kill_fd,
+            ),
+            RegistryGuard(client_id),
+        ),
+    };
 
     // Functions API registry — LAZY per connection (P-1 footprint): built on
     // first FUNCTION/FCALL/FCALL_RO via `ensure_function_registry`, so the
@@ -702,14 +755,21 @@ pub(crate) async fn handle_connection_sharded_monoio<
             // The predicate is stable while parked in read (no commands can
             // execute), and reaching this arm already excludes subscriber /
             // tracking / idle-timeout connections (each takes its own arm).
-            let parkable = can_park
+            let parkable = park.can_park
                 && S::SUPPORTS_TASK_PARK
                 && idle_park::park_after_ms() > 0
                 && read_buf.is_empty()
                 && write_buf.is_empty()
                 && !conn.in_multi
                 && conn.command_queue.is_empty()
-                && conn.active_cross_txn.is_none();
+                && conn.active_cross_txn.is_none()
+                // Replica-handshake conns (sent REPLCONF) never park: PSYNC
+                // on a resumed parked conn is unsupported (warn+close).
+                && !conn.saw_replconf
+                // Stream-side veto LAST (it can do real work): TLS refuses
+                // while its wrapper buffers / rustls session hold anything
+                // the raw fd's readability can't signal.
+                && stream.task_park_safe();
             if let (true, Some(reg)) = (parkable, idle_reg.as_ref()) {
                 let handle = reg.slot.handle();
                 reg.slot.mark_parked_stage2(ctx.cached_clock.ms());
@@ -722,11 +782,14 @@ pub(crate) async fn handle_connection_sharded_monoio<
                         read_buf.extend_from_slice(&tmp_buf[..n]);
                         downshifted = false;
                     }
+                    // 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,
                     Err(_) => {
-                        // Cancelled by the stage-2 sweep (or a real socket
-                        // error, which the resumed handler's first read will
-                        // surface as EOF/err): exit the task. read_buf is
-                        // empty (predicate), so no partial frame is at risk.
+                        // Cancelled by the stage-2 sweep: exit the task.
+                        // read_buf is empty (predicate), so no partial frame
+                        // is at risk.
                         let state = Box::new(MigratedConnectionState {
                             selected_db: conn.selected_db,
                             authenticated: conn.authenticated,
@@ -774,11 +837,15 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 Ok(n) => {
                     read_buf.extend_from_slice(&tmp_buf[..n]);
                 }
+                // Real socket/TLS error: terminate now. (Pre-P1 this arm
+                // could lump errors in with the cancel because stage 2
+                // 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,
                 Err(_) => {
-                    // Cancelled by the idle sweep — or a real socket error,
-                    // which the stage-2 re-park's own read surfaces
-                    // immediately. Either way: shed the working set, re-park
-                    // small. No error-kind matching needed.
+                    // Cancelled by the idle sweep: shed the working set,
+                    // re-park small.
                     idle_park::downshift_idle_buffers(&mut tmp_buf, &mut read_buf, &mut write_buf);
                     stream.on_idle_downshift();
                     downshifted = true;
@@ -1063,6 +1130,10 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 continue;
             }
             if cmd_len == 8 && dispatch::try_handle_replconf(cmd, cmd_args, ctx, &mut responses) {
+                // Likely a replica mid-handshake (PSYNC next): permanently
+                // exclude from task-parking — the resumed-parked path does
+                // not support the PSYNC hijack.
+                conn.saw_replconf = true;
                 continue;
             }
             // PSYNC: arrives only on a master, hijacks the connection. Encode