feat(server): task-exit parking for idle connections — 19.8 → 3.25 KB/conn (c1M P1)

CHANGELOG.md

@@ -6,6 +6,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **Task-exit parking for idle connections (c1M P1, `--conn-park-secs`,
+  default 60 s).** A plain-TCP monoio connection that stays idle past the
+  W11 downshift now has its handler task exit entirely: only a tiny
+  readiness watcher (boxed future holding the stream, ~100 B of session
+  state, and the client-registry guard) remains, reclaiming the ~6 KB task
+  state machine plus the remaining per-task buffers. The watcher wakes on
+  read-readiness (`readable(false)`, race-free on io_uring and
+  epoll/kqueue) or server shutdown and rehydrates a fresh handler through
+  the migration-restore path — wire-invisible across repeated park/wake
+  cycles. Parked connections stay in CLIENT LIST, keep their maxclients
+  slot, and CLIENT KILL still works (its `shutdown(2)` wakes the watcher;
+  the resumed handler sees EOF). Exclusions: subscriber/tracking/timeout
+  connections (structurally never reach the park arm), MULTI/EXEC or
+  cross-store-txn sessions, partial frames, TLS (keeps W11+P4b buffer
+  downshift), and the tokio runtime. `--conn-park-secs 0` disables.
+
 ### 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

src/config.rs

@@ -304,6 +304,13 @@ pub struct ServerConfig {
     #[arg(long = "uring-entries")]
     pub uring_entries: Option<u32>,
 
+    /// c1M P1: seconds a downshifted idle connection waits before its handler
+    /// task exits entirely, leaving only a tiny readiness watcher (task-exit
+    /// parking; plain-TCP monoio connections only — TLS and the tokio runtime
+    /// keep the buffer-downshift behavior). 0 disables task parking.
+    #[arg(long = "conn-park-secs", default_value_t = 60)]
+    pub conn_park_secs: u64,
+
     /// I/O driver for the monoio runtime. "auto" lets FusionDriver pick
     /// (io_uring on Linux when available, else epoll/kqueue); "epoll" forces
     /// the legacy poller. Measured on GCE ARM (c4a Axion, 2026-07): epoll is

src/main.rs

@@ -1097,6 +1097,21 @@ fn main() -> anyhow::Result<()> {
         tracing::warn!("--uring-entries has no effect under the tokio runtime");
     }
 
+    // c1M P1: stage-2 task-parking threshold — set BEFORE shard threads spawn
+    // (same contract as set_uring_entries above). Default 60s; 0 disables.
+    moon::runtime::set_conn_park_secs(config.conn_park_secs);
+    #[cfg(feature = "runtime-monoio")]
+    if config.conn_park_secs > 0 {
+        tracing::info!(
+            "Idle task-parking: downshifted connections park after {}s (--conn-park-secs)",
+            config.conn_park_secs
+        );
+    }
+    #[cfg(not(feature = "runtime-monoio"))]
+    if config.conn_park_secs != 60 {
+        tracing::warn!("--conn-park-secs has no effect under the tokio runtime");
+    }
+
     // Graph traversal timeout default — set BEFORE shard threads spawn so every
     // TraversalGuard::with_default_timeout observes it (per-query TIMEOUT overrides).
     #[cfg(feature = "graph")]

src/runtime/mod.rs

@@ -72,6 +72,25 @@ pub fn uring_entries() -> Option<u32> {
     }
 }
 
+/// c1M P1: stage-2 idle task-parking threshold (`--conn-park-secs`), in ms.
+/// 0 = task parking disabled. Set once from main BEFORE shard threads spawn
+/// (same contract as [`set_uring_entries`]); read by the monoio idle-park
+/// sweep and stage-2 read arm.
+static CONN_PARK_AFTER_MS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(60_000);
+
+/// Configure the task-parking threshold from `--conn-park-secs`.
+pub fn set_conn_park_secs(secs: u64) {
+    CONN_PARK_AFTER_MS.store(
+        secs.saturating_mul(1000),
+        std::sync::atomic::Ordering::Release,
+    );
+}
+
+/// Current task-parking threshold in ms; 0 = disabled.
+pub fn conn_park_after_ms() -> u64 {
+    CONN_PARK_AFTER_MS.load(std::sync::atomic::Ordering::Acquire)
+}
+
 /// True when the epoll busy-poll park is configured — via the
 /// `--io-busy-poll-us` flag (the caller passes the config value) or the
 /// `MOON_EPOLL_SPIN_US` env fallback the vendored driver also honors. Gates

src/server/conn/handler_monoio/idle_park.rs

@@ -39,6 +39,13 @@ use monoio::io::{AsyncReadRent, CancelHandle, Canceller};
 
 /// Park duration after which the sweep cancels a stage-1 read (ms).
 pub(super) const IDLE_DOWNSHIFT_MS: u64 = 1000;
