From 73733f96b9554eebc5360bfc2fd1be1084d43aab Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Wed, 29 Jul 2026 22:33:45 +0700 Subject: [PATCH 1/3] fix(server): registration handoff across park/wake + migrated-conn parking (c1M P1 follow-ups) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two follow-ups from the round-5 task-exit-parking review: 1. Registration handoff (closes the review's Low finding, and a worse bug found while fixing it): waking a parked connection deregistered then re-registered the client across a task-scheduling boundary. Beyond the transient CLIENT LIST invisibility / KILL-returns-0 window, the fresh registration silently RESET the connection's CLIENT SETNAME and age (register() inserts name: None; only the SETNAME dispatch path ever sets it). The watcher now passes the held RegistryGuard through spawn_resumed_parked_conn into the handler (ParkArgs::kept_registration); the handler reuses the live entry via the new client_registry::live_handle() instead of re-registering. The entry — name, connected_at, kill state, TOTAL_CLIENTS/shard counters — persists unbroken across any number of park/wake cycles; there is no deregistered window at all. The can_park bool param becomes ParkArgs {can_park, kept_registration} with NO_PARK/PARK consts at the six monoio call sites. 2. Migrated-spawn site routes ParkIdle: a connection that migrated shards (Linux) and then idled past --conn-park-secs now parks like a primary-accept connection instead of holding its handler task forever. Mechanical mirror of the reviewed plain-TCP routing (same watcher, same close-accounting transfer). No dedicated migration integration harness exists (migration is affinity-driven, Linux-only); covered by compile gates + the shared watcher paths. Red/green: new tests/parked_idle_parity.rs test resumed_connection_keeps_registry_identity (SETNAME → 2× park/data-wake cycles → name+id survive, CLIENT LIST holds exactly victim+control) was RED before (name= empty, age=0 after wake), GREEN after. Gates: parked parity 4/4 on kqueue AND io_uring (VM); TLS parity green; VM monoio lib 4466 pass; VM tokio lib 3628 pass; fmt; clippy --all-targets both matrices 0 lints. refs: PR #422 review Low, .planning/rfcs/c1m-connection-plane.md author: Tin Dang --- CHANGELOG.md | 19 ++++++ src/client_registry.rs | 8 +++ src/server/conn/handler_monoio/mod.rs | 88 +++++++++++++++++++++------ src/shard/conn_accept.rs | 69 +++++++++++++++------ tests/parked_idle_parity.rs | 56 +++++++++++++++++ 5 files changed, 202 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c4985f22..efc69a04 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed +- **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/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 4efc5f23..7f87f062 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -61,6 +61,34 @@ 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 { + 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 +106,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 +239,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 +276,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,7 +752,7 @@ 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() diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index 864be8d9..8d37b265 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -727,7 +727,7 @@ 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::NO_PARK, // TLS keeps W11+P4b (no task park) ) .await; } @@ -774,7 +774,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 +893,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 +950,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; } @@ -1030,18 +1030,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); } @@ -1055,6 +1056,7 @@ fn spawn_parked_idle_watcher( fn spawn_resumed_parked_conn( stream: monoio::net::TcpStream, mut state: Box, + registry_guard: crate::server::conn::handler_monoio::RegistryGuard, conn_ctx: Box, shutdown: CancellationToken, client_id: u64, @@ -1074,7 +1076,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) { @@ -1265,7 +1272,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 +1285,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..e85d9e8a 100644 --- a/tests/parked_idle_parity.rs +++ b/tests/parked_idle_parity.rs @@ -200,6 +200,62 @@ 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(); +} + /// An active sibling must be undisturbed while its neighbor parks and wakes. #[test] fn active_sibling_undisturbed_by_parking() { From 99f4b871110b367d82cd12a8413c0d9a69ece8f3 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Wed, 29 Jul 2026 23:14:30 +0700 Subject: [PATCH 2/3] =?UTF-8?q?feat(server):=20TLS=20task-exit=20parking?= =?UTF-8?q?=20=E2=80=94=20idle=20TLS=2047.0=20=E2=86=92=2026.0=20KB/conn?= =?UTF-8?q?=20(c1M=20P1-TLS)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends round-5 task-exit parking to TLS connections, closing the last c1M follow-up. No new unsafe; vendored additions are moon-patch style. Vendored (monoio-rustls + monoio-io-wrapper): - Stream::io_ref() — raw-transport reference so the parked watcher can await readable(false) on the underlying fd through the TLS wrapper. - Stream::task_park_safe() — park-safety veto: refuses while the TLS stack holds ANYTHING the raw fd's readability cannot signal (wrapper buffer bytes or pending EOF/error status via the new ReadBuffer/WriteBuffer::is_drained(), decrypted plaintext, a received close_notify, or pending session output). A partial record in the deframer is park-safe: completing it requires more socket bytes. - SafeRead/SafeWrite::is_drained() with moon-patch unit tests (7/7). moon: - IdleParkRead: TLS sets SUPPORTS_TASK_PARK=true; new per-park task_park_safe() hook (plain TCP: always true) appended LAST in the parkable predicate. - conn_accept: ParkWatchable trait (park_readable()) — TcpStream awaits its own readable(false), TLS awaits io_ref().readable(false); watcher + resume helpers genericized over the stream; the monoio-TLS accept site routes ParkIdle like the plain site. Spin bug found by E11 and fixed (affects plain TCP too): - The idle-park arms treated EVERY read error as the sweep cancel. With task parking, a dead connection whose error leaves the fd permanently readable (TLS client FIN without close_notify => read ERROR; plain RST) spun park→wake→park at 100% CPU forever — E11's first ON leg ended with 3000 CLOSE_WAIT conns and a pinned shard. The arms now match monoio's exact cancel error (raw os 125 on BOTH drivers — uring kernel -ECANCELED, legacy hardcodes 125): only the sweep cancel downshifts/parks, real errors tear down promptly (idle_park:: is_sweep_cancel). Small-N repro: fds released, 0 CLOSE_WAIT, 0% CPU. Measured (E11, moon-dev VM, 3000 idle TLS conns, shards=2, same-binary flag A/B, tmp/c10k/e11_tls_park_rss.sh): - TLSPARK_OFF (P4b downshift only): 47.0 KB/conn idle - TLSPARK_ON: 26.0 KB/conn idle (−45%); re-parks at 31.4 after a full wake sweep; sweeps bad=0 both legs; RSS + fds fully return on close - Remaining TLS floor ≈ rustls session (~15-20 KB, not shrinkable via public API) + registry entry + kernel socket Tests: - tls_idle_downshift_parity: +tls_parked_connection_serves_traffic_and_ stays_killable (3 park/wake cycles incl. 100-pipeline on ONE session; CLIENT LIST shows the parked conn WITH its name via registration handoff; CLIENT KILL closes it), +tls_fin_while_parked_tears_down_ promptly (the spin regression, TLS leg) - parked_idle_parity: +rst_while_parked_tears_down_promptly (spin regression, plain-TCP RST leg) - All parity suites green on kqueue AND io_uring (plain 5/5, TLS 3/3) Gates: VM monoio+tokio lib suites pass; E10 plain-park sanity re-run green; fmt; clippy --all-targets both matrices 0 lints; vendor wrapper moon-patch tests 7/7. refs: .planning/rfcs/c1m-connection-plane.md (P1-TLS follow-up) author: Tin Dang --- CHANGELOG.md | 21 +++ src/server/conn/handler_monoio/idle_park.rs | 39 +++- src/server/conn/handler_monoio/mod.rs | 32 +++- src/shard/conn_accept.rs | 114 +++++++++--- tests/parked_idle_parity.rs | 57 ++++++ tests/tls_idle_downshift_parity.rs | 190 ++++++++++++++++++++ vendor/monoio-io-wrapper/src/lib.rs | 23 +++ vendor/monoio-io-wrapper/src/safe_io.rs | 63 +++++++ vendor/monoio-rustls/src/stream.rs | 31 ++++ 9 files changed, 531 insertions(+), 39 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index efc69a04..2b7d611d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,28 @@ 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 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 7f87f062..c1f5a44e 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -79,6 +79,9 @@ pub(crate) struct ParkArgs { #[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, @@ -759,7 +762,11 @@ pub(crate) async fn handle_connection_sharded_monoio< && write_buf.is_empty() && !conn.in_multi && conn.command_queue.is_empty() - && conn.active_cross_txn.is_none(); + && conn.active_cross_txn.is_none() + // 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()); @@ -772,11 +779,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, @@ -824,11 +834,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; diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index 8d37b265..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, - crate::server::conn::handler_monoio::ParkArgs::NO_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!( @@ -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; @@ -1053,15 +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); @@ -1122,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( diff --git a/tests/parked_idle_parity.rs b/tests/parked_idle_parity.rs index e85d9e8a..dead9b8f 100644 --- a/tests/parked_idle_parity.rs +++ b/tests/parked_idle_parity.rs @@ -256,6 +256,63 @@ fn resumed_connection_keeps_registry_identity() { 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, + }; + 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..987f3a4f 100644 --- a/vendor/monoio-io-wrapper/src/safe_io.rs +++ b/vendor/monoio-io-wrapper/src/safe_io.rs @@ -220,6 +220,18 @@ 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. + pub fn is_drained(&self) -> bool { + self.buffer + .as_ref() + .expect("buffer ref expected") + .is_empty() + && matches!(self.status, ReadStatus::Ok) + } } impl io::Read for SafeRead { @@ -323,6 +335,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`. + pub fn is_drained(&self) -> bool { + self.buffer + .as_ref() + .expect("buffer ref expected") + .is_empty() + && matches!(self.status, WriteStatus::Ok) + } } impl io::Write for SafeWrite { @@ -521,4 +543,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>, From 5ba8ccd32816162083b8058fc28f1f359252724c Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Wed, 29 Jul 2026 23:34:36 +0700 Subject: [PATCH 3/3] =?UTF-8?q?fix(server):=20PR=20#424=20review=20fixes?= =?UTF-8?q?=20=E2=80=94=20REPLCONF=20park=20exclusion,=20SAFETY=20comment,?= =?UTF-8?q?=20expect-free=20is=5Fdrained?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit review of PR #424, all three findings addressed: 1. (Major) PSYNC-after-park closed the replica connection: connections that have issued REPLCONF (replica mid-handshake, PSYNC next) are now permanently excluded from task-parking via a sticky ConnectionState::saw_replconf flag checked in the parkable predicate — the unsupported HijackForPsync warn+close arm on the resumed path is now unreachable for real replicas. (A manual PSYNC with no prior REPLCONF after 60s of idle still lands on the loud warn+close and the replica's reconnect loop recovers — documented.) 2. (Minor) SO_LINGER unsafe block in rst_while_parked test gained its required // SAFETY: comment. 3. (Minor) Vendored SafeRead/SafeWrite::is_drained() are expect-free: a transiently-absent buffer answers false ("don't park" is always the safe verdict) instead of panicking. Gates: parity suites green kqueue (plain 5/5, TLS 3/3); vendor wrapper 7/7; fmt; clippy --all-targets both matrices 0 lints. refs: PR #424 review threads author: Tin Dang --- src/server/conn/core.rs | 14 ++++++++++++++ src/server/conn/handler_monoio/mod.rs | 7 +++++++ tests/parked_idle_parity.rs | 3 +++ vendor/monoio-io-wrapper/src/safe_io.rs | 24 +++++++++++++----------- 4 files changed, 37 insertions(+), 11 deletions(-) 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/mod.rs b/src/server/conn/handler_monoio/mod.rs index c1f5a44e..25f7d32f 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -763,6 +763,9 @@ pub(crate) async fn handle_connection_sharded_monoio< && !conn.in_multi && conn.command_queue.is_empty() && 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. @@ -1127,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/tests/parked_idle_parity.rs b/tests/parked_idle_parity.rs index dead9b8f..60aa4918 100644 --- a/tests/parked_idle_parity.rs +++ b/tests/parked_idle_parity.rs @@ -286,6 +286,9 @@ fn rst_while_parked_tears_down_promptly() { 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(), diff --git a/vendor/monoio-io-wrapper/src/safe_io.rs b/vendor/monoio-io-wrapper/src/safe_io.rs index 987f3a4f..a1685eb6 100644 --- a/vendor/monoio-io-wrapper/src/safe_io.rs +++ b/vendor/monoio-io-wrapper/src/safe_io.rs @@ -225,12 +225,14 @@ impl SafeRead { /// 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 { - self.buffer - .as_ref() - .expect("buffer ref expected") - .is_empty() - && matches!(self.status, ReadStatus::Ok) + match (self.buffer.as_ref(), &self.status) { + (Some(buffer), ReadStatus::Ok) => buffer.is_empty(), + _ => false, + } } } @@ -337,13 +339,13 @@ impl SafeWrite { } /// moon patch (c1M P1-TLS): true when no unflushed bytes and no stashed - /// write error are pending. See `SafeRead::is_drained`. + /// write error are pending. See `SafeRead::is_drained` (incl. the + /// `None`-buffer rationale). pub fn is_drained(&self) -> bool { - self.buffer - .as_ref() - .expect("buffer ref expected") - .is_empty() - && matches!(self.status, WriteStatus::Ok) + match (self.buffer.as_ref(), &self.status) { + (Some(buffer), WriteStatus::Ok) => buffer.is_empty(), + _ => false, + } } }