diff --git a/CHANGELOG.md b/CHANGELOG.md index c4985f22..2b7d611d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/client_registry.rs b/src/client_registry.rs index a739174b..b838d26f 100644 --- a/src/client_registry.rs +++ b/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> { + 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) { diff --git a/src/server/conn/core.rs b/src/server/conn/core.rs index 21350d7e..8e9cd06b 100644 --- a/src/server/conn/core.rs +++ b/src/server/conn/core.rs @@ -264,6 +264,15 @@ pub(crate) struct ConnectionState { pub workspace_id: Option, pub command_queue: Vec, + /// 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>, @@ -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(), diff --git a/src/server/conn/handler_monoio/idle_park.rs b/src/server/conn/handler_monoio/idle_park.rs index 435091eb..8620ca0e 100644 --- a/src/server/conn/handler_monoio/idle_park.rs +++ b/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, @@ -233,6 +258,14 @@ impl IdleParkRead for monoio::net::TcpStream { /// two 16 KiB stream buffers. impl IdleParkRead for monoio_rustls::ServerTlsStream { 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, diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 4efc5f23..25f7d32f 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/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, +} + +#[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, 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) { 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 diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index 864be8d9..bdab568a 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -717,7 +717,10 @@ pub(crate) fn spawn_monoio_connection( .await { Ok(Ok(tls_stream)) => { - let _ = handle_connection_sharded_monoio( + // c1M P1-TLS: kept for the ParkIdle routing below + // (`sd` moves into the handler call). + let sd_park = sd.clone(); + let result = handle_connection_sharded_monoio( tls_stream, peer_addr, &conn_ctx, @@ -727,9 +730,30 @@ pub(crate) fn spawn_monoio_connection( BytesMut::new(), None, // fresh connection kill_fd, - false, // can_park: TLS keeps W11+P4b (no task park) + crate::server::conn::handler_monoio::ParkArgs::PARK, // routes ParkIdle below ) .await; + if let ( + crate::server::conn::handler_monoio::MonoioHandlerResult::ParkIdle { + state, + registry_guard, + }, + Some(stream), + ) = result + { + // Watcher chain owns the close accounting now + // (the `return` skips this task's decrement). + spawn_parked_idle_watcher( + stream, + state, + registry_guard, + Box::new(conn_ctx), + sd_park, + cid, + kill_fd, + ); + return; + } } Ok(Err(e)) => { tracing::warn!( @@ -774,7 +798,7 @@ pub(crate) fn spawn_monoio_connection( BytesMut::new(), None, // fresh connection kill_fd, - true, // can_park: this site routes ParkIdle below + crate::server::conn::handler_monoio::ParkArgs::PARK, // this site routes ParkIdle below ) .await; @@ -893,7 +917,7 @@ pub(crate) fn spawn_monoio_connection( false, // can_migrate BytesMut::new(), Some(&state), kill_fd, - false, // can_park: result is discarded here + crate::server::conn::handler_monoio::ParkArgs::NO_PARK, // result is discarded here ) .await; } else { @@ -950,7 +974,7 @@ pub(crate) fn spawn_monoio_connection( false, // can_migrate: pin locally, no retry loop BytesMut::new(), Some(&payload.state), kill_fd, - false, // can_park: result is discarded here + crate::server::conn::handler_monoio::ParkArgs::NO_PARK, // result is discarded here ) .await; } @@ -973,27 +997,34 @@ pub(crate) fn spawn_monoio_connection( } } -/// Spawn a migrated connection handler on the target shard (monoio runtime). -/// -/// Reconstructs a `monoio::net::TcpStream` from a raw FD transferred via -/// `ShardMessage::MigrateConnection`, prepends synthetic RESP commands for state -/// restoration, and spawns a handler with `requirepass = None` (pre-authenticated). -/// -/// # Safety -/// -/// Same safety requirements as `spawn_migrated_tokio_connection`: the caller must -/// ensure `fd` is a valid, open file descriptor for a connected TCP socket. -/// -/// # Limitations -/// -/// TLS connections cannot be migrated (TLS session state is in userspace). -// `unix`-gated alongside `runtime-monoio` for the same reason as the tokio twin -// above: the `RawFd` signature + `from_raw_fd` are Unix-only. Always true on -// supported targets (Linux + macOS). +/// c1M P1: raw-fd readiness await for the parked-idle watcher. TLS parks on +/// its UNDERLYING socket's readability (the vendored `io_ref()` +/// passthrough) — sound because the handler's `task_park_safe()` veto +/// guarantees nothing was buffered inside the TLS stack at park time, so +/// the next wire byte is the only possible wake source. +#[cfg(all(feature = "runtime-monoio", unix))] +pub(crate) trait ParkWatchable { + fn park_readable(&self) -> impl std::future::Future>; +} + +#[cfg(all(feature = "runtime-monoio", unix))] +impl ParkWatchable for monoio::net::TcpStream { + fn park_readable(&self) -> impl std::future::Future> { + self.readable(false) + } +} + +#[cfg(all(feature = "runtime-monoio", unix))] +impl ParkWatchable for monoio_rustls::ServerTlsStream { + fn park_readable(&self) -> impl std::future::Future> { + self.io_ref().readable(false) + } +} + /// c1M P1: park an idle connection out of the task model. The fat handler /// task has exited; this tiny watcher owns the stream, the ~100 B session /// state, and the client-registry guard (CLIENT LIST/KILL entry + maxclients -/// slot stay live). It wakes on read-readiness (`readable(false)` is +/// slot stay live). It wakes on read-readiness (`park_readable()` is /// race-free on both drivers: uring PollAdd / legacy readiness+poll) or /// server shutdown. CLIENT KILL's `shutdown(2)` makes the fd read-ready, so /// a parked kill flows through the wake path and the resumed handler's first @@ -1004,18 +1035,24 @@ pub(crate) fn spawn_monoio_connection( /// park→resume→park type cycle (the resumed handler task constructs this /// watcher again on re-park). #[cfg(all(feature = "runtime-monoio", unix))] -fn spawn_parked_idle_watcher( - stream: monoio::net::TcpStream, +fn spawn_parked_idle_watcher( + stream: S, state: Box, registry_guard: crate::server::conn::handler_monoio::RegistryGuard, conn_ctx: Box, shutdown: CancellationToken, client_id: u64, kill_fd: i32, -) { +) where + S: monoio::io::AsyncReadRent + + monoio::io::AsyncWriteRent + + crate::server::conn::handler_monoio::idle_park::IdleParkRead + + ParkWatchable + + 'static, +{ let fut: std::pin::Pin>> = Box::pin(async move { let woke = monoio::select! { - res = stream.readable(false) => { + res = stream.park_readable() => { // Err (fd error) also resumes: the handler's first read // surfaces the real error and tears down cleanly. let _ = res; @@ -1030,18 +1067,19 @@ fn spawn_parked_idle_watcher( crate::admin::metrics_setup::record_connection_closed(); return; } - // Wake: the handler re-registers this client_id at entry, so drop - // the parked entry first. The deregistered window lasts until the - // executor first polls the resumed task (a task-scheduling boundary, - // not just a few instructions): a racing CLIENT LIST misses the - // conn and CLIENT KILL ID returns 0 — same observable as a - // reconnect race, and a kill_flag set in that window is superseded - // by the shutdown(2) the killer already issued, which is what woke - // us. Closing the gap needs registration handoff into the handler - // (pass the guard through instead of drop/re-register); not worth - // it for a transient-invisibility race on an actively-waking conn. - drop(registry_guard); - spawn_resumed_parked_conn(stream, state, conn_ctx, shutdown, client_id, kill_fd); + // Wake: hand the registration to the resumed handler (registration + // handoff) — the entry is never dropped, so there is no window where + // a racing CLIENT LIST misses the conn or CLIENT KILL ID returns 0, + // and name/connected_at/kill state persist across the wake. + spawn_resumed_parked_conn( + stream, + state, + registry_guard, + conn_ctx, + shutdown, + client_id, + kill_fd, + ); }); monoio::spawn(fut); } @@ -1052,14 +1090,21 @@ fn spawn_parked_idle_watcher( /// would need the primary spawn site's full migration routing) and route /// their own re-parks back to [`spawn_parked_idle_watcher`]. #[cfg(all(feature = "runtime-monoio", unix))] -fn spawn_resumed_parked_conn( - stream: monoio::net::TcpStream, +fn spawn_resumed_parked_conn( + stream: S, mut state: Box, + registry_guard: crate::server::conn::handler_monoio::RegistryGuard, conn_ctx: Box, shutdown: CancellationToken, client_id: u64, kill_fd: i32, -) { +) where + S: monoio::io::AsyncReadRent + + monoio::io::AsyncWriteRent + + crate::server::conn::handler_monoio::idle_park::IdleParkRead + + ParkWatchable + + 'static, +{ use crate::server::connection::handle_connection_sharded_monoio; monoio::spawn(async move { let initial = std::mem::take(&mut state.read_buf_remainder); @@ -1074,7 +1119,12 @@ fn spawn_resumed_parked_conn( initial, Some(&state), kill_fd, - true, // can_park: this site routes ParkIdle below + // Routes ParkIdle below; carries the parked registration through + // the wake so the registry entry is never dropped. + crate::server::conn::handler_monoio::ParkArgs { + can_park: true, + kept_registration: Some(registry_guard), + }, ) .await; match (outcome, stream_back) { @@ -1115,6 +1165,23 @@ fn spawn_resumed_parked_conn( }); } +/// Spawn a migrated connection handler on the target shard (monoio runtime). +/// +/// Reconstructs a `monoio::net::TcpStream` from a raw FD transferred via +/// `ShardMessage::MigrateConnection`, prepends synthetic RESP commands for state +/// restoration, and spawns a handler with `requirepass = None` (pre-authenticated). +/// +/// # Safety +/// +/// Same safety requirements as `spawn_migrated_tokio_connection`: the caller must +/// ensure `fd` is a valid, open file descriptor for a connected TCP socket. +/// +/// # Limitations +/// +/// TLS connections cannot be migrated (TLS session state is in userspace). +// `unix`-gated alongside `runtime-monoio` for the same reason as the tokio twin +// above: the `RawFd` signature + `from_raw_fd` are Unix-only. Always true on +// supported targets (Linux + macOS). #[cfg(all(feature = "runtime-monoio", unix))] #[allow(clippy::too_many_arguments)] pub(crate) fn spawn_migrated_monoio_connection( @@ -1265,7 +1332,10 @@ pub(crate) fn spawn_migrated_monoio_connection( #[cfg(not(unix))] let kill_fd = -1; monoio::spawn(async move { - let _ = handle_connection_sharded_monoio( + // c1M P1: kept for the ParkIdle routing below (`sd` moves + // into the handler call). + let sd_park = sd.clone(); + let result = handle_connection_sharded_monoio( tcp_stream, peer_addr, &conn_ctx, @@ -1275,9 +1345,30 @@ pub(crate) fn spawn_migrated_monoio_connection( migration_buf, Some(&state), kill_fd, - false, // can_park: this site discards the result (follow-up: route ParkIdle here too) + crate::server::conn::handler_monoio::ParkArgs::PARK, // routes ParkIdle below ) .await; + if let ( + crate::server::conn::handler_monoio::MonoioHandlerResult::ParkIdle { + state, + registry_guard, + }, + Some(stream), + ) = result + { + // Idle migrated connection parks like a plain one; the + // watcher chain now owns the balancing decrement below. + spawn_parked_idle_watcher( + stream, + state, + registry_guard, + Box::new(conn_ctx), + sd_park, + cid, + kill_fd, + ); + return; + } // Migrated connection: the source shard's wrapper skipped the // decrement (because `_migrated == true`), so this target-shard // spawn site owns the balancing decrement. Without this the diff --git a/tests/parked_idle_parity.rs b/tests/parked_idle_parity.rs index 267a192d..60aa4918 100644 --- a/tests/parked_idle_parity.rs +++ b/tests/parked_idle_parity.rs @@ -200,6 +200,122 @@ fn parked_connection_visible_and_killable() { let _ = child.wait(); } +/// Registry identity must survive a park/wake cycle unbroken: the resumed +/// connection keeps its CLIENT SETNAME, its id, and CLIENT LIST never +/// double-counts (registration handoff — the wake must not deregister and +/// re-register across a task-scheduling boundary). +#[test] +fn resumed_connection_keeps_registry_identity() { + let dir = tempfile::tempdir().expect("tempdir"); + let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p)); + + let mut victim = TcpStream::connect(("127.0.0.1", port)).expect("connect victim"); + victim.set_nodelay(true).ok(); + let r = command_reply(&mut victim, "CLIENT SETNAME wakekeeper\r\n"); + assert!(r.starts_with("+OK"), "SETNAME failed: {r}"); + + let mut control = TcpStream::connect(("127.0.0.1", port)).expect("connect control"); + let list = command_reply(&mut control, "CLIENT LIST\r\n"); + let orig_id: u64 = list + .lines() + .find(|l| l.contains("name=wakekeeper")) + .expect("named conn listed before park") + .split_whitespace() + .find_map(|kv| kv.strip_prefix("id=")) + .expect("id= field") + .parse() + .expect("numeric id"); + + // Two full park → data-wake cycles. + for cycle in 0..2 { + std::thread::sleep(PARK_WAIT); + ping(&mut victim); // data-wake + + let list = command_reply(&mut control, "CLIENT LIST\r\n"); + let line = list + .lines() + .find(|l| l.contains("name=wakekeeper")) + .unwrap_or_else(|| { + panic!("cycle {cycle}: resumed conn lost its name in CLIENT LIST:\n{list}") + }); + let id: u64 = line + .split_whitespace() + .find_map(|kv| kv.strip_prefix("id=")) + .expect("id= field") + .parse() + .expect("numeric id"); + assert_eq!(id, orig_id, "cycle {cycle}: client id changed across wake"); + let entries = list.lines().filter(|l| l.contains("id=")).count(); + assert_eq!( + entries, 2, + "cycle {cycle}: CLIENT LIST must hold exactly victim+control:\n{list}" + ); + } + + let _ = child.kill(); + let _ = child.wait(); +} + +/// A client that dies with an RST while its connection is parked must be +/// torn down promptly — an RST'd fd stays readable forever, so a handler +/// that mistakes the resulting read error for a sweep-cancel would spin the +/// connection through park→wake→park at 100% CPU (the E11 spin, plain-TCP +/// variant) and pin the fd in CLOSE_WAIT. +#[cfg(unix)] // SO_LINGER(0) via libc; std's set_linger is unstable +#[test] +fn rst_while_parked_tears_down_promptly() { + let dir = tempfile::tempdir().expect("tempdir"); + let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p)); + + let victim = TcpStream::connect(("127.0.0.1", port)).expect("connect victim"); + { + let mut v = &victim; + v.write_all(b"CLIENT SETNAME rstvictim\r\n").expect("write"); + let mut buf = [0u8; 8]; + victim + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("timeout"); + let _ = (&victim).read(&mut buf); + } + std::thread::sleep(PARK_WAIT); // parked now + + // RST close: SO_LINGER(0) makes drop send RST instead of FIN. + { + use std::os::unix::io::AsRawFd; + let linger = libc::linger { + l_onoff: 1, + l_linger: 0, + }; + // SAFETY: `victim` owns a live socket fd for the duration of this + // call, and the pointer/length describe a valid `libc::linger` on + // the stack. + let rc = unsafe { + libc::setsockopt( + victim.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_LINGER, + &linger as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + assert_eq!(rc, 0, "SO_LINGER(0) must apply"); + } + drop(victim); + + // The server must fully release the conn: gone from CLIENT LIST and no + // respawn loop keeping it alive. + std::thread::sleep(Duration::from_millis(2000)); + let mut control = TcpStream::connect(("127.0.0.1", port)).expect("connect control"); + let list = command_reply(&mut control, "CLIENT LIST\r\n"); + assert!( + !list.contains("name=rstvictim"), + "RST'd parked connection still registered:\n{list}" + ); + + let _ = child.kill(); + let _ = child.wait(); +} + /// An active sibling must be undisturbed while its neighbor parks and wakes. #[test] fn active_sibling_undisturbed_by_parking() { diff --git a/tests/tls_idle_downshift_parity.rs b/tests/tls_idle_downshift_parity.rs index 43422538..1758ec63 100644 --- a/tests/tls_idle_downshift_parity.rs +++ b/tests/tls_idle_downshift_parity.rs @@ -139,6 +139,15 @@ fn generate_cert(dir: &std::path::Path) -> bool { } fn spawn_moon_tls(dir: &std::path::Path, port: u16, tls_port: u16) -> std::process::Child { + spawn_moon_tls_park(dir, port, tls_port, 0) +} + +fn spawn_moon_tls_park( + dir: &std::path::Path, + port: u16, + tls_port: u16, + park_secs: u64, +) -> std::process::Child { Command::new(common::find_moon_binary()) .args([ "--port", @@ -151,6 +160,8 @@ fn spawn_moon_tls(dir: &std::path::Path, port: u16, tls_port: u16) -> std::proce "0", "--tls-port", &tls_port.to_string(), + "--conn-park-secs", + &park_secs.to_string(), ]) .arg("--tls-cert-file") .arg(dir.join("cert.pem")) @@ -230,3 +241,182 @@ fn tls_idle_connection_serves_all_traffic_after_downshift() { let _ = child.kill(); let _ = child.wait(); } + +/// c1M P1-TLS: task-exit parking must be invisible on the TLS wire. With +/// `--conn-park-secs 2`, an idle TLS connection downshifts (~1s), then its +/// handler task EXITS (~2s more), leaving a watcher on the raw fd via the +/// vendored `io_ref()` passthrough. Every byte sent afterwards must be +/// served over the SAME TLS session across repeated park/wake cycles — +/// a park while any TLS-stack buffer held data would hang the read here, +/// and a session desync would fail the decrypt loudly. A plain-TCP control +/// connection proves the parked TLS conn stays visible and killable. +#[test] +fn tls_parked_connection_serves_traffic_and_stays_killable() { + const PARK_WAIT: Duration = Duration::from_millis(4600); + + let dir = tempfile::tempdir().expect("tempdir"); + if !generate_cert(dir.path()) { + eprintln!("SKIP: openssl CLI not available for cert generation"); + return; + } + let tls_port = common::reserve_port(); + let (mut child, plain_port) = + common::spawn_listening(|p| spawn_moon_tls_park(dir.path(), p, tls_port, 2)); + let mut conn = None; + for _ in 0..50 { + match std::panic::catch_unwind(|| TlsConn::connect(tls_port)) { + Ok(c) => { + conn = Some(c); + break; + } + Err(_) => std::thread::sleep(Duration::from_millis(100)), + } + } + let mut conn = conn.expect("TLS connect"); + conn.write_all(b"CLIENT SETNAME tlspark\r\n"); + let r = conn.read_exact_deadline(5); + assert_eq!(&r, b"+OK\r\n"); + + // Cycle 1: park past the task-exit threshold, then a probe wake. + std::thread::sleep(PARK_WAIT); + conn.ping(); + + // Cycle 2: re-park (the resumed handler must re-arm both stages), then + // a 100-deep pipeline over the same session. + std::thread::sleep(PARK_WAIT); + let pipeline = b"PING\r\n".repeat(100); + conn.write_all(&pipeline); + let replies = conn.read_exact_deadline(7 * 100); + assert_eq!(replies.len(), 700, "all 100 pipelined replies must arrive"); + + // Cycle 3: re-park, then verify the parked conn is visible (with its + // name — registration handoff) and killable from a plain-TCP control. + std::thread::sleep(PARK_WAIT); + let mut control = TcpStream::connect(("127.0.0.1", plain_port)).expect("connect control"); + control + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("set timeout"); + let mut list = Vec::new(); + control.write_all(b"CLIENT LIST\r\n").expect("write LIST"); + let mut chunk = [0u8; 65536]; + loop { + let n = control.read(&mut chunk).expect("read LIST"); + assert!(n > 0, "control closed mid-reply"); + list.extend_from_slice(&chunk[..n]); + if list.ends_with(b"\r\n") && list.starts_with(b"$") { + let s = String::from_utf8_lossy(&list); + if let Some(pos) = s.find('\n') + && let Ok(len) = s[1..pos - 1].trim().parse::() + && list.len() >= pos + 1 + len + 2 + { + break; + } + } + } + let list = String::from_utf8_lossy(&list); + let victim_line = list + .lines() + .find(|l| l.contains("name=tlspark")) + .unwrap_or_else(|| panic!("parked TLS conn missing from CLIENT LIST:\n{list}")); + let victim_id: u64 = victim_line + .split_whitespace() + .find_map(|kv| kv.strip_prefix("id=")) + .expect("id= field") + .parse() + .expect("numeric id"); + + control + .write_all(format!("CLIENT KILL ID {victim_id}\r\n").as_bytes()) + .expect("write KILL"); + let mut r = [0u8; 16]; + let n = control.read(&mut r).expect("read KILL reply"); + assert!( + r[..n].starts_with(b":1"), + "CLIENT KILL must report 1 killed conn, got: {}", + String::from_utf8_lossy(&r[..n]) + ); + + // The parked TLS victim's socket must observe the close promptly. + conn.sock + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("set timeout"); + let mut buf = [0u8; 16]; + match conn.sock.read(&mut buf) { + Ok(0) => {} // EOF + Err(_) => {} // RST + Ok(n) => panic!("expected close, got {n} raw bytes"), + } + + let _ = child.kill(); + let _ = child.wait(); +} + +/// A TLS client that vanishes with a raw TCP FIN (no close_notify) while +/// parked must be torn down promptly. rustls surfaces that FIN as a read +/// ERROR (unexpected EOF), not `Ok(0)` — a handler that lumps read errors +/// in with the sweep-cancel re-parks the dead fd, which stays readable +/// forever: the park→wake→park spin found live in E11 (3k CLOSE_WAIT conns, +/// shard at 100% CPU). +#[test] +fn tls_fin_while_parked_tears_down_promptly() { + const PARK_WAIT: Duration = Duration::from_millis(4600); + + let dir = tempfile::tempdir().expect("tempdir"); + if !generate_cert(dir.path()) { + eprintln!("SKIP: openssl CLI not available for cert generation"); + return; + } + let tls_port = common::reserve_port(); + let (mut child, plain_port) = + common::spawn_listening(|p| spawn_moon_tls_park(dir.path(), p, tls_port, 2)); + let mut conn = None; + for _ in 0..50 { + match std::panic::catch_unwind(|| TlsConn::connect(tls_port)) { + Ok(c) => { + conn = Some(c); + break; + } + Err(_) => std::thread::sleep(Duration::from_millis(100)), + } + } + let mut conn = conn.expect("TLS connect"); + conn.write_all(b"CLIENT SETNAME finvictim\r\n"); + let r = conn.read_exact_deadline(5); + assert_eq!(&r, b"+OK\r\n"); + + std::thread::sleep(PARK_WAIT); // parked now + + // Raw TCP close WITHOUT a TLS close_notify: drop the socket only. + drop(conn); + + std::thread::sleep(Duration::from_millis(2000)); + let mut control = TcpStream::connect(("127.0.0.1", plain_port)).expect("connect control"); + control + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("set timeout"); + control.write_all(b"CLIENT LIST\r\n").expect("write LIST"); + let mut list = Vec::new(); + let mut chunk = [0u8; 65536]; + loop { + let n = control.read(&mut chunk).expect("read LIST"); + assert!(n > 0, "control closed mid-reply"); + list.extend_from_slice(&chunk[..n]); + if list.ends_with(b"\r\n") && list.starts_with(b"$") { + let s = String::from_utf8_lossy(&list); + if let Some(pos) = s.find('\n') + && let Ok(len) = s[1..pos - 1].trim().parse::() + && list.len() >= pos + 1 + len + 2 + { + break; + } + } + } + let list = String::from_utf8_lossy(&list); + assert!( + !list.contains("name=finvictim"), + "FIN'd parked TLS connection still registered:\n{list}" + ); + + let _ = child.kill(); + let _ = child.wait(); +} diff --git a/vendor/monoio-io-wrapper/src/lib.rs b/vendor/monoio-io-wrapper/src/lib.rs index 173e65cc..e816e35f 100644 --- a/vendor/monoio-io-wrapper/src/lib.rs +++ b/vendor/monoio-io-wrapper/src/lib.rs @@ -74,6 +74,18 @@ impl ReadBuffer { } } + /// moon patch (c1M P1-TLS): no bytes buffered and no deferred status + /// pending. Conservatively false for unsafe-io buffers (never + /// task-park them). + #[inline] + pub fn is_drained(&self) -> bool { + match self { + Self::Safe(b) => b.is_drained(), + #[cfg(feature = "unsafe_io")] + Self::Unsafe(_) => false, + } + } + #[inline] #[cfg(feature = "unsafe_io")] pub fn is_safe(&self) -> bool { @@ -146,6 +158,17 @@ impl WriteBuffer { } } + /// moon patch (c1M P1-TLS): no bytes buffered and no stashed write + /// error pending. Conservatively false for unsafe-io buffers. + #[inline] + pub fn is_drained(&self) -> bool { + match self { + Self::Safe(b) => b.is_drained(), + #[cfg(feature = "unsafe_io")] + Self::Unsafe(_) => false, + } + } + #[inline] #[cfg(feature = "unsafe_io")] pub fn is_safe(&self) -> bool { diff --git a/vendor/monoio-io-wrapper/src/safe_io.rs b/vendor/monoio-io-wrapper/src/safe_io.rs index 072e6711..a1685eb6 100644 --- a/vendor/monoio-io-wrapper/src/safe_io.rs +++ b/vendor/monoio-io-wrapper/src/safe_io.rs @@ -220,6 +220,20 @@ impl SafeRead { .expect("buffer mut expected") .release_if_empty() } + + /// moon patch (c1M P1-TLS): true when the buffer holds no bytes AND no + /// deferred status (EOF/error) is pending delivery. Task-parking on raw + /// fd readability is only correct when nothing is waiting here — a + /// buffered byte or a pending EOF would never make the fd readable. + /// (`buffer` is only `None` transiently inside `do_io`, which holds + /// `&mut self` — but answer `false` rather than panic if it ever is: + /// "don't park" is always the safe verdict.) + pub fn is_drained(&self) -> bool { + match (self.buffer.as_ref(), &self.status) { + (Some(buffer), ReadStatus::Ok) => buffer.is_empty(), + _ => false, + } + } } impl io::Read for SafeRead { @@ -323,6 +337,16 @@ impl SafeWrite { .expect("buffer mut expected") .release_if_empty() } + + /// moon patch (c1M P1-TLS): true when no unflushed bytes and no stashed + /// write error are pending. See `SafeRead::is_drained` (incl. the + /// `None`-buffer rationale). + pub fn is_drained(&self) -> bool { + match (self.buffer.as_ref(), &self.status) { + (Some(buffer), WriteStatus::Ok) => buffer.is_empty(), + _ => false, + } + } } impl io::Write for SafeWrite { @@ -521,4 +545,45 @@ mod moon_patch_tests { let replay = r.read(&mut out).unwrap_err(); assert_eq!(replay.kind(), io::ErrorKind::ConnectionReset); } + + // moon patch (c1M P1-TLS): is_drained must be false whenever a byte OR a + // deferred status is pending — task-parking on fd readability while + // either waits would hang the connection. + #[test] + fn read_is_drained_tracks_bytes_and_deferred_status() { + let mut r = SafeRead::new(64); + assert!(r.is_drained(), "fresh buffer is drained"); + let mut io = MockIo { + results: vec![Ok(b"data".to_vec())], + }; + now(r.do_io(&mut io)).unwrap(); + assert!(!r.is_drained(), "buffered bytes must block parking"); + let mut out = [0u8; 8]; + r.read(&mut out).unwrap(); + assert!(r.is_drained(), "drained after consume"); + // Pending EOF status must also block parking (it would never make + // the fd readable again). + let mut io = MockIo { + results: vec![Ok(Vec::new())], + }; + now(r.do_io(&mut io)).unwrap(); + assert!(!r.is_drained(), "pending EOF must block parking"); + // Pending stashed error likewise. + let mut r2 = SafeRead::new(64); + let mut io = MockIo { + results: vec![Err(io::Error::new(io::ErrorKind::ConnectionReset, "boom"))], + }; + let _ = now(r2.do_io(&mut io)); + assert!(!r2.is_drained(), "stashed error must block parking"); + } + + #[test] + fn write_is_drained_tracks_unflushed_bytes() { + let mut w = SafeWrite::new(64); + assert!(w.is_drained(), "fresh buffer is drained"); + w.write(b"abc").unwrap(); + assert!(!w.is_drained(), "unflushed bytes must block parking"); + w.buffer.as_mut().unwrap().advance(3); + assert!(w.is_drained(), "drained after flush"); + } } diff --git a/vendor/monoio-rustls/src/stream.rs b/vendor/monoio-rustls/src/stream.rs index 978c410c..7115411e 100644 --- a/vendor/monoio-rustls/src/stream.rs +++ b/vendor/monoio-rustls/src/stream.rs @@ -73,6 +73,12 @@ impl Stream { r || w } + /// moon patch (c1M P1-TLS): the underlying transport, for raw-fd + /// readiness watching while the connection is task-parked. + pub fn io_ref(&self) -> &IO { + &self.io + } + pub(crate) fn map_conn C2>(self, f: F) -> Stream { Stream { io: self.io, @@ -83,6 +89,31 @@ impl Stream { } } +impl Stream +where + C: DerefMut + Deref>, +{ + /// moon patch (c1M P1-TLS): true when NOTHING is buffered anywhere in + /// the TLS stack — task-parking on raw-fd readability is only correct + /// then. A buffered TLS record, decrypted plaintext, a pending + /// close_notify, or unflushed/unsent output would all wait forever + /// behind a `readable()` that never fires (or would deadlock a peer + /// waiting on our write). A partial record in the deframer is fine: + /// completing it requires more socket bytes, which fire readability. + pub fn task_park_safe(&mut self) -> bool { + if !self.r_buffer.is_drained() || !self.w_buffer.is_drained() || self.session.wants_write() + { + return false; + } + match self.session.process_new_packets() { + Ok(state) => state.plaintext_bytes_to_read() == 0 && !state.peer_has_closed(), + // Corrupt/unexpected TLS state: don't park; the next read + // surfaces the error and tears down cleanly. + Err(_) => false, + } + } +} + impl Stream where C: DerefMut + Deref>,