+
+/// c1M P1: stage-2 threshold — a downshifted connection parked this long has
+/// its read cancelled so the handler TASK can exit (task-exit parking, see
+/// `conn_accept::spawn_parked_idle_watcher`). 0 = task parking disabled.
+/// The value lives in `crate::runtime` (set once from main before shards
+/// spawn, same contract as `set_uring_entries`); this is a local alias.
+pub(crate) use crate::runtime::conn_park_after_ms as park_after_ms;
 /// Rent-buffer size while downshifted. A typical idle→active wake (one
 /// command) fits; larger bursts spill into the next full-size read.
 pub(super) const IDLE_PROBE_BUF: usize = 512;
@@ -55,6 +62,10 @@ pub(super) struct IdleSlot {
     /// Shard-cached-clock ms when the connection parked its stage-1 read;
     /// 0 = not parked in stage 1 (processing, downshifted, or gone).
     parked_since_ms: Cell<u64>,
+    /// c1M P1: true while the park is a STAGE-2 (downshifted, task-parkable)
+    /// read — the sweep then applies [`park_after_ms`] instead of
+    /// [`IDLE_DOWNSHIFT_MS`], and the woken handler exits its task.
+    stage2: Cell<bool>,
 }
 
 impl IdleSlot {
@@ -66,6 +77,14 @@ impl IdleSlot {
     /// Mark the connection parked (stage 1). `now_ms` comes from the shard
     /// 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.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.parked_since_ms.set(now_ms.max(1));
     }
 
@@ -98,6 +117,7 @@ pub(super) fn register(client_id: u64) -> IdleParkRegistration {
     let slot = Rc::new(IdleSlot {
         canceller: RefCell::new(Canceller::new()),
         parked_since_ms: Cell::new(0),
+        stage2: Cell::new(false),
     });
     REGISTRY.with(|r| r.borrow_mut().insert(client_id, slot.clone()));
     IdleParkRegistration { slot, client_id }
@@ -111,11 +131,23 @@ pub(super) fn register(client_id: u64) -> IdleParkRegistration {
 /// downshifts, and re-parks in stage 2 with `parked_since_ms == 0`, so each
 /// idle connection is cancelled exactly once per idle period.
 pub(crate) fn sweep(now_ms: u64) -> usize {
+    let park_after = park_after_ms();
     REGISTRY.with(|r| {
         let mut cancelled = 0usize;
         for slot in r.borrow().values() {
             let parked = slot.parked_since_ms.get();
-            if parked != 0 && now_ms.saturating_sub(parked) >= IDLE_DOWNSHIFT_MS {
+            // Stage-2 (task-park) entries use the operator threshold; a 0
+            // threshold means task parking is off and stage-2 entries are
+            // never registered, but guard anyway.
+            let threshold = if slot.stage2.get() {
+                if park_after == 0 {
+                    continue;
+                }
+                park_after
+            } else {
+                IDLE_DOWNSHIFT_MS
+            };
+            if parked != 0 && now_ms.saturating_sub(parked) >= threshold {
                 // 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());
@@ -159,6 +191,12 @@ pub(super) fn downshift_idle_buffers(
 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).
+    const SUPPORTS_TASK_PARK: bool = false;
+
     fn idle_park_read(
         &mut self,
         buf: Vec<u8>,
@@ -175,6 +213,7 @@ pub(crate) trait IdleParkRead: AsyncReadRent {
 
 impl IdleParkRead for monoio::net::TcpStream {
     const SUPPORTS_IDLE_PARK: bool = true;
+    const SUPPORTS_TASK_PARK: bool = true;
 
     fn idle_park_read(
         &mut self,
@@ -238,6 +277,23 @@ mod idle_park_tests {
         assert_eq!(sweep(60_000 + IDLE_DOWNSHIFT_MS), 1);
     }
 
+    #[test]
+    fn stage2_uses_park_after_threshold() {
+        let reg = register(90_003);
+        // Stage-2 park: the 1s downshift threshold must NOT fire it...
+        reg.slot.mark_parked_stage2(100_000);
+        assert_eq!(
+            sweep(100_000 + IDLE_DOWNSHIFT_MS),
+            0,
+            "stage-2 park must outlive the stage-1 threshold"
+        );
+        // ...only the operator threshold does (default 60s).
+        assert_eq!(sweep(100_000 + park_after_ms()), 1);
+        // A later stage-1 park on the same slot reverts to the 1s threshold.
+        reg.slot.mark_parked(300_000);
+        assert_eq!(sweep(300_000 + IDLE_DOWNSHIFT_MS), 1);
+    }
+
     #[test]
     fn registration_drop_deregisters() {
         {

src/server/conn/handler_monoio/mod.rs

@@ -47,6 +47,20 @@ use crate::shard::dispatch::ShardMessage;
 // AtomicWaker rides that mechanism. Transaction (txn.rs) and blocking-write
 // (write.rs) paths remain on oneshots — they are off the hot path.
 
+/// RAII client-registry entry: deregisters on drop. Lives as a handler
+/// local, EXCEPT for a task-parked connection (c1M P1), where it travels
+/// inside [`MonoioHandlerResult::ParkIdle`] to the readiness watcher so the
+/// CLIENT LIST/KILL entry and maxclients slot survive the parked lifetime.
+#[cfg(feature = "runtime-monoio")]
+pub struct RegistryGuard(pub(crate) u64);
+
+#[cfg(feature = "runtime-monoio")]
+impl Drop for RegistryGuard {
+    fn drop(&mut self) {
+        crate::client_registry::deregister(self.0);
+    }
+}
+
 /// Result of `handle_connection_sharded_monoio` execution.
 ///
 /// Same purpose as the Tokio handler's `HandlerResult`: the generic handler cannot
@@ -60,6 +74,19 @@ pub enum MonoioHandlerResult {
         state: MigratedConnectionState,
         target_shard: usize,
     },
+    /// c1M P1: the connection idled past `--conn-park-secs` in the
+    /// 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.
+    ParkIdle {
+        state: Box<MigratedConnectionState>,
+        registry_guard: RegistryGuard,
+    },
     /// PSYNC arrived on this connection. Caller must hand the underlying
     /// `monoio::net::TcpStream` to
     /// `crate::replication::master::handle_psync_inline_single_shard` /
@@ -183,6 +210,10 @@ 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,
 ) -> (MonoioHandlerResult, Option<S>) {
     use monoio::io::AsyncWriteRentExt;
 
@@ -225,13 +256,7 @@ pub(crate) async fn handle_connection_sharded_monoio<
         ctx.shard_id,
         kill_fd,
     );
-    struct RegistryGuard(u64);
-    impl Drop for RegistryGuard {
-        fn drop(&mut self) {
-            crate::client_registry::deregister(self.0);
-        }
-    }
-    let _registry_guard = RegistryGuard(client_id);
+    let registry_guard = 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
@@ -665,18 +690,75 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 continue;
             }
         } else if downshifted {
-            // c10k W11 stage 2: park with the probe buffer, no chore state.
-            // Real data restores the full working set (lazily, via the
-            // pre-park sizing above) and re-arms stage 1.
-            let (result, returned_buf) = stream.read(tmp_buf).await;
-            tmp_buf = returned_buf;
-            match result {
-                Ok(0) => break,
-                Ok(n) => {
-                    read_buf.extend_from_slice(&tmp_buf[..n]);
-                    downshifted = false;
+            // c10k W11 stage 2: park with the probe buffer. Real data
+            // restores the full working set (lazily, via the pre-park sizing
+            // above) and re-arms stage 1.
+            //
+            // c1M P1: when task parking is enabled and the session is
+            // parkable, the stage-2 read is ALSO cancelable and registered
+            // with the sweep under the longer `--conn-park-secs` threshold —
+            // a cancel here means "exit the task", leaving only a tiny
+            // readiness watcher (conn_accept::spawn_parked_idle_watcher).
+            // 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
+                && 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();
+            if let (true, Some(reg)) = (parkable, idle_reg.as_ref()) {
+                let handle = reg.slot.handle();
+                reg.slot.mark_parked_stage2(ctx.cached_clock.ms());
+                let (result, returned_buf) = stream.idle_park_read(tmp_buf, handle).await;
+                reg.slot.mark_unparked();
+                tmp_buf = returned_buf;
+                match result {
+                    Ok(0) => break,
+                    Ok(n) => {
+                        read_buf.extend_from_slice(&tmp_buf[..n]);
+                        downshifted = false;
+                    }
+                    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.
+                        let state = Box::new(MigratedConnectionState {
+                            selected_db: conn.selected_db,
+                            authenticated: conn.authenticated,
+                            client_name: conn.client_name.clone(),
+                            protocol_version: conn.protocol_version,
+                            current_user: conn.current_user.clone(),
+                            flags: 0,
+                            read_buf_remainder: read_buf.split(),
+                            client_id,
+                            peer_addr: peer_addr.clone(),
+                            workspace_id: conn.workspace_id,
+                        });
+                        return (
+                            MonoioHandlerResult::ParkIdle {
+                                state,
+                                registry_guard,
+                            },
+                            Some(stream),
+                        );
+                    }
+                }
+            } else {
+                let (result, returned_buf) = stream.read(tmp_buf).await;
+                tmp_buf = returned_buf;
+                match result {
+                    Ok(0) => break,
+                    Ok(n) => {
+                        read_buf.extend_from_slice(&tmp_buf[..n]);
+                        downshifted = false;
+                    }
+                    Err(_) => break,
                 }
-                Err(_) => break,
             }
         } else if let Some(reg) = idle_reg.as_ref() {
             // c10k W11 stage 1: full-size read, registered for the shard