From b835a43e61b0843043bdf0ea52d45500a1a30854 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 00:33:48 +0700 Subject: [PATCH 1/6] fix(server): connection-plane robustness hardening (Track B) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deep review of Moon's connection/event-loop plane surfaced four robustness gaps that bite at high parallel connection counts (1K-10K clients). All four are fixed here, each with red/green tests; both runtimes build, full lib suites pass (tokio 3115, monoio 3770), clippy/fmt clean. R-2 — inline-command length cap (src/protocol/inline.rs, frame.rs) The RESP-less inline path had no maximum line length, so a client that never sends CRLF (raw non-RESP bytes) grew the connection read buffer without bound — a per-connection memory-exhaustion vector. Added ParseConfig::max_inline_size (default 64 KB, mirrors Redis PROTO_INLINE_MAX_SIZE); parse_inline now rejects an over-cap line (complete or incomplete) with "Protocol error: too big inline request". R-1 — bounded cross-shard spsc_send (src/shard/coordinator.rs) The single dispatch primitive behind every scatter-gather command (MGET/MSET/DEL/EXISTS/SCAN/KEYS/DBSIZE, vector scatter, FT.INFO, graph traverse) retried a full SPSC ring in an UNBOUNDED loop { try_push; yield/sleep } — on tokio yield_now() reschedules with no backoff, busy-spinning a full core, and a wedged target could block graceful shutdown forever. Replaced with a bounded retry (CROSS_SHARD_PUSH_MAX_RETRIES x CROSS_SHARD_PUSH_BACKOFF, same budget as push_with_backpressure) returning PushOutcome. On give-up the message is dropped; reply-carrying callers observe the closed reply channel and synthesize a per-shard error via the path they already handle — no call-site signature change. R-4 — accept-loop backoff on fd exhaustion (src/server/accept_backoff.rs, listener.rs, shard/event_loop.rs) All six socket-accept loops (tokio and monoio; sharded, non-sharded, TLS) retried accept() errors with zero backoff, so EMFILE/ENFILE under a connection storm pinned a shard core at 100% and flooded the log. Added AcceptBackoff: capped exponential backoff (1ms -> 1s) on resource-exhaustion errors only, plus rate-limited logging; benign per-connection errors (ECONNABORTED) log without sleeping. The io_uring multishot-accept resubmit (opt-in MOON_URING=1 bridge) is documented as a scoped follow-up (synchronous CQE handler, no async context to sleep in). R-3 — CLIENT KILL force-closes idle connections (src/client_registry.rs, handlers, conn_accept.rs) CLIENT KILL only set a cooperative kill_flag checked once per batch, so a connection parked in read().await (idle, default timeout 0) was never torn down until it next sent bytes. The registry now stores each connection's socket fd and kill_clients also shutdown(2)s it, so the parked read returns Ok(0)/Err immediately (existing disconnect path). The raw-fd close is race-free: kill_clients holds the registry read lock, and a connection's RegistryGuard deregisters (needs the write lock) strictly before its stream drops — so a visible entry always has a live fd. No-op on non-unix. Review report: tmp/CONN-DEEP-REVIEW.md (full 5-dimension findings, incl. Tier-0 multi-tenant isolation and Tier-2 head-of-line-blocking tracks not in this PR). author: Tin Dang --- .gitignore | 1 + CHANGELOG.md | 41 +++++ fuzz/fuzz_targets/inline_parse.rs | 2 +- fuzz/fuzz_targets/resp_parse.rs | 1 + fuzz/fuzz_targets/resp_parse_differential.rs | 1 + src/client_registry.rs | 97 +++++++++- src/protocol/frame.rs | 11 ++ src/protocol/inline.rs | 94 +++++++++- src/protocol/parse.rs | 4 +- src/server/accept_backoff.rs | 135 ++++++++++++++ src/server/conn/handler_monoio/mod.rs | 5 + src/server/conn/handler_sharded/mod.rs | 15 ++ src/server/listener.rs | 24 ++- src/server/mod.rs | 1 + src/shard/conn_accept.rs | 39 ++++ src/shard/coordinator.rs | 180 +++++++++++++++++-- src/shard/event_loop.rs | 20 ++- src/shard/uring_handler.rs | 10 +- 18 files changed, 636 insertions(+), 45 deletions(-) create mode 100644 src/server/accept_backoff.rs diff --git a/.gitignore b/.gitignore index d0b4eb79..5437bc97 100644 --- a/.gitignore +++ b/.gitignore @@ -104,3 +104,4 @@ tmp/ /target-check-tokio/ /target-check-monoio/ /target-check/ +target-fast/ diff --git a/CHANGELOG.md b/CHANGELOG.md index e6c6e232..faaa6f37 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -132,6 +132,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 shared `READY_TIMEOUT` constant (poll-based loop notices readiness within 100ms either way; the fixed 15s was a documented host-load flake, observed on the macOS CI runner). +### Fixed — connection-plane robustness hardening (Track B) (PR #230) + +- **R-3 — `CLIENT KILL` force-closes idle connections** (`src/client_registry.rs`, + handlers, `conn_accept.rs`): `CLIENT KILL` only set a cooperative `kill_flag` + checked once per batch, so a connection parked in `read().await` (idle, the + default `timeout 0`) was never torn down until it next sent bytes. Now the + registry stores each connection's socket fd and `kill_clients` also + `shutdown(2)`s it, so the parked read returns `Ok(0)`/`Err` immediately (the + existing disconnect path). The raw-fd close is race-free: `kill_clients` holds + the registry read lock, and a connection's `RegistryGuard` deregisters (needs + the write lock) strictly before its stream drops — so a visible entry always + has a live fd. No-op on non-unix (`CLIENT KILL` stays cooperative there). +- **R-2 — inline-command length cap** (`src/protocol/inline.rs`, + `frame.rs`): the RESP-less inline path had no maximum line length, so a + client that never sends `\r\n` (raw non-RESP bytes) grew the connection read + buffer without bound — a per-connection memory-exhaustion vector. Added + `ParseConfig::max_inline_size` (default 64 KB, mirroring Redis's + `PROTO_INLINE_MAX_SIZE`); `parse_inline` now rejects an over-cap line + (complete or incomplete) with `Protocol error: too big inline request` + instead of buffering forever. +- **R-1 — bounded cross-shard `spsc_send`** (`src/shard/coordinator.rs`): the + single dispatch primitive behind every scatter-gather command (MGET/MSET/ + DEL/EXISTS/SCAN/KEYS/DBSIZE, vector scatter, FT.INFO, graph traverse) retried + a full SPSC ring in an **unbounded** `loop { try_push; yield/sleep }` — on + tokio `yield_now()` reschedules with no backoff, busy-spinning a full core, + and a wedged target could block graceful shutdown forever. Replaced with a + bounded retry (`CROSS_SHARD_PUSH_MAX_RETRIES` × `CROSS_SHARD_PUSH_BACKOFF`, + the same budget as `push_with_backpressure`) returning `PushOutcome`. On + give-up the message is dropped; reply-carrying callers observe the closed + reply channel and synthesize a per-shard error via the path they already + handle. +- **R-4 — accept-loop backoff on fd exhaustion** (`src/server/accept_backoff.rs`, + `listener.rs`, `shard/event_loop.rs`): all six socket-accept loops (tokio + and monoio; sharded, non-sharded, and TLS) retried `accept()` errors with + zero backoff, so `EMFILE`/`ENFILE` under a connection storm pinned a shard + core at 100% and flooded the log. Added `AcceptBackoff`: capped exponential + backoff (1 ms → 1 s) on resource-exhaustion errors only, plus rate-limited + logging; benign per-connection errors (e.g. `ECONNABORTED`) log without + sleeping. The io_uring multishot-accept resubmit (opt-in `MOON_URING=1` + bridge) is documented as a scoped follow-up (synchronous CQE handler, no + async context to sleep in). ### Changed — consolidated dependency bumps (PR #TBD) diff --git a/fuzz/fuzz_targets/inline_parse.rs b/fuzz/fuzz_targets/inline_parse.rs index ed3aa78c..b96a6e13 100644 --- a/fuzz/fuzz_targets/inline_parse.rs +++ b/fuzz/fuzz_targets/inline_parse.rs @@ -10,5 +10,5 @@ use moon::protocol::inline; /// splits on whitespace, handles quoted strings, looks for CRLF termination. fuzz_target!(|data: &[u8]| { let mut buf = BytesMut::from(data); - let _ = inline::parse_inline(&mut buf); + let _ = inline::parse_inline(&mut buf, 64 * 1024); }); diff --git a/fuzz/fuzz_targets/resp_parse.rs b/fuzz/fuzz_targets/resp_parse.rs index a25da1c6..b9b29338 100644 --- a/fuzz/fuzz_targets/resp_parse.rs +++ b/fuzz/fuzz_targets/resp_parse.rs @@ -16,6 +16,7 @@ fuzz_target!(|data: &[u8]| { max_bulk_string_size: 64 * 1024, // 64 KB — enough to exercise bulk paths max_array_depth: 4, // Prevent stack overflow from nesting max_array_length: 256, // Bound allocation from large arrays + max_inline_size: 64 * 1024, // Bound inline-line growth }; let mut buf = BytesMut::from(data); diff --git a/fuzz/fuzz_targets/resp_parse_differential.rs b/fuzz/fuzz_targets/resp_parse_differential.rs index 2a95334a..5cae1171 100644 --- a/fuzz/fuzz_targets/resp_parse_differential.rs +++ b/fuzz/fuzz_targets/resp_parse_differential.rs @@ -14,6 +14,7 @@ fuzz_target!(|data: &[u8]| { max_bulk_string_size: 64 * 1024, max_array_depth: 4, max_array_length: 256, + max_inline_size: 64 * 1024, }; let mut buf1 = BytesMut::from(data); diff --git a/src/client_registry.rs b/src/client_registry.rs index 65b2a5ea..a7ab8242 100644 --- a/src/client_registry.rs +++ b/src/client_registry.rs @@ -31,6 +31,13 @@ pub struct ClientLiveState { pub flags: AtomicU8, /// Set by CLIENT KILL — the handler checks this and closes the connection. pub kill_flag: AtomicBool, + /// Raw socket fd for out-of-band force-close (R-1/R-3). `CLIENT KILL` sets + /// `kill_flag` (cooperative) AND `shutdown(2)`s this fd, so a connection + /// parked in `read().await` (idle, `timeout 0`) is torn down immediately + /// instead of waiting until it next sends bytes. `-1` = no fd (tests / + /// non-unix). Set once at registration; only read by `kill_clients` while + /// holding the registry lock (see the drop-ordering safety note there). + pub kill_fd: i32, } impl ClientLiveState { @@ -110,7 +117,13 @@ impl ClientFlags { /// /// Returns the connection's lock-free live-state handle; the connection task /// keeps it for per-batch `touch()` and `is_killed()` without the registry lock. -pub fn register(id: u64, addr: String, user: String, shard: usize) -> Arc { +pub fn register( + id: u64, + addr: String, + user: String, + shard: usize, + kill_fd: i32, +) -> Arc { let live = Arc::new(ClientLiveState { connected_at: Instant::now(), connected_at_epoch_ms: crate::storage::entry::current_time_ms(), @@ -118,6 +131,7 @@ pub fn register(id: u64, addr: String, user: String, shard: usize) -> Arc Option { } /// Kill clients matching the given filter. Returns count of killed clients. +/// +/// Sets the cooperative `kill_flag` (checked once per batch by the handler) +/// AND force-closes the socket via `shutdown(2)`, so a connection currently +/// parked in `read().await` — idle with `timeout 0` — is torn down at once +/// rather than lingering until it next sends bytes. The parked read returns +/// `Ok(0)`/`Err`, which the handler already treats as disconnect. pub fn kill_clients(filter: &KillFilter) -> u64 { + // Hold the read lock across the whole loop. This is what makes the raw-fd + // `shutdown` free of a use-after-reuse race: a connection's `RegistryGuard` + // (a local) drops — and calls `deregister`, which needs the registry WRITE + // lock — strictly before its `stream` (a parameter) drops and closes the + // fd. So while we hold the READ lock and an entry is present, `deregister` + // for it is blocked, its `stream` has not dropped, and `kill_fd` is still + // an open socket. Once an entry is gone, we never touch its fd. let registry = REGISTRY.read(); let mut count = 0u64; for entry in registry.values() { @@ -197,12 +224,32 @@ pub fn kill_clients(filter: &KillFilter) -> u64 { }; if matches { entry.live.kill_flag.store(true, Ordering::Relaxed); + force_close_fd(entry.live.kill_fd); count += 1; } } count } +/// `shutdown(SHUT_RDWR)` the given socket fd to wake a parked `read`. No-op for +/// `-1` (no fd) and on non-unix targets (CLIENT KILL stays cooperative there). +#[inline] +fn force_close_fd(fd: i32) { + #[cfg(unix)] + if fd >= 0 { + // SAFETY: `libc::shutdown` is an FFI call that cannot cause memory + // unsafety for any integer argument — an invalid fd merely returns + // `EBADF`. Per the lock-ordering note in `kill_clients`, `fd` is a live + // socket for the duration of this call. We intentionally ignore the + // return value: a already-closed/half-closed socket is a benign no-op. + unsafe { + libc::shutdown(fd, libc::SHUT_RDWR); + } + } + #[cfg(not(unix))] + let _ = fd; +} + /// Filter for CLIENT KILL. pub enum KillFilter { Id(u64), @@ -269,7 +316,7 @@ mod tests { #[test] fn test_register_and_list() { let id = 999_000; - register(id, "127.0.0.1:12345".into(), "default".into(), 0); + register(id, "127.0.0.1:12345".into(), "default".into(), 0, -1); let list = client_list(); assert!(list.contains("id=999000")); assert!(list.contains("addr=127.0.0.1:12345")); @@ -282,7 +329,7 @@ mod tests { #[test] fn test_client_info() { let id = 999_001; - register(id, "10.0.0.1:5000".into(), "alice".into(), 1); + register(id, "10.0.0.1:5000".into(), "alice".into(), 1, -1); let info = client_info(id); assert!(info.is_some()); assert!(info.as_ref().is_some_and(|s| s.contains("user=alice"))); @@ -293,7 +340,7 @@ mod tests { #[test] fn test_kill_by_id() { let id = 999_002; - let live = register(id, "10.0.0.2:6000".into(), "bob".into(), 0); + let live = register(id, "10.0.0.2:6000".into(), "bob".into(), 0, -1); assert!(!is_killed(id)); assert!(!live.is_killed()); let count = kill_clients(&KillFilter::Id(id)); @@ -303,12 +350,46 @@ mod tests { deregister(id); } + // R-3: CLIENT KILL must force-close the socket so a connection blocked in + // read() (idle, `timeout 0`) is torn down immediately, not only on its next + // byte. We register one end of a socketpair and assert the peer observes EOF + // after the kill — proving the fd was actually shut down. + #[cfg(unix)] + #[test] + fn test_kill_force_closes_socket_fd() { + use std::io::Read; + use std::os::unix::io::AsRawFd; + use std::os::unix::net::UnixStream; + + let (killed_end, mut peer) = UnixStream::pair().expect("socketpair"); + let id = 999_020; + let live = register( + id, + "unix:socketpair".into(), + "default".into(), + 0, + killed_end.as_raw_fd(), + ); + + let count = kill_clients(&KillFilter::Id(id)); + assert_eq!(count, 1); + assert!(live.is_killed()); + + // After shutdown(SHUT_RDWR) on killed_end, the peer's read returns 0 (EOF). + let mut buf = [0u8; 8]; + let n = peer.read(&mut buf).expect("peer read after kill"); + assert_eq!(n, 0, "peer must see EOF once the killed fd is shut down"); + + deregister(id); + drop(killed_end); + } + #[test] fn test_kill_by_user() { let id1 = 999_010; let id2 = 999_011; - register(id1, "10.0.0.3:7000".into(), "eve".into(), 0); - register(id2, "10.0.0.4:7001".into(), "eve".into(), 1); + register(id1, "10.0.0.3:7000".into(), "eve".into(), 0, -1); + register(id2, "10.0.0.4:7001".into(), "eve".into(), 1, -1); let count = kill_clients(&KillFilter::User("eve".into())); assert_eq!(count, 2); assert!(is_killed(id1)); @@ -320,7 +401,7 @@ mod tests { #[test] fn test_update_and_touch() { let id = 999_003; - let live = register(id, "10.0.0.5:8000".into(), "default".into(), 0); + let live = register(id, "10.0.0.5:8000".into(), "default".into(), 0, -1); update(id, |e| { e.name = Some("myconn".into()); }); @@ -334,7 +415,7 @@ mod tests { #[test] fn test_touch_uses_caller_clock_not_instant_now() { let id = 999_004; - let live = register(id, "10.0.0.6:9000".into(), "default".into(), 0); + let live = register(id, "10.0.0.6:9000".into(), "default".into(), 0, -1); // touch takes the caller's (shard-cached) epoch clock — no per-op // Instant::now(). last_cmd_ms stays "ms since connect". live.touch(2, ClientFlags::default(), live.connected_at_epoch_ms + 5000); diff --git a/src/protocol/frame.rs b/src/protocol/frame.rs index dfd21b74..0c9e9d22 100644 --- a/src/protocol/frame.rs +++ b/src/protocol/frame.rs @@ -109,6 +109,14 @@ pub const DEFAULT_MAX_ARRAY_DEPTH: usize = 8; /// Default maximum number of elements in an array. pub const DEFAULT_MAX_ARRAY_LENGTH: usize = 1024 * 1024; +/// Default maximum size for a single inline command line (64 KB). +/// +/// Mirrors Redis's `PROTO_INLINE_MAX_SIZE`. Bounds the RESP-less inline path, +/// which the framed-parser limits above do not cover: without this, a client +/// that never sends a `\r\n` (raw non-RESP bytes) grows the read buffer without +/// limit — a per-connection memory-exhaustion vector. +pub const DEFAULT_MAX_INLINE_SIZE: usize = 64 * 1024; + /// A RESP2/RESP3 protocol frame. /// /// All string payloads use `Bytes` for zero-copy semantics. @@ -237,6 +245,8 @@ pub struct ParseConfig { pub max_array_depth: usize, /// Maximum number of elements in a single array. pub max_array_length: usize, + /// Maximum size in bytes of a single inline (non-RESP) command line. + pub max_inline_size: usize, } impl Default for ParseConfig { @@ -245,6 +255,7 @@ impl Default for ParseConfig { max_bulk_string_size: DEFAULT_MAX_BULK_STRING_SIZE, max_array_depth: DEFAULT_MAX_ARRAY_DEPTH, max_array_length: DEFAULT_MAX_ARRAY_LENGTH, + max_inline_size: DEFAULT_MAX_INLINE_SIZE, } } } diff --git a/src/protocol/inline.rs b/src/protocol/inline.rs index 3f548db9..de7353e5 100644 --- a/src/protocol/inline.rs +++ b/src/protocol/inline.rs @@ -13,13 +13,42 @@ use super::frame::{Frame, FrameVec, ParseError}; /// Returns `Ok(Some(Frame::Array(...)))` with each argument as a `BulkString`, /// `Ok(None)` if the buffer doesn't contain a complete line (no `\r\n` found), /// or `Ok(None)` for empty/whitespace-only lines (after advancing past the CRLF). -pub fn parse_inline(buf: &mut BytesMut) -> Result, ParseError> { +/// +/// `max_inline_size` bounds the length of a single inline line. If the buffer +/// grows past it without a terminating `\r\n`, the line is rejected with a +/// protocol error instead of returning `Ok(None)` forever — this is what stops +/// a client that never sends `\r\n` from growing the read buffer without limit +/// (mirrors Redis's `PROTO_INLINE_MAX_SIZE`). +pub fn parse_inline( + buf: &mut BytesMut, + max_inline_size: usize, +) -> Result, ParseError> { // Find the CRLF terminator let crlf_pos = match find_crlf_position(&buf[..]) { Some(pos) => pos, - None => return Ok(None), // Incomplete -- need more data + None => { + // No complete line yet. Reject before the buffer can grow unbounded: + // a non-RESP stream with no `\r\n` would otherwise "need more data" + // forever. Redis caps this at PROTO_INLINE_MAX_SIZE. + if buf.len() > max_inline_size { + return Err(ParseError::Invalid { + message: "Protocol error: too big inline request".into(), + offset: 0, + }); + } + return Ok(None); // Incomplete -- need more data + } }; + // Reject an oversize completed line too (parity with Redis, which rejects + // any inline request past the cap regardless of termination). + if crlf_pos > max_inline_size { + return Err(ParseError::Invalid { + message: "Protocol error: too big inline request".into(), + offset: 0, + }); + } + // Extract line content before CRLF let line = &buf[..crlf_pos]; @@ -89,9 +118,13 @@ mod tests { use super::*; use crate::framevec; + // Generous cap for the behavioural tests below; the cap itself is exercised + // by the dedicated tests at the end of this module. + const TEST_MAX_INLINE: usize = 64 * 1024; + fn parse_inline_bytes(input: &[u8]) -> Result, ParseError> { let mut buf = BytesMut::from(input); - parse_inline(&mut buf) + parse_inline(&mut buf, TEST_MAX_INLINE) } #[test] @@ -150,7 +183,7 @@ mod tests { #[test] fn test_parse_inline_sequential() { let mut buf = BytesMut::from(&b"GET key\r\nPING\r\n"[..]); - let frame1 = parse_inline(&mut buf).unwrap().unwrap(); + let frame1 = parse_inline(&mut buf, TEST_MAX_INLINE).unwrap().unwrap(); assert_eq!( frame1, Frame::Array(framevec![ @@ -158,7 +191,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"key")), ]) ); - let frame2 = parse_inline(&mut buf).unwrap().unwrap(); + let frame2 = parse_inline(&mut buf, TEST_MAX_INLINE).unwrap().unwrap(); assert_eq!( frame2, Frame::Array(framevec![Frame::BulkString(Bytes::from_static(b"PING"))]) @@ -190,15 +223,62 @@ mod tests { #[test] fn test_parse_inline_buffer_consumed() { let mut buf = BytesMut::from(&b"PING\r\nremaining"[..]); - let _ = parse_inline(&mut buf).unwrap().unwrap(); + let _ = parse_inline(&mut buf, TEST_MAX_INLINE).unwrap().unwrap(); assert_eq!(&buf[..], b"remaining"); } #[test] fn test_parse_inline_empty_line_buffer_consumed() { let mut buf = BytesMut::from(&b"\r\nPING\r\n"[..]); - let result = parse_inline(&mut buf).unwrap(); + let result = parse_inline(&mut buf, TEST_MAX_INLINE).unwrap(); assert!(result.is_none()); assert_eq!(&buf[..], b"PING\r\n"); } + + #[test] + fn test_parse_inline_incomplete_under_cap_needs_more_data() { + // A partial line shorter than the cap is still "need more data", not an error. + let mut buf = BytesMut::from(&b"PARTIAL WITHOUT TERMINATOR"[..]); + let result = parse_inline(&mut buf, TEST_MAX_INLINE).unwrap(); + assert!(result.is_none()); + // Buffer is untouched so the caller can append more bytes. + assert_eq!(&buf[..], b"PARTIAL WITHOUT TERMINATOR"); + } + + #[test] + fn test_parse_inline_incomplete_over_cap_rejected() { + // No CRLF and length exceeds the cap -> reject instead of Ok(None) forever. + // This is the memory-exhaustion guard: without it the read buffer grows + // without bound for a client that never sends `\r\n`. + let cap = 16; + let mut buf = BytesMut::from(&b"AAAAAAAAAAAAAAAAAAAAAAAAAAAA"[..]); // 28 bytes, no CRLF + let err = parse_inline(&mut buf, cap).unwrap_err(); + match err { + ParseError::Invalid { message, .. } => { + assert!(message.contains("too big inline request"), "got: {message}"); + } + other => panic!("expected Invalid, got {other:?}"), + } + } + + #[test] + fn test_parse_inline_complete_over_cap_rejected() { + // A terminated line that is itself larger than the cap is also rejected. + let cap = 8; + let mut buf = BytesMut::from(&b"THIS LINE IS WAY TOO LONG\r\n"[..]); + let err = parse_inline(&mut buf, cap).unwrap_err(); + assert!(matches!(err, ParseError::Invalid { .. })); + } + + #[test] + fn test_parse_inline_exactly_at_cap_ok() { + // A complete line of exactly the cap length is accepted (boundary is inclusive). + let cap = 4; + let mut buf = BytesMut::from(&b"PING\r\n"[..]); // line content "PING" == 4 bytes + let frame = parse_inline(&mut buf, cap).unwrap().unwrap(); + assert_eq!( + frame, + Frame::Array(framevec![Frame::BulkString(Bytes::from_static(b"PING"))]) + ); + } } diff --git a/src/protocol/parse.rs b/src/protocol/parse.rs index e62a53ec..f53a4f7f 100644 --- a/src/protocol/parse.rs +++ b/src/protocol/parse.rs @@ -26,7 +26,7 @@ pub fn parse(buf: &mut BytesMut, config: &ParseConfig) -> Result, b'+' | b'-' | b':' | b'$' | b'*' // RESP2 | b'%' | b'~' | b',' | b'#' | b'_' | b'=' | b'(' | b'>' // RESP3 => { /* fall through to RESP parsing below */ } - _ => return inline::parse_inline(buf), + _ => return inline::parse_inline(buf, config.max_inline_size), } // Pass 1: Validate structure and compute total byte length (zero allocations) @@ -1212,6 +1212,7 @@ mod tests { max_bulk_string_size: 64 * 1024, max_array_depth: 4, max_array_length: 256, + max_inline_size: 64 * 1024, }; let mut buf = BytesMut::from(data); // Must not panic — any combination of Ok/Err is acceptable @@ -1278,6 +1279,7 @@ mod tests { max_bulk_string_size: 64 * 1024, max_array_depth: 4, max_array_length: 256, + max_inline_size: 64 * 1024, }; for _ in 0..16 { if buf.is_empty() { diff --git a/src/server/accept_backoff.rs b/src/server/accept_backoff.rs new file mode 100644 index 00000000..87d38903 --- /dev/null +++ b/src/server/accept_backoff.rs @@ -0,0 +1,135 @@ +//! Bounded backoff for connection-accept loops (R-4, design-for-failure). +//! +//! Every accept loop previously retried immediately on error: +//! `Err(e) => { error!("Accept error: {e}"); }` falling straight back into the +//! loop. On `EMFILE`/`ENFILE` (per-process / system-wide fd exhaustion), +//! `accept()` returns an error *synchronously* rather than pending on +//! readiness, so the loop becomes a 100%-of-one-core spin plus one log line per +//! iteration — worsening the very fd/memory pressure that caused it. This is +//! exactly the connection-storm regime (`ulimit -n` exhaustion) documented in +//! the project's own gotchas. +//! +//! [`AcceptBackoff`] rate-limits the error log and, for resource-exhaustion +//! errors only, sleeps a capped exponential backoff (1ms → 1s). Benign +//! per-connection errors (a client that reset mid-handshake, `ECONNABORTED`) +//! are logged but not slept on, so normal accept throughput is unaffected. +//! Call [`AcceptBackoff::reset`] after every successful `accept()`. + +use std::time::Duration; + +/// Ceiling for the exponential backoff (1 second). +const MAX_BACKOFF_MS: u64 = 1000; + +/// Per-accept-loop backoff + log rate-limiter. One per accept loop; reset on +/// every successful accept so a transient error never permanently slows the +/// loop. +pub(crate) struct AcceptBackoff { + consecutive: u32, +} + +impl AcceptBackoff { + pub(crate) const fn new() -> Self { + Self { consecutive: 0 } + } + + /// Call after a successful `accept()` to clear the error streak. + #[inline] + pub(crate) fn reset(&mut self) { + self.consecutive = 0; + } + + /// Record an accept error: log it (rate-limited) and, when it indicates + /// resource exhaustion, sleep a capped exponential backoff so the loop + /// cannot hot-spin. `context` labels the loop (e.g. `"Accept error"`). + pub(crate) async fn record_error(&mut self, context: &str, err: &std::io::Error) { + self.consecutive = self.consecutive.saturating_add(1); + let exhausted = is_resource_exhaustion(err); + + // Rate-limit: always log the first few, then only at power-of-two + // counts, so a sustained storm cannot flood the log. + if self.consecutive <= 3 || self.consecutive.is_power_of_two() { + if exhausted { + tracing::error!( + "{context}: {err} (resource exhaustion; consecutive={}, backing off)", + self.consecutive + ); + } else { + tracing::error!("{context}: {err} (consecutive={})", self.consecutive); + } + } + + if exhausted { + let delay = backoff_for(self.consecutive); + // Use each runtime's native timer directly rather than the boxed + // `TimerImpl::sleep` (which is `!Send`): the tokio TLS listener runs + // under `tokio::spawn`, which requires a `Send` future. + #[cfg(feature = "runtime-tokio")] + tokio::time::sleep(delay).await; + #[cfg(feature = "runtime-monoio")] + monoio::time::sleep(delay).await; + } + } +} + +/// Capped exponential backoff: 1ms, 2ms, 4ms, … saturating at [`MAX_BACKOFF_MS`]. +/// `consecutive` is 1-based (the count returned by the first error is 1). +fn backoff_for(consecutive: u32) -> Duration { + let shift = consecutive.saturating_sub(1).min(20); + let ms = (1u64 << shift).min(MAX_BACKOFF_MS); + Duration::from_millis(ms.max(1)) +} + +/// Errors that indicate the process/system is out of a resource needed to +/// accept — retrying immediately would just spin. Everything else is treated +/// as a transient per-connection error (logged, loop continues, no sleep). +fn is_resource_exhaustion(err: &std::io::Error) -> bool { + matches!( + err.raw_os_error(), + Some(libc::EMFILE) | Some(libc::ENFILE) | Some(libc::ENOBUFS) | Some(libc::ENOMEM) + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn backoff_grows_exponentially_then_caps() { + assert_eq!(backoff_for(1), Duration::from_millis(1)); + assert_eq!(backoff_for(2), Duration::from_millis(2)); + assert_eq!(backoff_for(3), Duration::from_millis(4)); + assert_eq!(backoff_for(4), Duration::from_millis(8)); + // Caps at 1s and stays there regardless of streak length. + assert_eq!(backoff_for(11), Duration::from_millis(1000)); + assert_eq!(backoff_for(1000), Duration::from_millis(1000)); + } + + #[test] + fn backoff_never_zero() { + // Even a bogus 0 count yields a non-zero sleep (never a busy 0-sleep). + assert!(backoff_for(0) >= Duration::from_millis(1)); + } + + #[test] + fn reset_clears_streak() { + let mut b = AcceptBackoff::new(); + b.consecutive = 7; + b.reset(); + assert_eq!(b.consecutive, 0); + } + + #[test] + fn resource_exhaustion_classification() { + let emfile = std::io::Error::from_raw_os_error(libc::EMFILE); + let enfile = std::io::Error::from_raw_os_error(libc::ENFILE); + assert!(is_resource_exhaustion(&emfile)); + assert!(is_resource_exhaustion(&enfile)); + + // A reset-by-peer during accept is transient, not exhaustion. + let aborted = std::io::Error::from_raw_os_error(libc::ECONNABORTED); + assert!(!is_resource_exhaustion(&aborted)); + // A non-OS error (e.g. synthesized) is not classified as exhaustion. + let other = std::io::Error::new(std::io::ErrorKind::Other, "synthetic"); + assert!(!is_resource_exhaustion(&other)); + } +} diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 71965798..103979b1 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -119,6 +119,10 @@ pub(crate) async fn handle_connection_sharded_monoio< // and the event loop still sweeps it every iteration. _pending_wakers: Rc>>, migrated_state: Option<&MigratedConnectionState>, + // Raw socket fd for CLIENT KILL force-close (R-3), or -1 if unavailable + // (non-unix). Threaded from the concrete spawn site; the generic `S` here + // has no `AsRawFd` bound. + kill_fd: i32, ) -> (MonoioHandlerResult, Option) { use monoio::io::AsyncWriteRentExt; @@ -159,6 +163,7 @@ pub(crate) async fn handle_connection_sharded_monoio< peer_addr.clone(), conn.current_user.clone(), ctx.shard_id, + kill_fd, ); struct RegistryGuard(u64); impl Drop for RegistryGuard { diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index b3674498..ae5330f5 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -91,6 +91,15 @@ pub(crate) async fn handle_connection_sharded( .peer_addr() .map(|a| a.to_string()) .unwrap_or_else(|_| "unknown".to_string()); + // R-3: capture the socket fd for CLIENT KILL force-close before the stream + // is moved into the generic inner handler (which has no AsRawFd bound). + #[cfg(unix)] + let kill_fd = { + use std::os::unix::io::AsRawFd; + stream.as_raw_fd() + }; + #[cfg(not(unix))] + let kill_fd = -1; let result = handle_connection_sharded_inner( stream, peer_addr, @@ -105,6 +114,7 @@ pub(crate) async fn handle_connection_sharded( cfg!(unix), BytesMut::new(), None, // fresh connection, no migrated state + kill_fd, ) .await; @@ -228,6 +238,10 @@ pub(crate) async fn handle_connection_sharded_inner< can_migrate: bool, initial_read_buf: BytesMut, migrated_state: Option<&MigratedConnectionState>, + // Raw socket fd for CLIENT KILL force-close (R-3), or -1 if unavailable + // (non-unix). Threaded from the concrete spawn site; the generic `S` here + // has no `AsRawFd` bound. + kill_fd: i32, ) -> (HandlerResult, Option) { use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -267,6 +281,7 @@ pub(crate) async fn handle_connection_sharded_inner< peer_addr.clone(), conn.current_user.clone(), ctx.shard_id, + kill_fd, ); struct RegistryGuard(u64); impl Drop for RegistryGuard { diff --git a/src/server/listener.rs b/src/server/listener.rs index c2a9ea33..1def4daf 100644 --- a/src/server/listener.rs +++ b/src/server/listener.rs @@ -215,11 +215,13 @@ pub async fn run_with_shutdown( let graph_store: Arc> = Arc::new(Mutex::new(crate::graph::store::GraphStore::new())); + let mut accept_backoff = crate::server::accept_backoff::AcceptBackoff::new(); loop { tokio::select! { result = listener.accept() => { match result { Ok((mut stream, addr)) => { + accept_backoff.reset(); // Protected mode: reject non-loopback connections when no auth configured if config.protected_mode == "yes" && config.requirepass.is_none() @@ -274,7 +276,7 @@ pub async fn run_with_shutdown( )); } Err(e) => { - error!("Accept error: {}", e); + accept_backoff.record_error("Accept error", &e).await; } } } @@ -364,11 +366,13 @@ pub async fn run_sharded( tokio::spawn(async move { let mut tls_next_shard: usize = 0; let tls_num_shards = tls_txs.len(); + let mut accept_backoff = crate::server::accept_backoff::AcceptBackoff::new(); loop { tokio::select! { result = tls_listener.accept() => { match result { Ok((mut stream, addr)) => { + accept_backoff.reset(); // Protected mode: reject non-loopback connections when no auth configured if protected_mode_active && !addr.ip().is_loopback() { tracing::warn!( @@ -387,7 +391,7 @@ pub async fn run_sharded( tls_next_shard = (tls_next_shard + 1) % tls_num_shards; } Err(e) => { - error!("TLS accept error: {}", e); + accept_backoff.record_error("TLS accept error", &e).await; } } } @@ -403,6 +407,7 @@ pub async fn run_sharded( // When per_shard_accept is true (Linux production with SO_REUSEPORT), each shard // has its own listener and the central listener only handles TLS. // When false (tests, non-Linux), the central listener distributes via round-robin MPSC. + let mut accept_backoff = crate::server::accept_backoff::AcceptBackoff::new(); loop { tokio::select! { // Plain TCP accept -- disabled when per-shard SO_REUSEPORT listeners are active. @@ -415,6 +420,7 @@ pub async fn run_sharded( } => { match result { Ok((mut stream, addr)) => { + accept_backoff.reset(); // Protected mode: reject non-loopback connections when no auth configured if protected_mode_active && !addr.ip().is_loopback() { tracing::warn!( @@ -450,7 +456,9 @@ pub async fn run_sharded( } } Err(e) => { - error!("Accept error: {}", e); + // R-4: capped backoff + rate-limited log so fd exhaustion + // can't hot-spin this accept loop. + accept_backoff.record_error("Accept error", &e).await; } } } @@ -510,6 +518,7 @@ pub async fn run_sharded( None }; let mut tls_next_shard: usize = 0; + let mut accept_backoff = crate::server::accept_backoff::AcceptBackoff::new(); loop { // If TLS listener is configured, select on both plain and TLS accepts @@ -525,6 +534,7 @@ pub async fn run_sharded( } => { match result { Ok((mut stream, addr)) => { + accept_backoff.reset(); if protected_mode_active && !addr.ip().is_loopback() { tracing::warn!( "Protected mode: rejected connection from {} (no password set, use --protected-mode no to disable)", @@ -562,12 +572,13 @@ pub async fn run_sharded( error!("Failed to send connection to shard {}", target_shard); } } - Err(e) => { error!("Accept error: {}", e); } + Err(e) => { accept_backoff.record_error("Accept error", &e).await; } } } result = tls_listener.accept() => { match result { Ok((mut stream, addr)) => { + accept_backoff.reset(); // Protected mode: reject non-loopback connections when no auth configured if protected_mode_active && !addr.ip().is_loopback() { tracing::warn!( @@ -592,7 +603,7 @@ pub async fn run_sharded( } tls_next_shard = (tls_next_shard + 1) % num_shards; } - Err(e) => { error!("TLS accept error: {}", e); } + Err(e) => { accept_backoff.record_error("TLS accept error", &e).await; } } } _ = shutdown.cancelled() => { @@ -612,6 +623,7 @@ pub async fn run_sharded( } => { match result { Ok((mut stream, addr)) => { + accept_backoff.reset(); if protected_mode_active && !addr.ip().is_loopback() { tracing::warn!( "Protected mode: rejected connection from {} (no password set, use --protected-mode no to disable)", @@ -651,7 +663,7 @@ pub async fn run_sharded( } } Err(e) => { - error!("Accept error: {}", e); + accept_backoff.record_error("Accept error", &e).await; } } } diff --git a/src/server/mod.rs b/src/server/mod.rs index cb601f6a..b93c9d60 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1,3 +1,4 @@ +pub mod accept_backoff; pub mod codec; pub mod conn; pub mod conn_state; diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index 950291cd..536c0cf9 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -250,6 +250,15 @@ pub(crate) fn spawn_tokio_connection( ); return; } + // R-3: capture the underlying TCP fd before the handshake moves the + // stream — shutdown on it force-closes the TLS session too. + #[cfg(unix)] + let kill_fd = { + use std::os::unix::io::AsRawFd; + tcp_stream.as_raw_fd() + }; + #[cfg(not(unix))] + let kill_fd = -1; let acceptor = tokio_rustls::TlsAcceptor::from(tls_cfg); match acceptor.accept(tcp_stream).await { Ok(tls_stream) => { @@ -262,6 +271,7 @@ pub(crate) fn spawn_tokio_connection( false, // can_migrate: TLS connections cannot transfer session state BytesMut::new(), None, // fresh connection + kill_fd, ) .await; } @@ -445,6 +455,14 @@ pub(crate) fn spawn_migrated_tokio_connection( // State restoration happens directly via migrated_state parameter — // no synthetic RESP commands, no leaked responses. + // R-3: capture the migrated socket fd for CLIENT KILL force-close. + #[cfg(unix)] + let kill_fd = { + use std::os::unix::io::AsRawFd; + tcp_stream.as_raw_fd() + }; + #[cfg(not(unix))] + let kill_fd = -1; tokio::task::spawn_local(async move { let _ = handle_connection_sharded_inner( tcp_stream, @@ -455,6 +473,7 @@ pub(crate) fn spawn_migrated_tokio_connection( false, // can_migrate: already-migrated connections skip re-migration sampling migration_buf, Some(&state), + kill_fd, ) .await; }); @@ -516,6 +535,15 @@ pub(crate) fn spawn_monoio_connection( match monoio::net::TcpStream::from_std(std_tcp_stream) { Ok(tcp_stream) => { + // R-3: capture the socket fd for CLIENT KILL force-close before the + // stream is moved into a spawned handler (generic `S`, no AsRawFd). + #[cfg(unix)] + let kill_fd = { + use std::os::unix::io::AsRawFd; + tcp_stream.as_raw_fd() + }; + #[cfg(not(unix))] + let kill_fd = -1; let aff = affinity_tracker.clone(); let rsm = remote_subscriber_map.clone(); let sdbs = shard_databases.clone(); @@ -627,6 +655,7 @@ pub(crate) fn spawn_monoio_connection( BytesMut::new(), pw, None, // fresh connection + kill_fd, ) .await; } @@ -664,6 +693,7 @@ pub(crate) fn spawn_monoio_connection( BytesMut::new(), pw, None, // fresh connection + kill_fd, ) .await; @@ -940,6 +970,14 @@ pub(crate) fn spawn_migrated_monoio_connection( do_dir, ); + // R-3: capture the migrated socket fd for CLIENT KILL force-close. + #[cfg(unix)] + let kill_fd = { + use std::os::unix::io::AsRawFd; + tcp_stream.as_raw_fd() + }; + #[cfg(not(unix))] + let kill_fd = -1; monoio::spawn(async move { let _ = handle_connection_sharded_monoio( tcp_stream, @@ -951,6 +989,7 @@ pub(crate) fn spawn_migrated_monoio_connection( migration_buf, pw, Some(&state), + kill_fd, ) .await; // Migrated connection: the source shard's wrapper skipped the diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index 96dc79ef..8992242e 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -21,7 +21,9 @@ use crate::runtime::channel; // Coordinator uses oneshot channels (not ResponseSlotPool) for cross-thread safety. // ResponseSlotPool's AtomicWaker doesn't work with monoio's !Send executor. // The oneshot overhead (~80ns) is negligible on the multi-key coordination path. -use crate::shard::dispatch::{ShardMessage, key_to_shard}; +use crate::shard::dispatch::{ + CROSS_SHARD_PUSH_BACKOFF, CROSS_SHARD_PUSH_MAX_RETRIES, PushOutcome, ShardMessage, key_to_shard, +}; use crate::shard::mesh::ChannelMesh; use crate::storage::entry::CachedClock; @@ -768,11 +770,30 @@ fn extract_key(frame: &Frame) -> Option { } } -/// Send a ShardMessage via SPSC with spin-retry on full buffer. +/// Send a ShardMessage via SPSC with **bounded** backpressure retry on a full +/// ring (R-1, design-for-failure). /// -/// Calls `notify_one()` on the target shard's notifier after successful push +/// Calls `notify_one()` on the target shard's notifier after a successful push /// for immediate wake (avoids relying on the 1ms periodic timer safety net). /// +/// Retries up to [`CROSS_SHARD_PUSH_MAX_RETRIES`] with a [`CROSS_SHARD_PUSH_BACKOFF`] +/// sleep between attempts, then gives up and returns [`PushOutcome::Backpressure`]. +/// This replaces the previous **unbounded** `loop { try_push; yield/sleep }`, +/// which (a) busy-spun a full core on tokio — `yield_now()` reschedules with no +/// backoff — and (b) could block graceful shutdown forever on a wedged target. +/// +/// **Give-up semantics:** on `Backpressure` the message (`pending`) is dropped. +/// For a reply-carrying message (`MultiExecute`/`MultiExecuteSlotted`/…) this +/// drops the embedded reply sender, so the awaiting caller's `reply_rx.recv()` +/// resolves to `Err` and it synthesizes a per-shard error for that slice of the +/// response — the same closed-channel path callers already handle. Fire-and- +/// forget messages become best-effort: a target that stayed full for the whole +/// ~0.5s budget is effectively wedged, so dropping is the correct failure mode +/// rather than spinning forever. +/// +/// The return value lets future call sites branch on the outcome explicitly; +/// existing statement-form callers (`spsc_send(...).await;`) simply discard it. +/// /// Exposed at `pub(crate)` so sibling files (e.g. `scatter_aggregate`) /// can dispatch via the same contention-safe path. pub(crate) async fn spsc_send( @@ -781,29 +802,74 @@ pub(crate) async fn spsc_send( target_shard: usize, msg: ShardMessage, spsc_notifiers: &[Arc], -) { +) -> PushOutcome { + spsc_send_bounded( + dispatch_tx, + my_shard, + target_shard, + msg, + spsc_notifiers, + CROSS_SHARD_PUSH_MAX_RETRIES, + CROSS_SHARD_PUSH_BACKOFF, + ) + .await +} + +/// Retry budget is parameterized so tests can drive the give-up path with a +/// tiny bound; production always calls it via [`spsc_send`] with the shared +/// [`CROSS_SHARD_PUSH_MAX_RETRIES`] / [`CROSS_SHARD_PUSH_BACKOFF`] constants. +async fn spsc_send_bounded( + dispatch_tx: &Rc>>>, + my_shard: usize, + target_shard: usize, + msg: ShardMessage, + spsc_notifiers: &[Arc], + max_retries: u32, + backoff: std::time::Duration, +) -> PushOutcome { let target_idx = ChannelMesh::target_index(my_shard, target_shard); let mut pending = msg; - loop { + + // Fast path: try once with no sleep. The borrow is scoped to the block so + // it is dropped before any `.await` (a RefCell borrow must not be held + // across a yield) and before `notify_one`. + let push_result = { + let mut producers = dispatch_tx.borrow_mut(); + producers[target_idx].try_push(pending) + }; + match push_result { + Ok(()) => { + spsc_notifiers[target_shard].notify_one(); + return PushOutcome::Pushed; + } + Err(val) => pending = val, + } + + // Bounded retry with backoff. + for _ in 0..max_retries { + // Back off before retrying so a full ring cannot hot-spin the core. + // No borrow is held across this await. + #[cfg(feature = "runtime-tokio")] + tokio::time::sleep(backoff).await; + #[cfg(feature = "runtime-monoio")] + monoio::time::sleep(backoff).await; + let push_result = { let mut producers = dispatch_tx.borrow_mut(); producers[target_idx].try_push(pending) - }; // borrow dropped before yield + }; match push_result { Ok(()) => { spsc_notifiers[target_shard].notify_one(); - return; - } - Err(val) => { - pending = val; - // Yield to other tasks to avoid busy-spinning on full SPSC buffer. - #[cfg(feature = "runtime-tokio")] - tokio::task::yield_now().await; - #[cfg(feature = "runtime-monoio")] - monoio::time::sleep(std::time::Duration::from_micros(10)).await; + return PushOutcome::Pushed; } + Err(val) => pending = val, } } + + // Budget exhausted: the target ring never drained. Drop `pending` (and any + // embedded reply sender) so awaiting callers fail loud instead of hanging. + PushOutcome::Backpressure } /// Coordinate MGET across shards using VLL pattern. @@ -2542,6 +2608,90 @@ mod tests { ); } + // ── R-1: bounded spsc_send backpressure ── + // A single-capacity ring at `target_idx` for (my_shard=1, target_shard=0), + // since `ChannelMesh::target_index(1, 0) == 0`. + #[cfg(feature = "runtime-tokio")] + fn make_ring( + capacity: usize, + ) -> ( + Rc>>>, + ringbuf::HeapCons, + Vec>, + ) { + use ringbuf::HeapRb; + use ringbuf::traits::Split; + let (prod, cons) = HeapRb::::new(capacity).split(); + let dispatch_tx = Rc::new(RefCell::new(vec![prod])); + let notifiers = vec![Arc::new(channel::Notify::new())]; + (dispatch_tx, cons, notifiers) + } + + #[cfg(feature = "runtime-tokio")] + #[tokio::test] + async fn spsc_send_pushes_when_ring_has_space() { + use ringbuf::traits::Consumer; + let (dispatch_tx, mut cons, notifiers) = make_ring(4); + let outcome = spsc_send( + &dispatch_tx, + 1, // my_shard + 0, // target_shard -> target_idx 0 + ShardMessage::BlockCancel { wait_id: 42 }, + ¬ifiers, + ) + .await; + assert_eq!(outcome, PushOutcome::Pushed); + // The message actually landed in the ring. + assert!( + matches!( + cons.try_pop(), + Some(ShardMessage::BlockCancel { wait_id: 42 }) + ), + "pushed message must be present in the target ring" + ); + } + + #[cfg(feature = "runtime-tokio")] + #[tokio::test] + async fn spsc_send_gives_up_when_ring_never_drains() { + // Fill the capacity-1 ring so every push attempt fails: a wedged target. + // The bounded retry MUST terminate with Backpressure rather than spin + // forever — a regression to the old unbounded loop would hang here. A + // tiny budget (5 retries × 1ms) keeps the test fast and deterministic. + let (dispatch_tx, _cons, notifiers) = make_ring(1); + // Pre-fill the ring (hold `_cons` so nothing drains it). + { + use ringbuf::traits::Producer; + let mut prods = dispatch_tx.borrow_mut(); + assert!( + prods[0] + .try_push(ShardMessage::BlockCancel { wait_id: 1 }) + .is_ok() + ); + } + let start = std::time::Instant::now(); + let outcome = spsc_send_bounded( + &dispatch_tx, + 1, + 0, + ShardMessage::BlockCancel { wait_id: 2 }, + ¬ifiers, + 5, + std::time::Duration::from_millis(1), + ) + .await; + assert_eq!( + outcome, + PushOutcome::Backpressure, + "a ring that never drains must yield Backpressure, not hang" + ); + assert!( + start.elapsed() < std::time::Duration::from_secs(2), + "bounded retry must terminate promptly, took {:?}", + start.elapsed() + ); + } + #[cfg(feature = "runtime-tokio")] #[tokio::test] async fn test_coordinate_mget_all_local() { diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index be9f2e30..79b8b5d0 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -320,9 +320,11 @@ impl super::Shard { let (tx, rx) = flume::bounded(256); let shard_id_copy = self.id; monoio::spawn(async move { + let mut accept_backoff = crate::server::accept_backoff::AcceptBackoff::new(); loop { match listener.accept().await { Ok((stream, _addr)) => { + accept_backoff.reset(); let std_stream = { use std::os::unix::io::{FromRawFd, IntoRawFd}; let fd = stream.into_raw_fd(); @@ -335,11 +337,10 @@ impl super::Shard { } } Err(e) => { - tracing::error!( - "Shard {}: per-shard accept error: {}", - shard_id_copy, - e - ); + // R-4: capped backoff + rate-limited log so an + // fd-exhaustion storm can't hot-spin this shard. + let ctx = format!("Shard {shard_id_copy}: per-shard accept error"); + accept_backoff.record_error(&ctx, &e).await; } } } @@ -1061,6 +1062,11 @@ impl super::Shard { #[cfg(feature = "runtime-monoio")] let pending_wakers: Rc>> = Rc::new(RefCell::new(Vec::new())); + // R-4: backoff for the tokio per-shard SO_REUSEPORT accept branch so an + // fd-exhaustion storm can't hot-spin this shard's select loop. + #[cfg(feature = "runtime-tokio")] + let mut per_shard_accept_backoff = crate::server::accept_backoff::AcceptBackoff::new(); + loop { #[cfg(feature = "runtime-tokio")] tokio::select! { @@ -1111,6 +1117,7 @@ impl super::Shard { } => { match result { Ok((tcp_stream, _addr)) => { + per_shard_accept_backoff.reset(); conn_accept::spawn_tokio_connection( tcp_stream, false, &tls_config, &shard_databases, &dispatch_tx, &pubsub_arc, &blocking_rc, @@ -1124,7 +1131,8 @@ impl super::Shard { ); } Err(e) => { - tracing::error!("Shard {}: per-shard accept error: {}", shard_id, e); + let ctx = format!("Shard {shard_id}: per-shard accept error"); + per_shard_accept_backoff.record_error(&ctx, &e).await; } } } diff --git a/src/shard/uring_handler.rs b/src/shard/uring_handler.rs index de00adb8..633bb1c7 100644 --- a/src/shard/uring_handler.rs +++ b/src/shard/uring_handler.rs @@ -369,7 +369,15 @@ pub(crate) fn handle_uring_event( let _ = driver.register_connection(raw_fd); } IoEvent::AcceptError { .. } => { - // Multishot accept cancelled on error -- re-submit + // Multishot accept cancelled on error -- re-submit. + // + // R-4 follow-up: under persistent fd exhaustion (EMFILE/ENFILE) the + // immediate resubmit here can hot-spin the CQ loop. The socket-based + // accept loops back off (see `server::accept_backoff`), but this is a + // *synchronous* CQE handler with no async context to sleep in, so a + // backoff needs deferred-resubmit state (submit after a timer CQE). + // Scoped separately; this bridge is opt-in (MOON_URING=1), not the + // default accept path. if let Some(lfd) = uring_listener_fd { let _ = driver.submit_multishot_accept(lfd); } From 6828e6b0d601b901c158e9d3b9d6087628a6c6e9 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 09:38:52 +0700 Subject: [PATCH 2/6] fix(server): address PR #230 self-review findings (Track B follow-up) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five fixes from a deep-dive review of the connection-plane hardening PR: 1. CLIENT KILL self-kill stays cooperative (R-3 regression): kill_clients now takes the executing connection's id and skips the fd shutdown(2) for a matching self entry — flag only — so the +count reply is still flushed before the handler closes (Redis replies first on self-kill). Red/green: test_self_kill_does_not_force_close_socket_fd. 2. R-1 give-up fails loud at side-effect-bearing senders: a dropped MqTxnMaterialize leg turns TXN.COMMIT's reply into an explicit "MOONERR TXN.COMMIT partial" error (the commit is already WAL-durable; only foreign MQ materialization was lost — previously a silent +OK with lost messages) and logs at error!; a dropped GraphRollback logs at error! (leaked remote graph intents), both on both runtimes. 3. PushOutcome is #[must_use]: no call site can silently drop a cross-shard message by accident. All reply-carrying sites (which fail loud via the closed reply channel) now discard explicitly (let _ =). 4. spsc_send latency: a bounded ≤64-iteration spin before the first timed backoff preserves the old ~10µs-class recovery when the target ring is only transiently full — the 100µs timer sleep may round up to ~1ms on coarse runtime timers. Wedged-target budget unchanged. 5. Accept backoff cap 1s → 100ms: the sleep runs inside select! arms where the shutdown branch cannot preempt it, so the cap bounds shutdown latency during an fd-exhaustion storm (≤10 wakeups/s). Both runtimes: fmt clean, clippy clean, full lib suites green. Refs: PR #230 author: Tin Dang --- CHANGELOG.md | 18 +++++-- src/client_registry.rs | 58 +++++++++++++++++++-- src/server/accept_backoff.rs | 18 ++++--- src/server/conn/handler_monoio/dispatch.rs | 2 +- src/server/conn/handler_monoio/txn.rs | 30 ++++++++++- src/server/conn/handler_monoio/write.rs | 6 +-- src/server/conn/handler_sharded/dispatch.rs | 2 +- src/server/conn/handler_sharded/txn.rs | 34 ++++++++++-- src/server/conn/handler_sharded/write.rs | 6 +-- src/shard/coordinator.rs | 58 ++++++++++++++------- src/shard/dispatch.rs | 7 +++ src/shard/scatter_aggregate.rs | 2 +- src/shard/scatter_hybrid.rs | 4 +- src/transaction/abort.rs | 17 +++++- 14 files changed, 212 insertions(+), 50 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index faaa6f37..935c8af8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -144,6 +144,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the registry read lock, and a connection's `RegistryGuard` deregisters (needs the write lock) strictly before its stream drops — so a visible entry always has a live fd. No-op on non-unix (`CLIENT KILL` stays cooperative there). + Self-kill (the filter matching the executing connection) stays cooperative — + flag only, no fd shutdown — so the `+count` reply is still delivered before + the handler closes, matching Redis's reply-first self-kill semantics. - **R-2 — inline-command length cap** (`src/protocol/inline.rs`, `frame.rs`): the RESP-less inline path had no maximum line length, so a client that never sends `\r\n` (raw non-RESP bytes) grew the connection read @@ -162,15 +165,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the same budget as `push_with_backpressure`) returning `PushOutcome`. On give-up the message is dropped; reply-carrying callers observe the closed reply channel and synthesize a per-shard error via the path they already - handle. + handle. `PushOutcome` is `#[must_use]` so no call site can drop a message + silently by accident, and the two side-effect-bearing senders fail loud: + a dropped `MqTxnMaterialize` leg turns `TXN.COMMIT`'s reply into an explicit + `MOONERR TXN.COMMIT partial` error (the commit is already WAL-durable; only + foreign MQ materialization was lost) and a dropped `GraphRollback` logs at + `error!`. A brief bounded spin (≤64 iterations) before the first timed + backoff preserves the old ~10µs-class latency when the target ring is only + transiently full (the 100µs timer sleep may round up to ~1ms). - **R-4 — accept-loop backoff on fd exhaustion** (`src/server/accept_backoff.rs`, `listener.rs`, `shard/event_loop.rs`): all six socket-accept loops (tokio and monoio; sharded, non-sharded, and TLS) retried `accept()` errors with zero backoff, so `EMFILE`/`ENFILE` under a connection storm pinned a shard core at 100% and flooded the log. Added `AcceptBackoff`: capped exponential - backoff (1 ms → 1 s) on resource-exhaustion errors only, plus rate-limited + backoff (1 ms → 100 ms) on resource-exhaustion errors only, plus rate-limited logging; benign per-connection errors (e.g. `ECONNABORTED`) log without - sleeping. The io_uring multishot-accept resubmit (opt-in `MOON_URING=1` + sleeping. The cap is 100 ms (not 1 s) because the sleep runs inside `select!` + arms where the shutdown branch cannot preempt it — this bounds shutdown + latency during a storm. The io_uring multishot-accept resubmit (opt-in `MOON_URING=1` bridge) is documented as a scoped follow-up (synchronous CQE handler, no async context to sleep in). diff --git a/src/client_registry.rs b/src/client_registry.rs index a7ab8242..d08ec8af 100644 --- a/src/client_registry.rs +++ b/src/client_registry.rs @@ -206,7 +206,13 @@ pub fn client_info(id: u64) -> Option { /// parked in `read().await` — idle with `timeout 0` — is torn down at once /// rather than lingering until it next sends bytes. The parked read returns /// `Ok(0)`/`Err`, which the handler already treats as disconnect. -pub fn kill_clients(filter: &KillFilter) -> u64 { +/// +/// `self_id` is the id of the connection *executing* CLIENT KILL, if any. +/// A matching self entry gets the cooperative flag only — never the fd +/// shutdown — so the `+count` reply is still delivered before the handler's +/// per-batch kill check closes the connection (Redis replies first on +/// self-kill, then closes). +pub fn kill_clients(filter: &KillFilter, self_id: Option) -> u64 { // Hold the read lock across the whole loop. This is what makes the raw-fd // `shutdown` free of a use-after-reuse race: a connection's `RegistryGuard` // (a local) drops — and calls `deregister`, which needs the registry WRITE @@ -224,7 +230,11 @@ pub fn kill_clients(filter: &KillFilter) -> u64 { }; if matches { entry.live.kill_flag.store(true, Ordering::Relaxed); - force_close_fd(entry.live.kill_fd); + // Self-kill stays cooperative: shutting down our own socket here + // would happen mid-command, before the reply is flushed. + if self_id != Some(entry.id) { + force_close_fd(entry.live.kill_fd); + } count += 1; } } @@ -343,7 +353,7 @@ mod tests { let live = register(id, "10.0.0.2:6000".into(), "bob".into(), 0, -1); assert!(!is_killed(id)); assert!(!live.is_killed()); - let count = kill_clients(&KillFilter::Id(id)); + let count = kill_clients(&KillFilter::Id(id), None); assert_eq!(count, 1); assert!(is_killed(id)); assert!(live.is_killed(), "live handle observes the kill lock-free"); @@ -371,7 +381,7 @@ mod tests { killed_end.as_raw_fd(), ); - let count = kill_clients(&KillFilter::Id(id)); + let count = kill_clients(&KillFilter::Id(id), None); assert_eq!(count, 1); assert!(live.is_killed()); @@ -384,13 +394,51 @@ mod tests { drop(killed_end); } + // Self-kill (CLIENT KILL matching the executing connection) must stay + // cooperative: the fd is NOT shut down, so the `+count` reply can still be + // flushed; only the kill_flag is set (handler closes after the batch). + #[cfg(unix)] + #[test] + fn test_self_kill_does_not_force_close_socket_fd() { + use std::io::{Read, Write}; + use std::os::unix::io::AsRawFd; + use std::os::unix::net::UnixStream; + + let (self_end, mut peer) = UnixStream::pair().expect("socketpair"); + let id = 999_021; + let live = register( + id, + "unix:socketpair-self".into(), + "default".into(), + 0, + self_end.as_raw_fd(), + ); + + let count = kill_clients(&KillFilter::Id(id), Some(id)); + assert_eq!(count, 1); + assert!(live.is_killed(), "cooperative flag must still be set"); + + // The socket must remain writable in both directions — the reply to + // the killer is flushed over exactly this fd after kill_clients runs. + (&self_end) + .write_all(b"+1\r\n") + .expect("self fd must still accept the reply write"); + let mut buf = [0u8; 8]; + let n = peer.read(&mut buf).expect("peer read"); + assert_eq!(n, 4, "peer must receive the reply, not EOF"); + assert_eq!(&buf[..4], b"+1\r\n"); + + deregister(id); + drop(self_end); + } + #[test] fn test_kill_by_user() { let id1 = 999_010; let id2 = 999_011; register(id1, "10.0.0.3:7000".into(), "eve".into(), 0, -1); register(id2, "10.0.0.4:7001".into(), "eve".into(), 1, -1); - let count = kill_clients(&KillFilter::User("eve".into())); + let count = kill_clients(&KillFilter::User("eve".into()), None); assert_eq!(count, 2); assert!(is_killed(id1)); assert!(is_killed(id2)); diff --git a/src/server/accept_backoff.rs b/src/server/accept_backoff.rs index 87d38903..6f6644a8 100644 --- a/src/server/accept_backoff.rs +++ b/src/server/accept_backoff.rs @@ -10,15 +10,20 @@ //! the project's own gotchas. //! //! [`AcceptBackoff`] rate-limits the error log and, for resource-exhaustion -//! errors only, sleeps a capped exponential backoff (1ms → 1s). Benign +//! errors only, sleeps a capped exponential backoff (1ms → 100ms). The cap is +//! deliberately modest: several accept loops sleep *inside* a `select!` arm, +//! where the shutdown branch cannot fire mid-sleep — 100ms bounds shutdown +//! latency during a storm while still cutting the spin to ≤10 wakeups/s. Benign //! per-connection errors (a client that reset mid-handshake, `ECONNABORTED`) //! are logged but not slept on, so normal accept throughput is unaffected. //! Call [`AcceptBackoff::reset`] after every successful `accept()`. use std::time::Duration; -/// Ceiling for the exponential backoff (1 second). -const MAX_BACKOFF_MS: u64 = 1000; +/// Ceiling for the exponential backoff (100ms). Bounded well below the +/// shutdown-responsiveness budget because the sleep runs inside `select!` +/// arms where cancellation cannot preempt it. +const MAX_BACKOFF_MS: u64 = 100; /// Per-accept-loop backoff + log rate-limiter. One per accept loop; reset on /// every successful accept so a transient error never permanently slows the @@ -99,9 +104,10 @@ mod tests { assert_eq!(backoff_for(2), Duration::from_millis(2)); assert_eq!(backoff_for(3), Duration::from_millis(4)); assert_eq!(backoff_for(4), Duration::from_millis(8)); - // Caps at 1s and stays there regardless of streak length. - assert_eq!(backoff_for(11), Duration::from_millis(1000)); - assert_eq!(backoff_for(1000), Duration::from_millis(1000)); + // Caps at 100ms (shutdown-responsiveness bound) regardless of streak. + assert_eq!(backoff_for(8), Duration::from_millis(100)); + assert_eq!(backoff_for(11), Duration::from_millis(100)); + assert_eq!(backoff_for(1000), Duration::from_millis(100)); } #[test] diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index c833330a..500550f0 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -874,7 +874,7 @@ pub(super) fn try_handle_client_admin( .collect(); match crate::client_registry::parse_kill_args(&raw_args) { Some(filter) => { - let count = crate::client_registry::kill_clients(&filter); + let count = crate::client_registry::kill_clients(&filter, Some(client_id)); responses.push(Frame::Integer(count as i64)); } None => { diff --git a/src/server/conn/handler_monoio/txn.rs b/src/server/conn/handler_monoio/txn.rs index 61777df1..03d46680 100644 --- a/src/server/conn/handler_monoio/txn.rs +++ b/src/server/conn/handler_monoio/txn.rs @@ -126,6 +126,10 @@ pub(super) async fn try_handle_txn_commit( crate::shard::dispatch::key_to_shard(&intent.queue_key, ctx.num_shards); by_shard.entry(owner).or_default().push(intent); } + // The commit is already WAL-durable here: a dropped foreign + // leg cannot fail the commit, but it must fail LOUD at the + // client — never a silent `+OK` with lost MQ messages. + let mut mq_lost: Option<(usize, usize)> = None; // (shard, intents) for (owner, intents) in by_shard { if owner == ctx.shard_id { // Self: apply locally via slice. @@ -141,13 +145,14 @@ pub(super) async fn try_handle_txn_commit( }); } else { // Foreign: send MqTxnMaterialize hop and await ack. + let intent_count = intents.len(); let (reply_tx, reply_rx) = crate::runtime::channel::oneshot(); let msg = crate::shard::dispatch::ShardMessage::MqTxnMaterialize { db_index: conn.selected_db as usize, intents, reply_tx, }; - crate::shard::coordinator::spsc_send( + let outcome = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, @@ -155,10 +160,31 @@ pub(super) async fn try_handle_txn_commit( &ctx.spsc_notifiers, ) .await; + if outcome != crate::shard::dispatch::PushOutcome::Pushed { + // Dropped: the intents were NEVER delivered. + tracing::error!( + owner, + intent_count, + "TXN.COMMIT MQ materialize: dispatch backpressure — \ + intents dropped, commit reported as partial" + ); + mq_lost.get_or_insert((owner, intent_count)); + continue; + } // Await the ack before replying OK to the client. let _ = reply_rx.recv().await; } } + if let Some((owner, intent_count)) = mq_lost { + // KV/vector/graph legs ARE committed and durable; only + // foreign MQ materialization was lost. + responses.push(Frame::Error(Bytes::from(format!( + "MOONERR TXN.COMMIT partial: committed, but {intent_count} \ + MQ intent(s) for shard {owner} were dropped under \ + dispatch backpressure" + )))); + return true; + } } responses.push(Frame::SimpleString(Bytes::from_static(b"OK"))); @@ -265,7 +291,7 @@ pub(super) async fn try_handle_temporal_invalidate( command: std::sync::Arc::new(frame.clone()), reply_tx, }; - crate::shard::coordinator::spsc_send( + let _ = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index 7260b01e..8dbefbed 100644 --- a/src/server/conn/handler_monoio/write.rs +++ b/src/server/conn/handler_monoio/write.rs @@ -142,7 +142,7 @@ pub(super) async fn try_handle_ws_command( prefix: prefix_bytes, reply_tx, }; - crate::shard::coordinator::spsc_send( + let _ = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, @@ -544,7 +544,7 @@ async fn mq_hop_or_local( reply_tx, }; let msg = crate::shard::dispatch::ShardMessage::MqCommand(Box::new(payload)); - crate::shard::coordinator::spsc_send( + let _ = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, @@ -661,7 +661,7 @@ pub(super) async fn try_handle_graph_command( command: std::sync::Arc::new(frame.clone()), reply_tx, }; - crate::shard::coordinator::spsc_send( + let _ = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 73a2bffa..d598d99f 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -148,7 +148,7 @@ pub(super) fn try_handle_client_command( .collect(); match crate::client_registry::parse_kill_args(&raw_args) { Some(filter) => { - let count = crate::client_registry::kill_clients(&filter); + let count = crate::client_registry::kill_clients(&filter, Some(client_id)); responses.push(Frame::Integer(count as i64)); } None => { diff --git a/src/server/conn/handler_sharded/txn.rs b/src/server/conn/handler_sharded/txn.rs index 33983db2..3112d7fa 100644 --- a/src/server/conn/handler_sharded/txn.rs +++ b/src/server/conn/handler_sharded/txn.rs @@ -152,15 +152,21 @@ pub(super) async fn try_handle_txn_commit( } }); } - // Send MqTxnMaterialize to each foreign shard and await all acks. + // Send MqTxnMaterialize to each foreign shard and await all + // acks. The commit is already WAL-durable at this point, so + // a dropped/failed leg cannot fail the commit — but it must + // fail LOUD at the client, never a silent `+OK` with lost + // MQ messages. + let mut mq_lost: Option<(usize, usize)> = None; // (shard, intents) for (owner, intents) in foreign { + let intent_count = intents.len(); let (reply_tx, reply_rx) = crate::runtime::channel::oneshot(); let msg = crate::shard::dispatch::ShardMessage::MqTxnMaterialize { db_index: conn.selected_db, intents, reply_tx, }; - crate::shard::coordinator::spsc_send( + let outcome = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, @@ -168,6 +174,18 @@ pub(super) async fn try_handle_txn_commit( &ctx.spsc_notifiers, ) .await; + if outcome != crate::shard::dispatch::PushOutcome::Pushed { + // Message dropped (target ring never drained): the + // intents were NEVER delivered. Escalate. + tracing::error!( + owner, + intent_count, + "TXN.COMMIT MQ materialize: dispatch backpressure — \ + intents dropped, commit reported as partial" + ); + mq_lost.get_or_insert((owner, intent_count)); + continue; + } match reply_rx.recv().await { Ok(()) => {} Err(_) => { @@ -179,6 +197,16 @@ pub(super) async fn try_handle_txn_commit( } } } + if let Some((owner, intent_count)) = mq_lost { + // KV/vector/graph legs of the txn ARE committed and + // durable; only foreign MQ materialization was lost. + responses.push(Frame::Error(Bytes::from(format!( + "MOONERR TXN.COMMIT partial: committed, but {intent_count} \ + MQ intent(s) for shard {owner} were dropped under \ + dispatch backpressure" + )))); + return true; + } } responses.push(Frame::SimpleString(Bytes::from_static(b"OK"))); @@ -286,7 +314,7 @@ pub(super) async fn try_handle_temporal_invalidate( command: std::sync::Arc::new(frame.clone()), reply_tx, }; - crate::shard::coordinator::spsc_send( + let _ = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, diff --git a/src/server/conn/handler_sharded/write.rs b/src/server/conn/handler_sharded/write.rs index bf0c06da..6a71ee71 100644 --- a/src/server/conn/handler_sharded/write.rs +++ b/src/server/conn/handler_sharded/write.rs @@ -137,7 +137,7 @@ pub(super) async fn try_handle_ws_command( prefix: prefix_bytes, reply_tx, }; - crate::shard::coordinator::spsc_send( + let _ = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, cleanup_owner, @@ -321,7 +321,7 @@ async fn mq_dispatch_to_owner( reply_tx, }; let msg = crate::shard::dispatch::ShardMessage::MqCommand(Box::new(payload)); - crate::shard::coordinator::spsc_send( + let _ = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, @@ -612,7 +612,7 @@ pub(super) async fn try_handle_graph_command( command: std::sync::Arc::new(frame.clone()), reply_tx, }; - crate::shard::coordinator::spsc_send( + let _ = crate::shard::coordinator::spsc_send( &ctx.dispatch_tx, ctx.shard_id, owner, diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index 8992242e..eb9c43e2 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -195,7 +195,7 @@ async fn run_remote( commands: vec![(routing_key.clone(), command)], reply_tx, }; - spsc_send(dispatch_tx, my_shard, target_shard, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target_shard, msg, spsc_notifiers).await; match reply_rx.recv().await { Ok(mut frames) if !frames.is_empty() => frames.swap_remove(0), _ => Frame::Error(Bytes::from_static(b"ERR cross-shard reply channel closed")), @@ -503,7 +503,7 @@ async fn coordinate_bitop( commands, reply_tx, }; - spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; pending.push((indices, reply_rx)); } } @@ -845,6 +845,28 @@ async fn spsc_send_bounded( Err(val) => pending = val, } + // Brief spin before the first timed sleep: the consumer is another OS + // thread actively draining, so a transiently-full ring often frees a slot + // within nanoseconds — far below the 100µs backoff (which runtime timers + // may round up to ~1ms). This preserves the old 10µs-class latency for + // bursty-but-healthy targets without touching the wedged-target budget. + // Bounded and tiny (≤64 iterations), so it cannot convoy siblings the way + // an unbounded reply-side spin did (see the C2 solo-conn gate lesson). + for _ in 0..64 { + std::hint::spin_loop(); + let push_result = { + let mut producers = dispatch_tx.borrow_mut(); + producers[target_idx].try_push(pending) + }; + match push_result { + Ok(()) => { + spsc_notifiers[target_shard].notify_one(); + return PushOutcome::Pushed; + } + Err(val) => pending = val, + } + } + // Bounded retry with backoff. for _ in 0..max_retries { // Back off before retrying so a full ring cannot hot-spin the core. @@ -953,7 +975,7 @@ async fn coordinate_mget( commands, reply_tx, }; - spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; pending_shards.push((original_indices, reply_rx)); } } @@ -1083,7 +1105,7 @@ async fn coordinate_mset( commands, reply_tx, }; - spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; pending_shards.push(reply_rx); } } @@ -1321,7 +1343,7 @@ async fn coordinate_multi_del_or_exists( commands, reply_tx, }; - spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; pending_shards.push(reply_rx); } } @@ -1402,7 +1424,7 @@ pub async fn coordinate_keys( command: std::sync::Arc::new(cmd_frame), reply_tx, }; - spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; pending_shards.push(reply_rx); } @@ -1491,7 +1513,7 @@ pub async fn coordinate_scan( command: std::sync::Arc::new(cmd_frame), reply_tx, }; - spsc_send(dispatch_tx, my_shard, target_shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target_shard_id, msg, spsc_notifiers).await; match reply_rx.recv().await { Ok(frame) => frame, Err(_) => Frame::Error(Bytes::from_static( @@ -1573,7 +1595,7 @@ pub async fn coordinate_dbsize( command: std::sync::Arc::new(cmd_frame), reply_tx, }; - spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; pending_shards.push(reply_rx); } @@ -1633,7 +1655,7 @@ pub async fn coordinate_hotkeys( command: std::sync::Arc::new(cmd_frame), reply_tx, }; - spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; pending_shards.push(reply_rx); } @@ -1718,7 +1740,7 @@ pub async fn scatter_vector_search( as_of_lsn, reply_tx, })); - spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; receivers.push(reply_rx); } } @@ -1796,7 +1818,7 @@ pub async fn scatter_vector_search_remote( as_of_lsn, reply_tx, })); - spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; receivers.push(reply_rx); } @@ -1846,7 +1868,7 @@ pub async fn broadcast_vector_command( command: command.clone(), reply_tx, }; - spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; receivers.push(reply_rx); } @@ -1942,7 +1964,7 @@ pub async fn scatter_invalidate_range( command: command.clone(), reply_tx, }; - spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; receivers.push(reply_rx); } @@ -2022,7 +2044,7 @@ pub async fn scatter_ft_info( command: command.clone(), reply_tx, }; - spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; receivers.push(reply_rx); } @@ -2194,7 +2216,7 @@ pub async fn scatter_text_search( field_queries: field_queries.clone(), reply_tx, }; - spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; doc_freq_receivers.push(reply_rx); } } @@ -2282,7 +2304,7 @@ pub async fn scatter_text_search( summarize_opts: summarize_opts.clone(), reply_tx, })); - spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; search_receivers.push(reply_rx); } } @@ -2401,7 +2423,7 @@ pub async fn scatter_text_search_filter( reply_tx, }, )); - spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; receivers.push(reply_rx); } } @@ -2518,7 +2540,7 @@ pub async fn coordinate_swapdb( } let (reply_tx, reply_rx) = channel::oneshot(); let msg = ShardMessage::SwapDb { a, b, reply_tx }; - spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; receivers.push(reply_rx); } diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index 3d71f10c..9ea0e950 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -768,6 +768,13 @@ pub(crate) const CROSS_SHARD_PUSH_BACKOFF: std::time::Duration = pub(crate) const CROSS_SHARD_PUSH_MAX_RETRIES: u32 = 5_000; /// Outcome of a bounded cross-shard SPSC push ([`push_with_backpressure`], F3). +/// +/// `#[must_use]`: on `Backpressure` the message was DROPPED (including any +/// embedded reply sender). Reply-carrying call sites may discard with +/// `let _ =` — their `reply_rx.recv()` observes the closed channel and fails +/// loud — but fire-and-forget or side-effect-bearing sites (txn materialize, +/// rollback) must branch on the outcome explicitly. +#[must_use] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum PushOutcome { /// The target ring accepted the message. diff --git a/src/shard/scatter_aggregate.rs b/src/shard/scatter_aggregate.rs index a6da0330..5d4bfc69 100644 --- a/src/shard/scatter_aggregate.rs +++ b/src/shard/scatter_aggregate.rs @@ -108,7 +108,7 @@ pub async fn scatter_text_aggregate( pipeline: pipeline.clone(), reply_tx, })); - spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; receivers.push(reply_rx); } } diff --git a/src/shard/scatter_hybrid.rs b/src/shard/scatter_hybrid.rs index 27b9892c..81624812 100644 --- a/src/shard/scatter_hybrid.rs +++ b/src/shard/scatter_hybrid.rs @@ -163,7 +163,7 @@ pub async fn scatter_hybrid_search( field_queries: field_queries.clone(), reply_tx, }; - spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; dfs_receivers.push(reply_rx); } } @@ -242,7 +242,7 @@ pub async fn scatter_hybrid_search( reply_tx, }; let msg = ShardMessage::FtHybrid(Box::new(payload)); - spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; + let _ = spsc_send(dispatch_tx, my_shard, shard_id, msg, spsc_notifiers).await; hyb_receivers.push(reply_rx); } } diff --git a/src/transaction/abort.rs b/src/transaction/abort.rs index e127be7a..fc7828e9 100644 --- a/src/transaction/abort.rs +++ b/src/transaction/abort.rs @@ -481,8 +481,21 @@ pub async fn abort_cross_store_txn_routed( reply_tx, }, )); - crate::shard::coordinator::spsc_send(dispatch_tx, shard_id, owner, msg, spsc_notifiers) - .await; + let outcome = + crate::shard::coordinator::spsc_send(dispatch_tx, shard_id, owner, msg, spsc_notifiers) + .await; + if outcome != crate::shard::dispatch::PushOutcome::Pushed { + // The rollback was never delivered: the owner shard keeps the + // txn's graph intents un-undone until its own reconcile/GC path + // catches them. Loud, not warn — this is leaked remote state. + tracing::error!( + txn_id, + owner, + "txn abort: remote graph rollback DROPPED under dispatch \ + backpressure — remote graph intents not undone" + ); + continue; + } if reply_rx.recv().await.is_err() { tracing::warn!( txn_id, From 3b546973ac83c7b98ba81eafe2751871d0ae3f0f Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 10:13:32 +0700 Subject: [PATCH 3/6] fix(server): connection-plane durability & lifecycle wave (arch review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the recommended fixes from the connection-plane architecture review (multi-connection × multi-db), prioritized for long-term durability: D-1 (durability, HIGH): cross-store TXN crash replay hardcoded db 0 (wal_v3/replay.rs "TODO: support multi-db"). CrossStoreTxn now records the BEGIN-time db index; commits write a new XactCommitV2 WAL record (0x53) carrying it in the header, and replay restores KV ops into that db (out-of-range → db 0 with loud warn; v1 records keep db-0 replay). Red/green: test_xact_commit_v2_replays_into_selected_db + fallback test. R-6 (lifecycle): connection migration failed closed — the source shard dropped its socket before the SPSC hand-off was confirmed, losing the client exactly under overload. Both runtimes now keep the original stream until the push succeeds (bounded retry 8×100µs) and on give-up resume serving locally with migration disabled, restoring state from the undelivered message. Also fixes the tokio loss path's connected_clients counter leak. Observability: moon_xshard_backpressure_drops_total{target_shard} counter + warn at the R-1 give-up point; per-shard moon_shard_connected_clients{shard} gauge at registry register/deregister (SO_REUSEPORT imbalance / affinity funnels). P-1 (footprint): FunctionRegistry + LuaEvictionCtx now lazy per connection (built on first FUNCTION/FCALL/FCALL_RO) instead of eager — zero cost for the >99% of connections that never call Functions. S-5: --tcp-backlog flag replaces the hardcoded listen(1024) (default unchanged; process-wide, set before any listener binds). Verified: fmt + clippy clean both runtimes; macOS monoio, tokio, and Linux (OrbStack) checks green; full lib suites green on both runtimes. Refs: PR #230, tmp/CONN-DEEP-REVIEW.md author: Tin Dang --- CHANGELOG.md | 34 +++++++ src/admin/metrics_setup.rs | 33 ++++++ src/client_registry.rs | 5 +- src/config.rs | 8 ++ src/main.rs | 4 + src/persistence/wal_v3/record.rs | 56 +++++++++++ src/persistence/wal_v3/replay.rs | 111 ++++++++++++++++++++- src/server/conn/core.rs | 29 ++++++ src/server/conn/handler_monoio/dispatch.rs | 25 ++++- src/server/conn/handler_monoio/mod.rs | 22 ++-- src/server/conn/handler_monoio/txn.rs | 28 ++++-- src/server/conn/handler_sharded/mod.rs | 104 ++++++++++++++----- src/server/conn/handler_sharded/txn.rs | 17 +++- src/server/conn/shared.rs | 4 +- src/shard/conn_accept.rs | 105 +++++++++++++------ src/shard/coordinator.rs | 8 ++ src/transaction/mod.rs | 27 +++-- 17 files changed, 517 insertions(+), 103 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 935c8af8..63b5ac54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -132,6 +132,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 shared `READY_TIMEOUT` constant (poll-based loop notices readiness within 100ms either way; the fixed 15s was a documented host-load flake, observed on the macOS CI runner). +### Fixed — connection-plane durability & lifecycle wave (PR #230) + +- **D-1 — cross-store TXN crash replay restores into the correct db** + (`src/transaction/mod.rs`, `wal_v3/record.rs`, `wal_v3/replay.rs`, txn + handlers): `CrossStoreTxn` had no db field and WAL replay hardcoded + `databases[0]`, so a `TXN.BEGIN…COMMIT` issued under `SELECT n≠0` replayed + its KV ops into db 0 after a crash — silent recovery corruption. Commits now + write a new `XactCommitV2` record (`0x53`) whose header carries the + BEGIN-time db index; replay targets that db (out-of-range falls back to + db 0 with a loud warning). v1 records still replay into db 0 (faithful to + the builds that wrote them). +- **R-6 — connection migration fails open** (`conn_accept.rs`, + `handler_sharded/mod.rs`): the source shard dropped its socket before the + SPSC hand-off to the target was confirmed, so a full ring lost the client — + precisely under the overload that triggers migration. Both runtimes now + keep the original stream alive until the push succeeds (bounded retry, + 8 × 100µs) and on give-up resume serving the connection on the source shard + with migration disabled, using the state recovered from the undelivered + message. Also fixes the tokio path's `connected_clients` leak on the old + loss path. +- **Observability** (`admin/metrics_setup.rs`): new + `moon_xshard_backpressure_drops_total{target_shard}` counter + warn on every + R-1 give-up, and `moon_shard_connected_clients{shard}` gauge (maintained at + registry register/deregister) to surface SO_REUSEPORT imbalance and + affinity funnels without parsing `CLIENT LIST`. +- **P-1 — lazy per-connection `FunctionRegistry`** (both handlers, + `conn/core.rs`): the Functions API registry + `LuaEvictionCtx` (6 Arc/Rc + clones) were built eagerly for every connection; now built on first + FUNCTION/FCALL/FCALL_RO, cutting per-connection setup cost at high + connection counts. +- **S-5 — `--tcp-backlog`** (`config.rs`, `conn_accept.rs`, `main.rs`): the + per-socket listen backlog was hardcoded 1024; now configurable (default + unchanged) for connection-storm tuning alongside `ulimit -n`. + ### Fixed — connection-plane robustness hardening (Track B) (PR #230) - **R-3 — `CLIENT KILL` force-closes idle connections** (`src/client_registry.rs`, diff --git a/src/admin/metrics_setup.rs b/src/admin/metrics_setup.rs index c6558325..abaa4198 100644 --- a/src/admin/metrics_setup.rs +++ b/src/admin/metrics_setup.rs @@ -672,6 +672,39 @@ pub fn connected_clients() -> u64 { CONNECTED_CLIENTS.load(Ordering::Relaxed) } +/// Adjust the per-shard connected-clients gauge (S-6 observability): the +/// only way to see SO_REUSEPORT imbalance or an affinity-funnel pile-up +/// without parsing CLIENT LIST. Called from client-registry +/// register/deregister — connect/disconnect rate, never per-command. +#[inline] +pub fn record_shard_connection_delta(shard: usize, delta: f64) { + if !METRICS_INITIALIZED.load(Ordering::Relaxed) { + return; + } + gauge!("moon_shard_connected_clients", "shard" => shard.to_string()).increment(delta); +} + +/// Record a cross-shard dispatch message dropped after the bounded +/// backpressure budget expired ([`PushOutcome::Backpressure`] give-up, R-1). +/// +/// Fires only on the give-up path (target ring never drained for ~0.5s), so +/// the label allocation is off the hot path. A non-zero rate here means a +/// shard is wedged or persistently saturated — reply-carrying callers have +/// surfaced per-command errors, but this counter is the aggregate signal. +/// +/// [`PushOutcome::Backpressure`]: crate::shard::dispatch::PushOutcome::Backpressure +#[inline] +pub fn record_xshard_backpressure_drop(target_shard: usize) { + if !METRICS_INITIALIZED.load(Ordering::Relaxed) { + return; + } + counter!( + "moon_xshard_backpressure_drops_total", + "target_shard" => target_shard.to_string() + ) + .increment(1); +} + /// Try to open a connection if under the maxclients limit. /// Returns true if the connection was accepted, false if at limit. /// When maxclients is 0, the limit is disabled (unlimited). diff --git a/src/client_registry.rs b/src/client_registry.rs index d08ec8af..e5b438a4 100644 --- a/src/client_registry.rs +++ b/src/client_registry.rs @@ -142,12 +142,15 @@ pub fn register( live: Arc::clone(&live), }; REGISTRY.write().insert(id, entry); + crate::admin::metrics_setup::record_shard_connection_delta(shard, 1.0); live } /// Deregister a client connection. pub fn deregister(id: u64) { - REGISTRY.write().remove(&id); + if let Some(entry) = REGISTRY.write().remove(&id) { + crate::admin::metrics_setup::record_shard_connection_delta(entry.shard, -1.0); + } } /// Update mutable fields for a client (CLIENT SETNAME and similar — rare, diff --git a/src/config.rs b/src/config.rs index ab15f52d..f27356cc 100644 --- a/src/config.rs +++ b/src/config.rs @@ -78,6 +78,14 @@ pub struct ServerConfig { #[arg(long, default_value_t = 16)] pub databases: usize, + /// TCP listen backlog per listening socket. With per-shard SO_REUSEPORT + /// accept the kernel splits the SYN load across shards, so this is a + /// per-socket bound (Redis single-socket default is 511; Moon's + /// historical hardcoded value was 1024). Raise for connection-storm + /// workloads alongside `ulimit -n` and net.core.somaxconn. + #[arg(long = "tcp-backlog", default_value_t = 1024)] + pub tcp_backlog: i32, + /// Require clients to authenticate with this password #[arg(long)] pub requirepass: Option, diff --git a/src/main.rs b/src/main.rs index a87333a4..8d1e9074 100644 --- a/src/main.rs +++ b/src/main.rs @@ -142,6 +142,10 @@ fn main() -> anyhow::Result<()> { // current directory when it already holds pre-v0.2.0 moon data. config.resolve_dir(); + // Apply the TCP listen backlog before any listener binds (S-5, + // `--tcp-backlog`; per-socket — SO_REUSEPORT splits load across shards). + moon::shard::conn_accept::set_tcp_backlog(config.tcp_backlog); + // ── AOF v1→v2 migration (FIX-W3-2): early-exit before normal boot ── // When `--migrate-aof-from` is set, run the migration tool and exit. // This must run BEFORE any shard/AOF initialization so the source diff --git a/src/persistence/wal_v3/record.rs b/src/persistence/wal_v3/record.rs index 5f87c154..1d3a8cab 100644 --- a/src/persistence/wal_v3/record.rs +++ b/src/persistence/wal_v3/record.rs @@ -61,6 +61,10 @@ pub enum WalRecordType { XactCommit = 0x51, /// Cross-store transaction abort. XactAbort = 0x52, + /// Cross-store transaction commit, v2: header carries the logical db + /// index so crash replay restores KV ops into the db the transaction ran + /// in (v1 records replay into db 0 — the pre-multi-db behaviour). + XactCommitV2 = 0x53, /// Workspace creation record. WorkspaceCreate = 0x60, /// Workspace deletion record. @@ -92,6 +96,7 @@ impl WalRecordType { 0x50 => Some(Self::XactBegin), 0x51 => Some(Self::XactCommit), 0x52 => Some(Self::XactAbort), + 0x53 => Some(Self::XactCommitV2), 0x60 => Some(Self::WorkspaceCreate), 0x61 => Some(Self::WorkspaceDrop), 0x70 => Some(Self::MqCreate), @@ -362,6 +367,57 @@ pub fn encode_xact_commit_payload( payload } +/// Encode an [`WalRecordType::XactCommitV2`] payload. +/// +/// Identical to the v1 format except the header carries the logical db index +/// the transaction ran in, so crash replay restores the KV ops into the +/// correct db (v1 records hardcoded db 0 on replay). +/// +/// Layout (little-endian): `txn_id: u64, db_index: u32, kv_op_count: u32, +/// ops…` (op wire format unchanged from v1). +pub fn encode_xact_commit_payload_v2( + txn_id: u64, + db_index: u32, + undo_records: &[crate::transaction::UndoRecord], + db: &crate::storage::Database, +) -> Vec { + let mut payload = Vec::with_capacity(16 + undo_records.len() * 32); + payload.extend_from_slice(&txn_id.to_le_bytes()); + payload.extend_from_slice(&db_index.to_le_bytes()); + // Count placeholder — patched at the end + let count_offset = payload.len(); + payload.extend_from_slice(&0u32.to_le_bytes()); + let mut count = 0u32; + + for record in undo_records { + match record { + crate::transaction::UndoRecord::Insert { key } + | crate::transaction::UndoRecord::Update { key, .. } => { + // Read current (post-dispatch) value for forward-image WAL + if let Some(entry) = db.data().get(key.as_ref()) { + if let Some(value) = entry.value.as_bytes_owned() { + payload.push(0u8); // op_type = SET + payload.extend_from_slice(&(key.len() as u32).to_le_bytes()); + payload.extend_from_slice(key); + payload.extend_from_slice(&(value.len() as u32).to_le_bytes()); + payload.extend_from_slice(&value); + count += 1; + } + } + } + crate::transaction::UndoRecord::Delete { key, .. } => { + payload.push(1u8); // op_type = DEL + payload.extend_from_slice(&(key.len() as u32).to_le_bytes()); + payload.extend_from_slice(key); + count += 1; + } + } + } + // Patch count field + payload[count_offset..count_offset + 4].copy_from_slice(&count.to_le_bytes()); + payload +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/persistence/wal_v3/replay.rs b/src/persistence/wal_v3/replay.rs index c3412e8c..2cdb339b 100644 --- a/src/persistence/wal_v3/replay.rs +++ b/src/persistence/wal_v3/replay.rs @@ -77,6 +77,11 @@ pub fn replay_wal_auto( replay_xact_commit(databases, &record.payload); tracing::trace!(lsn = record.lsn, "WAL replay: XactCommit"); } + WalRecordType::XactCommitV2 => { + // XactCommitV2: db-index-aware KV replay + replay_xact_commit_v2(databases, &record.payload); + tracing::trace!(lsn = record.lsn, "WAL replay: XactCommitV2"); + } WalRecordType::XactAbort => { // XactAbort: no action - changes were never committed tracing::trace!(lsn = record.lsn, "WAL replay: XactAbort"); @@ -310,6 +315,7 @@ pub fn replay_wal_v3_file_until( | WalRecordType::FileTierChange | WalRecordType::XactBegin | WalRecordType::XactCommit + | WalRecordType::XactCommitV2 | WalRecordType::XactAbort | WalRecordType::TemporalUpsert | WalRecordType::GraphTemporal @@ -446,13 +452,57 @@ fn replay_xact_commit(databases: &mut [crate::storage::Database], payload: &[u8] tracing::debug!(txn_id, kv_op_count, "Replaying XactCommit"); - if kv_op_count == 0 { + // v1 records predate the db-index header: they were only ever written by + // builds that replayed into db 0, so db 0 is the faithful target. + replay_xact_kv_ops(&mut databases[0], payload, 12, kv_op_count); +} + +/// Replay a cross-store transaction commit record, v2 (db-index-aware). +/// +/// Payload layout (little-endian): `txn_id: u64, db_index: u32, +/// kv_op_count: u32, ops…` — op wire format identical to v1. +/// An out-of-range `db_index` (server restarted with fewer `--databases`) +/// falls back to db 0 with a loud warning rather than dropping the ops. +fn replay_xact_commit_v2(databases: &mut [crate::storage::Database], payload: &[u8]) { + if payload.len() < 16 { + tracing::warn!("XactCommitV2 payload too short: {} bytes", payload.len()); return; } - let mut offset = 12; - let db = &mut databases[0]; // TODO: support multi-db in cross-store txn + let txn_id = u64::from_le_bytes([ + payload[0], payload[1], payload[2], payload[3], payload[4], payload[5], payload[6], + payload[7], + ]); + let db_index = u32::from_le_bytes([payload[8], payload[9], payload[10], payload[11]]) as usize; + let kv_op_count = + u32::from_le_bytes([payload[12], payload[13], payload[14], payload[15]]) as usize; + + tracing::debug!(txn_id, db_index, kv_op_count, "Replaying XactCommitV2"); + let target = if db_index < databases.len() { + db_index + } else { + tracing::warn!( + txn_id, + db_index, + db_count = databases.len(), + "XactCommitV2 db index out of range (server restarted with fewer \ + --databases?) — replaying into db 0" + ); + 0 + }; + replay_xact_kv_ops(&mut databases[target], payload, 16, kv_op_count); +} + +/// Shared op-loop for XactCommit v1/v2: apply `kv_op_count` SET/DEL ops read +/// from `payload` starting at `offset` to `db`. Truncation-defensive — a +/// short payload warns and stops rather than panicking. +fn replay_xact_kv_ops( + db: &mut crate::storage::Database, + payload: &[u8], + mut offset: usize, + kv_op_count: usize, +) { for _ in 0..kv_op_count { if offset >= payload.len() { tracing::warn!("XactCommit payload truncated at op boundary"); @@ -541,6 +591,61 @@ mod tests { header } + // D-1: XactCommitV2 must replay KV ops into the db the transaction ran + // in — the v1 path hardcoded db 0, silently corrupting recovery for any + // TXN.BEGIN…COMMIT issued under SELECT != 0. + #[test] + fn test_xact_commit_v2_replays_into_selected_db() { + use crate::persistence::wal_v3::record::encode_xact_commit_payload_v2; + use crate::storage::Database; + use crate::transaction::UndoRecord; + use bytes::Bytes; + + // Committed state lives in db 3. + let mut src_db = Database::new(); + src_db.set_string(Bytes::from_static(b"txnkey"), Bytes::from_static(b"v3val")); + let records = vec![UndoRecord::Insert { + key: Bytes::from_static(b"txnkey"), + }]; + let payload = encode_xact_commit_payload_v2(7, 3, &records, &src_db); + + let mut databases: Vec = (0..4).map(|_| Database::new()).collect(); + replay_xact_commit_v2(&mut databases, &payload); + + assert!( + databases[3].data().get(b"txnkey".as_ref()).is_some(), + "XactCommitV2 must restore the key into db 3" + ); + assert!( + databases[0].data().get(b"txnkey".as_ref()).is_none(), + "db 0 must NOT receive a db-3 transaction's keys" + ); + } + + #[test] + fn test_xact_commit_v2_out_of_range_db_falls_back_to_db0() { + use crate::persistence::wal_v3::record::encode_xact_commit_payload_v2; + use crate::storage::Database; + use crate::transaction::UndoRecord; + use bytes::Bytes; + + let mut src_db = Database::new(); + src_db.set_string(Bytes::from_static(b"k"), Bytes::from_static(b"v")); + let records = vec![UndoRecord::Insert { + key: Bytes::from_static(b"k"), + }]; + // db_index 9 but the restarted server only has 2 dbs. + let payload = encode_xact_commit_payload_v2(8, 9, &records, &src_db); + + let mut databases: Vec = (0..2).map(|_| Database::new()).collect(); + replay_xact_commit_v2(&mut databases, &payload); + + assert!( + databases[0].data().get(b"k".as_ref()).is_some(), + "out-of-range db index must fall back to db 0, not drop the ops" + ); + } + #[test] fn test_v3_replay_commands() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/server/conn/core.rs b/src/server/conn/core.rs index 385dbb64..fa2f648d 100644 --- a/src/server/conn/core.rs +++ b/src/server/conn/core.rs @@ -166,6 +166,35 @@ impl ConnectionContext { disk_offload_dir, } } + + /// Build the eviction context for FCALL-internal `redis.call` writes + /// (Gap B — same OOM gate as EVAL/EVALSHA). Used by the lazy + /// `FunctionRegistry` init in both connection handlers: constructing this + /// eagerly per connection cost 6 Arc/Rc clones for the >99% of + /// connections that never call FUNCTION/FCALL. + pub fn build_lua_eviction_ctx(&self) -> crate::scripting::bridge::LuaEvictionCtx { + crate::scripting::bridge::LuaEvictionCtx::new( + self.shard_databases.clone(), + self.runtime_config.clone(), + self.shard_id, + self.spill_sender.clone(), + self.spill_file_id.clone(), + self.disk_offload_dir.clone(), + ) + } +} + +/// Get-or-init helper for the per-connection lazy `FunctionRegistry` slot. +/// Returns after guaranteeing the slot is `Some`. +pub(crate) fn ensure_function_registry( + slot: &RefCell>, + ctx: &ConnectionContext, +) { + if slot.borrow().is_none() { + *slot.borrow_mut() = Some(crate::scripting::FunctionRegistry::new( + ctx.build_lua_eviction_ctx(), + )); + } } /// Per-connection mutable state. diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 500550f0..9dcc48b5 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -1147,23 +1147,33 @@ pub(super) fn try_handle_functions( cmd_args: &[Frame], conn: &ConnectionState, ctx: &ConnectionContext, - func_registry: &Rc>, + func_registry: &Rc>>, responses: &mut Vec, ) -> bool { if conn.in_multi { return false; } if cmd.eq_ignore_ascii_case(b"FUNCTION") { + crate::server::conn::core::ensure_function_registry(func_registry, ctx); + let mut guard = func_registry.borrow_mut(); + #[allow(clippy::unwrap_used)] + // ensure_function_registry guarantees Some let response = - crate::command::functions::handle_function(&mut func_registry.borrow_mut(), cmd_args); + crate::command::functions::handle_function(guard.as_mut().unwrap(), cmd_args); + drop(guard); responses.push(response); return true; } if cmd.eq_ignore_ascii_case(b"FCALL") { + crate::server::conn::core::ensure_function_registry(func_registry, ctx); + let guard = func_registry.borrow(); + #[allow(clippy::unwrap_used)] + // ensure_function_registry guarantees Some + let reg = guard.as_ref().unwrap(); let response = crate::shard::slice::with_shard(|s| { let db_count = s.databases.len(); crate::command::functions::handle_fcall( - &func_registry.borrow(), + reg, cmd_args, &mut s.databases[conn.selected_db], ctx.shard_id, @@ -1172,14 +1182,20 @@ pub(super) fn try_handle_functions( db_count, ) }); + drop(guard); responses.push(response); return true; } if cmd.eq_ignore_ascii_case(b"FCALL_RO") { + crate::server::conn::core::ensure_function_registry(func_registry, ctx); + let guard = func_registry.borrow(); + #[allow(clippy::unwrap_used)] + // ensure_function_registry guarantees Some + let reg = guard.as_ref().unwrap(); let response = crate::shard::slice::with_shard(|s| { let db_count = s.databases.len(); crate::command::functions::handle_fcall_ro( - &func_registry.borrow(), + reg, cmd_args, &mut s.databases[conn.selected_db], ctx.shard_id, @@ -1188,6 +1204,7 @@ pub(super) fn try_handle_functions( db_count, ) }); + drop(guard); responses.push(response); return true; } diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 103979b1..31cfe7bb 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -173,20 +173,14 @@ pub(crate) async fn handle_connection_sharded_monoio< } let _registry_guard = RegistryGuard(client_id); - // Functions API registry (per-connection, lazy init) — kept as local because Rc> is !Send. - // Real eviction ctx (Gap B): FCALL-internal `redis.call` writes must run - // the same OOM gate as EVAL/EVALSHA, same handles `LuaEvictionCtx::new` - // uses elsewhere on this shard (conn_accept.rs's `setup_lua_vm` call). - let func_registry = Rc::new(RefCell::new(crate::scripting::FunctionRegistry::new( - crate::scripting::bridge::LuaEvictionCtx::new( - ctx.shard_databases.clone(), - ctx.runtime_config.clone(), - ctx.shard_id, - ctx.spill_sender.clone(), - ctx.spill_file_id.clone(), - ctx.disk_offload_dir.clone(), - ), - ))); + // Functions API registry — LAZY per connection (P-1 footprint): built on + // first FUNCTION/FCALL/FCALL_RO via `ensure_function_registry`, so the + // >99% of connections that never touch the Functions API pay zero + // registry + eviction-ctx cost. Kept as a local because Rc> is + // !Send. The eviction ctx (Gap B — FCALL-internal `redis.call` writes run + // the same OOM gate as EVAL) is built by `ctx.build_lua_eviction_ctx()`. + let func_registry: Rc>> = + Rc::new(RefCell::new(None)); // Pre-allocate read buffer outside the loop to avoid per-read heap allocation. // Monoio's ownership I/O takes ownership and returns the buffer, so we reassign. diff --git a/src/server/conn/handler_monoio/txn.rs b/src/server/conn/handler_monoio/txn.rs index 03d46680..a2f4b382 100644 --- a/src/server/conn/handler_monoio/txn.rs +++ b/src/server/conn/handler_monoio/txn.rs @@ -33,7 +33,11 @@ pub(super) fn try_handle_txn_begin( // Get next txn_id and snapshot_lsn from vector store's transaction manager let active = crate::shard::slice::with_shard(|s| s.vector_store.txn_manager_mut().begin()); - conn.active_cross_txn = Some(CrossStoreTxn::new(active.txn_id, active.snapshot_lsn)); + conn.active_cross_txn = Some(CrossStoreTxn::new( + active.txn_id, + active.snapshot_lsn, + conn.selected_db, + )); responses.push(Frame::SimpleString(Bytes::from_static(b"OK"))); } Err(e) => responses.push(e), @@ -83,22 +87,24 @@ pub(super) async fn try_handle_txn_commit( return true; } - // Write XactCommit WAL record with committed KV state + // Write XactCommitV2 WAL record with committed KV state. + // Both the forward-image read and the record header use the + // txn's BEGIN-time db so commit and replay agree on the db. let txn_id = txn.txn_id; if !txn.kv_undo.is_empty() { - let payload = - crate::shard::slice::with_shard_db(conn.selected_db as usize, |db| { - crate::persistence::wal_v3::record::encode_xact_commit_payload( - txn_id, - txn.kv_undo.records(), - db, - ) - }); + let payload = crate::shard::slice::with_shard_db(txn.db_index, |db| { + crate::persistence::wal_v3::record::encode_xact_commit_payload_v2( + txn_id, + txn.db_index as u32, + txn.kv_undo.records(), + db, + ) + }); let mut wal_buf = Vec::new(); crate::persistence::wal_v3::record::write_wal_v3_record( &mut wal_buf, txn_id, - crate::persistence::wal_v3::record::WalRecordType::XactCommit, + crate::persistence::wal_v3::record::WalRecordType::XactCommitV2, &payload, ); ctx.shard_databases diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index ae5330f5..85c84f54 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -100,6 +100,10 @@ pub(crate) async fn handle_connection_sharded( }; #[cfg(not(unix))] let kill_fd = -1; + // R-6 fail-open: keep what we need to re-serve the client locally if a + // migration hand-off cannot be delivered. + #[cfg(unix)] + let (shutdown2, peer_addr2) = (shutdown.clone(), peer_addr.clone()); let result = handle_connection_sharded_inner( stream, peer_addr, @@ -179,22 +183,59 @@ pub(crate) async fn handle_connection_sharded( } } if let Some(ShardMessage::MigrateConnection(payload)) = pending { + tracing::warn!( + "Shard {}: migration SPSC full, keeping connection {} local", + ctx.shard_id, + client_id + ); + // Fail-open (R-6): the target never drained — resume + // serving the client on this shard with the state + // recovered from the undelivered message, instead of + // dropping the connection. Migration disabled so the + // affinity tracker cannot loop. use std::os::unix::io::FromRawFd; - // SAFETY: fd is a valid, uniquely-owned file descriptor obtained - // from TcpStream::into_raw_fd() above. OwnedFd closes it on drop. - drop(unsafe { std::os::unix::io::OwnedFd::from_raw_fd(payload.fd) }); + // SAFETY: fd is a valid, uniquely-owned file descriptor + // obtained from TcpStream::into_raw_fd() above; the push + // never succeeded so ownership stayed with us. + let std_stream = + unsafe { std::net::TcpStream::from_raw_fd(payload.fd) }; + match tokio::net::TcpStream::from_std(std_stream) { + Ok(tcp_stream) => { + let _ = handle_connection_sharded_inner( + tcp_stream, + peer_addr2, + ctx, + shutdown2, + client_id, + false, // can_migrate + BytesMut::new(), + Some(&payload.state), + payload.fd, + ) + .await; + } + Err(e) => { + tracing::warn!( + "Shard {}: migration fail-open from_std failed: {} \ + — connection {} lost", + ctx.shard_id, + e, + client_id + ); + } + } + // The connection has now truly ended (served locally to + // completion or lost on from_std failure) — balance the + // accept-side counter that the migration arm skips. + crate::admin::metrics_setup::record_connection_closed(); } - tracing::warn!( - "Shard {}: migration SPSC full, connection {} lost", - ctx.shard_id, - client_id - ); } } } Err(e) => { tracing::warn!("Shard {}: migration into_std failed: {}", ctx.shard_id, e); - // Stream consumed by into_std attempt, connection lost either way + // Stream consumed by into_std attempt, connection lost either way. + crate::admin::metrics_setup::record_connection_closed(); } } } else { @@ -291,20 +332,14 @@ pub(crate) async fn handle_connection_sharded_inner< } let _registry_guard = RegistryGuard(client_id); - // Functions API registry (per-shard, lazy init) — kept as local because Rc> is !Send. - // Real eviction ctx (Gap B): FCALL-internal `redis.call` writes must run - // the same OOM gate as EVAL/EVALSHA, same handles `LuaEvictionCtx::new` - // uses elsewhere on this shard (conn_accept.rs's `setup_lua_vm` call). - let func_registry = std::rc::Rc::new(std::cell::RefCell::new( - crate::scripting::FunctionRegistry::new(crate::scripting::bridge::LuaEvictionCtx::new( - ctx.shard_databases.clone(), - ctx.runtime_config.clone(), - ctx.shard_id, - ctx.spill_sender.clone(), - ctx.spill_file_id.clone(), - ctx.disk_offload_dir.clone(), - )), - )); + // Functions API registry — LAZY per connection (P-1 footprint): built on + // first FUNCTION/FCALL/FCALL_RO via `ensure_function_registry`, so the + // >99% of connections that never touch the Functions API pay zero + // registry + eviction-ctx cost. Kept as a local because Rc> is + // !Send. The eviction ctx (Gap B — FCALL-internal `redis.call` writes run + // the same OOM gate as EVAL) is built by `ctx.build_lua_eviction_ctx()`. + let func_registry: std::rc::Rc>> = + std::rc::Rc::new(std::cell::RefCell::new(None)); // Per-connection arena for batch processing temporaries. // 4KB initial capacity, grows on demand (rarely exceeds 16KB per batch). @@ -758,33 +793,50 @@ pub(crate) async fn handle_connection_sharded_inner< // fall through to the MULTI queue gate instead of executing. if !conn.in_multi { if cmd.eq_ignore_ascii_case(b"FUNCTION") { + crate::server::conn::core::ensure_function_registry(&func_registry, ctx); + let mut guard = func_registry.borrow_mut(); + #[allow(clippy::unwrap_used)] + // ensure_function_registry guarantees Some let response = crate::command::functions::handle_function( - &mut func_registry.borrow_mut(), cmd_args, + guard.as_mut().unwrap(), cmd_args, ); + drop(guard); responses.push(response); continue; } if cmd.eq_ignore_ascii_case(b"FCALL") { + crate::server::conn::core::ensure_function_registry(&func_registry, ctx); let db_count = ctx.shard_databases.db_count(); + let guard = func_registry.borrow(); + #[allow(clippy::unwrap_used)] + // ensure_function_registry guarantees Some + let reg = guard.as_ref().unwrap(); // Unconditional slice path: ShardSlice is always initialized. let response = crate::shard::slice::with_shard_db(conn.selected_db, |db| { crate::command::functions::handle_fcall( - &func_registry.borrow(), cmd_args, db, + reg, cmd_args, db, ctx.shard_id, ctx.num_shards, conn.selected_db, db_count, ) }); + drop(guard); responses.push(response); continue; } if cmd.eq_ignore_ascii_case(b"FCALL_RO") { + crate::server::conn::core::ensure_function_registry(&func_registry, ctx); let db_count = ctx.shard_databases.db_count(); + let guard = func_registry.borrow(); + #[allow(clippy::unwrap_used)] + // ensure_function_registry guarantees Some + let reg = guard.as_ref().unwrap(); // Unconditional slice path: ShardSlice is always initialized. let response = crate::shard::slice::with_shard_db(conn.selected_db, |db| { crate::command::functions::handle_fcall_ro( - &func_registry.borrow(), cmd_args, db, + reg, cmd_args, db, ctx.shard_id, ctx.num_shards, conn.selected_db, db_count, ) }); + drop(guard); responses.push(response); continue; } diff --git a/src/server/conn/handler_sharded/txn.rs b/src/server/conn/handler_sharded/txn.rs index 3112d7fa..cac87e1d 100644 --- a/src/server/conn/handler_sharded/txn.rs +++ b/src/server/conn/handler_sharded/txn.rs @@ -34,7 +34,11 @@ pub(super) fn try_handle_txn_begin( // Unconditional slice path: ShardSlice is always initialized. let active = crate::shard::slice::with_shard(|s| s.vector_store.txn_manager_mut().begin()); - conn.active_cross_txn = Some(CrossStoreTxn::new(active.txn_id, active.snapshot_lsn)); + conn.active_cross_txn = Some(CrossStoreTxn::new( + active.txn_id, + active.snapshot_lsn, + conn.selected_db, + )); responses.push(Frame::SimpleString(Bytes::from_static(b"OK"))); } Err(e) => responses.push(e), @@ -83,13 +87,16 @@ pub(super) async fn try_handle_txn_commit( return true; } - // Write XactCommit WAL record with committed KV state + // Write XactCommitV2 WAL record with committed KV state. + // Both the forward-image read and the record header use the + // txn's BEGIN-time db so commit and replay agree on the db. let txn_id = txn.txn_id; if !txn.kv_undo.is_empty() { // Unconditional slice path: ShardSlice is always initialized. - let payload = crate::shard::slice::with_shard_db(conn.selected_db, |db| { - crate::persistence::wal_v3::record::encode_xact_commit_payload( + let payload = crate::shard::slice::with_shard_db(txn.db_index, |db| { + crate::persistence::wal_v3::record::encode_xact_commit_payload_v2( txn_id, + txn.db_index as u32, txn.kv_undo.records(), db, ) @@ -98,7 +105,7 @@ pub(super) async fn try_handle_txn_commit( crate::persistence::wal_v3::record::write_wal_v3_record( &mut wal_buf, txn_id, - crate::persistence::wal_v3::record::WalRecordType::XactCommit, + crate::persistence::wal_v3::record::WalRecordType::XactCommitV2, &payload, ); ctx.shard_databases diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index aabd0192..61a8fd3d 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -514,7 +514,7 @@ mod as_of_tests { fn resolve_ft_search_as_of_lsn_uses_txn_snapshot_when_no_explicit_as_of() { let fixture = build_fixture(); let args = ft_search_args(None); - let txn = CrossStoreTxn::new(1, 99); + let txn = CrossStoreTxn::new(1, 99, 0); let got = resolve_ft_search_as_of_lsn(&args, Some(&fixture), Some(&txn)); assert_eq!(got, Ok(99)); } @@ -523,7 +523,7 @@ mod as_of_tests { fn resolve_ft_search_as_of_lsn_explicit_as_of_beats_txn_snapshot() { let fixture = build_fixture(); let args = ft_search_args(Some(1_000)); - let txn = CrossStoreTxn::new(1, 99); + let txn = CrossStoreTxn::new(1, 99, 0); let got = resolve_ft_search_as_of_lsn(&args, Some(&fixture), Some(&txn)); // Registry binding at wall_ms=1_000 is lsn=42, NOT txn.snapshot_lsn=99. assert_eq!(got, Ok(42)); diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index 536c0cf9..63f7331e 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -31,6 +31,18 @@ use super::shared_databases::ShardDatabases; /// from parking_lot::RwLock (used for pubsub types). type StdRwLock = std::sync::RwLock; +/// Process-wide TCP listen backlog (S-5, `--tcp-backlog`). Set once at +/// startup before any listener binds; 1024 is the historical hardcoded +/// value. A process-wide static (rather than threading a param through the +/// four bind sites) because backlog is a bind-time server property. +static TCP_BACKLOG: std::sync::atomic::AtomicI32 = std::sync::atomic::AtomicI32::new(1024); + +/// Set the TCP listen backlog applied to every subsequently-bound listener. +/// Called once from startup after config parsing. +pub fn set_tcp_backlog(backlog: i32) { + TCP_BACKLOG.store(backlog.max(1), std::sync::atomic::Ordering::Relaxed); +} + /// Create a SO_REUSEPORT TCP listener socket using socket2. /// /// Returns a `std::net::TcpListener` that can be converted to @@ -56,7 +68,7 @@ pub(crate) fn create_reuseport_socket(addr: &str) -> std::io::Result { - _migrated = true; - notifiers2[target_shard].notify_one(); - tracing::info!( - "Shard {}: migrated connection {} to shard {} (monoio)", - shard_id, cid, target_shard - ); + // Bounded retry: migration is elective — a briefly-full + // ring is worth a few 100µs waits, but a target that + // stays full gets the fail-open path, never a lost conn. + for attempt in 0..8u32 { + if attempt > 0 { + monoio::time::sleep(std::time::Duration::from_micros(100)).await; } - Err(returned_msg) => { - if let ShardMessage::MigrateConnection(payload) = returned_msg { - // SAFETY: fd is a valid dup'd socket from libc::dup above. - // SPSC push failed, so ownership returns to us for cleanup. - drop(unsafe { - std::os::unix::io::OwnedFd::from_raw_fd(payload.fd) - }); + #[allow(clippy::unwrap_used)] + // pending_msg is always re-filled on Err below + let msg = pending_msg.take().unwrap(); + let push_result = { + let mut producers = dtx2.borrow_mut(); + producers[target_idx].try_push(msg) + }; + match push_result { + Ok(()) => { + _migrated = true; + notifiers2[target_shard].notify_one(); + tracing::info!( + "Shard {}: migrated connection {} to shard {} (monoio)", + shard_id, cid, target_shard + ); + break; } - tracing::warn!( - "Shard {}: migration SPSC full, connection {} lost (monoio)", - shard_id, cid - ); + Err(returned_msg) => pending_msg = Some(returned_msg), } } + if _migrated { + drop(stream); // hand-off confirmed: target owns dup_fd + } else if let Some(ShardMessage::MigrateConnection(payload)) = pending_msg { + // SAFETY: dup_fd is a valid dup'd socket from libc::dup + // above; the push never succeeded so we still own it. + drop(unsafe { + std::os::unix::io::OwnedFd::from_raw_fd(payload.fd) + }); + tracing::warn!( + "Shard {}: migration SPSC full, keeping connection {} local (monoio)", + shard_id, cid + ); + // Fail-open (R-6): resume serving on this shard with + // the state recovered from the undelivered message. + let _ = handle_connection_sharded_monoio( + stream, peer2, &conn_ctx, sd2, cid, + false, // can_migrate: pin locally, no retry loop + BytesMut::new(), pw2, + Some(&payload.state), kill_fd, + ) + .await; + } } } diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index eb9c43e2..c46f304f 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -891,6 +891,14 @@ async fn spsc_send_bounded( // Budget exhausted: the target ring never drained. Drop `pending` (and any // embedded reply sender) so awaiting callers fail loud instead of hanging. + // Rare by construction (~0.5s of failed retries), so the warn + labeled + // counter cannot flood. + tracing::warn!( + my_shard, + target_shard, + "cross-shard dispatch dropped after backpressure budget — target not draining" + ); + crate::admin::metrics_setup::record_xshard_backpressure_drop(target_shard); PushOutcome::Backpressure } diff --git a/src/transaction/mod.rs b/src/transaction/mod.rs index 3b694689..6f678ddb 100644 --- a/src/transaction/mod.rs +++ b/src/transaction/mod.rs @@ -97,6 +97,13 @@ pub struct CrossStoreTxn { pub txn_id: u64, /// Snapshot LSN for reads. pub snapshot_lsn: u64, + /// Logical db index the connection had SELECTed at `TXN.BEGIN`. Written + /// into the `XactCommitV2` WAL header so crash replay restores the KV ops + /// into this db (previously replay hardcoded db 0). KV writes inside the + /// txn use the connection's live `selected_db` per command; a mid-txn + /// `SELECT` is a pre-existing single-db assumption shared with the + /// forward-image encoder, which reads back from one db at commit. + pub db_index: usize, /// KV undo log for rollback. pub kv_undo: UndoLog, /// Vector point_ids modified in this transaction. @@ -112,12 +119,14 @@ pub struct CrossStoreTxn { } impl CrossStoreTxn { - /// Create a new cross-store transaction. + /// Create a new cross-store transaction bound to the connection's + /// currently SELECTed logical db. #[inline] - pub fn new(txn_id: u64, snapshot_lsn: u64) -> Self { + pub fn new(txn_id: u64, snapshot_lsn: u64, db_index: usize) -> Self { Self { txn_id, snapshot_lsn, + db_index, kv_undo: UndoLog::new(), vector_intents: SmallVec::new(), graph_intents: SmallVec::new(), @@ -215,7 +224,7 @@ mod tests { #[test] fn test_cross_store_txn_new() { - let txn = CrossStoreTxn::new(42, 41); + let txn = CrossStoreTxn::new(42, 41, 0); assert_eq!(txn.txn_id, 42); assert_eq!(txn.snapshot_lsn, 41); assert!(!txn.has_modifications()); @@ -223,7 +232,7 @@ mod tests { #[test] fn test_has_modifications_kv() { - let mut txn = CrossStoreTxn::new(1, 0); + let mut txn = CrossStoreTxn::new(1, 0, 0); assert!(!txn.has_modifications()); txn.record_kv_insert(Bytes::from_static(b"key")); assert!(txn.has_modifications()); @@ -231,21 +240,21 @@ mod tests { #[test] fn test_has_modifications_vector() { - let mut txn = CrossStoreTxn::new(1, 0); + let mut txn = CrossStoreTxn::new(1, 0, 0); txn.record_vector(100, Bytes::from_static(b"idx")); assert!(txn.has_modifications()); } #[test] fn test_has_modifications_graph() { - let mut txn = CrossStoreTxn::new(1, 0); + let mut txn = CrossStoreTxn::new(1, 0, 0); txn.record_graph(200, true, Bytes::from_static(b"g")); assert!(txn.has_modifications()); } #[test] fn test_graph_intent_records_graph_name() { - let mut txn = CrossStoreTxn::new(1, 0); + let mut txn = CrossStoreTxn::new(1, 0, 0); txn.record_graph(10, true, Bytes::from_static(b"g1")); txn.record_graph(11, false, Bytes::from_static(b"g2")); assert_eq!(txn.graph_intents.len(), 2); @@ -259,7 +268,7 @@ mod tests { #[test] fn test_graph_intent_reverse_order_is_lifo() { - let mut txn = CrossStoreTxn::new(1, 0); + let mut txn = CrossStoreTxn::new(1, 0, 0); // Push three intents; reverse iteration must yield them LIFO so Plan // 166-03 can remove edges before their endpoint nodes on rollback. txn.record_graph(1, true, Bytes::from_static(b"g")); @@ -278,7 +287,7 @@ mod tests { #[test] fn test_mq_intent_tracking() { - let mut txn = CrossStoreTxn::new(10, 9); + let mut txn = CrossStoreTxn::new(10, 9, 0); assert!(!txn.has_modifications()); txn.record_mq( From f1a01a5cf48613c454919e4d9ecccff131ce66c1 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 10:41:24 +0700 Subject: [PATCH 4/6] =?UTF-8?q?fix(shard):=20D-2=20=E2=80=94=20FLUSHDB/FLU?= =?UTF-8?q?SHALL=20broadcast=20to=20every=20shard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FLUSHDB and FLUSHALL are keyless commands, so key-based routing executed them only on the shard that owned the issuing connection. On a `--shards 4` server a client FLUSHALL therefore left ~3/4 of the keyspace intact (red/green measured 49/64 keys surviving; DBSIZE scatter-gathers, so the survivors were directly visible — and served stale reads as ghost keys). Fix: after the local shard executes the flush, the originating handler calls the new `coordinate_flush_broadcast` (src/shard/coordinator.rs), which sends the original FLUSH frame to every peer shard as a single-command `ShardMessage::MultiExecute` leg. Reusing MultiExecute means each remote shard gets, for free, the existing remote-arm behavior: dispatch execution, per-shard AOF/WAL persistence via `wal_append_and_fanout` (fail-loud `AOF_APPEND_LOST_ERR`), and `auto_flush_indexes`. The broadcast collects every leg's ack; any failed / unreachable leg turns the client reply into a loud `MOONERR FLUSH partial: shard N leg failed or unreachable — local shard flushed; retry the FLUSH command` plus a `tracing::error!`, never a silent partial flush. Hooked in both sharded handler stacks: - handler_sharded (tokio): after the R3 auto_flush_indexes block. - handler_monoio: after the with_shard closure exits (the broadcast awaits cross-shard acks, so it cannot run inside the closure), guarded on a successful local response. Red/green TDD (tests/flush_cross_shard_scatter.rs, spawns a real `--shards 4` server, honors MOON_BIN): - RED (pre-fix HEAD binary): 0/3 — FLUSHALL 49/64 survived, FLUSHDB 20/32 survived, db-1 FLUSHDB left db-1 keys on remote shards. - GREEN (fixed binary): 3/3 — DBSIZE 0 on all shards, ghost-key GETs empty, db-0 keys survive a db-1 FLUSHDB. (The first red run false-passed: the harness passed a nonexistent `--persistence-dir` flag so the server never started and the tests skipped; fixed to `--dir` before trusting either color.) Documented limits (CHANGELOG): the flush is not atomic across shards (same relaxed semantics as SWAPDB); FLUSH inside MULTI/EXEC still executes local-only (follow-up); FLUSHALL still clears only the selected db per the documented v0.1.5 single-active-DB behavior — all-dbs semantics needs dispatch+replay parity and is a separate follow-up. Refs: tmp/CONN-DEEP-REVIEW.md D-2 (Tier 1), PR #230 author: Tin Dang --- CHANGELOG.md | 14 ++ src/server/conn/handler_monoio/mod.rs | 23 +++ src/server/conn/handler_sharded/mod.rs | 19 +++ src/shard/coordinator.rs | 70 ++++++++ tests/flush_cross_shard_scatter.rs | 221 +++++++++++++++++++++++++ 5 files changed, 347 insertions(+) create mode 100644 tests/flush_cross_shard_scatter.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 63b5ac54..cee12a13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -134,6 +134,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 observed on the macOS CI runner). ### Fixed — connection-plane durability & lifecycle wave (PR #230) +- **D-2 — FLUSHDB/FLUSHALL now clear every shard, not just the local one** + (`src/shard/coordinator.rs` `coordinate_flush_broadcast`, both sharded + handlers): FLUSHDB/FLUSHALL are keyless, so key-based routing executed them + only on the issuing connection's shard — on `--shards 4` a FLUSHALL left + ~3/4 of the keyspace intact (red/green: 49/64 keys survived, now 0). The + originating shard now broadcasts the flush to every peer shard as a + `MultiExecute` leg, which reuses the existing remote dispatch + per-shard + AOF/WAL persistence + index-flush path. Any failed leg returns a loud + `MOONERR FLUSH partial` error (local shard already flushed; client should + retry). Known limits: the flush is not atomic across shards (same relaxed + semantics as SWAPDB), and a FLUSH issued inside MULTI/EXEC still executes + local-only (follow-up); FLUSHALL still clears only the selected db + (documented v0.1.5 single-active-DB behavior — all-dbs semantics is a + separate follow-up needing replay parity). - **D-1 — cross-store TXN crash replay restores into the correct db** (`src/transaction/mod.rs`, `wal_v3/record.rs`, `wal_v3/replay.rs`, txn handlers): `CrossStoreTxn` had no db field and WAL replay hardcoded diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 31cfe7bb..54f3b980 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1445,6 +1445,29 @@ pub(crate) async fn handle_connection_sharded_monoio< } } + // D-2: keyless FLUSHDB/FLUSHALL routed local-only cleared just + // this shard — broadcast to every other shard (outside the + // with_shard closure: this awaits). Any failed leg turns the + // reply into an explicit partial-flush error, never silent +OK. + if !matches!(response, Frame::Error(_)) + && ctx.num_shards > 1 + && (cmd.eq_ignore_ascii_case(b"FLUSHDB") + || cmd.eq_ignore_ascii_case(b"FLUSHALL")) + { + if let Err(e) = crate::shard::coordinator::coordinate_flush_broadcast( + &frame, + ctx.shard_id, + ctx.num_shards, + conn.selected_db, + &ctx.dispatch_tx, + &ctx.spsc_notifiers, + ) + .await + { + response = e; + } + } + // AOF logging for successful local writes. // H1: durable path awaits fsync under appendfsync=always. // On AOF failure we override `response` to an error diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 85c84f54..2645226d 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -1535,6 +1535,25 @@ pub(crate) async fn handle_connection_sharded_inner< &mut s.text_store, ); }); + // D-2: keyless flush routed local-only cleared just this + // shard — broadcast to every other shard so the whole + // keyspace flushes. Any failed leg turns the reply into + // an explicit partial-flush error (never silent +OK). + if ctx.num_shards > 1 { + if let Err(e) = + crate::shard::coordinator::coordinate_flush_broadcast( + &frame, + ctx.shard_id, + ctx.num_shards, + conn.selected_db, + &ctx.dispatch_tx, + &ctx.spsc_notifiers, + ) + .await + { + response = e; + } + } } // H1: durable path under appendfsync=always. let mut aof_failed = false; diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index c46f304f..a89747bf 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -902,6 +902,76 @@ async fn spsc_send_bounded( PushOutcome::Backpressure } +/// Broadcast a keyless flush (FLUSHDB/FLUSHALL) to every OTHER shard (D-2). +/// +/// FLUSHDB/FLUSHALL are keyless, so `extract_primary_key` routes them +/// local-only — a normal client's flush previously cleared just the local +/// shard's selected db, silently leaving the other `num_shards - 1` shards' +/// keyspaces intact (ghost keys, stale reads, DBSIZE > 0 after FLUSHALL). +/// +/// Each remote leg is shipped as a `MultiExecute` carrying the original +/// command frame, so the target shard runs it through its normal SPSC arm: +/// dispatch + per-shard AOF/WAL persistence (fail-loud `AOF_APPEND_LOST`) + +/// vector/text index clearing all apply exactly as for the local leg. +/// +/// Legs run concurrently (all sends first, then all acks awaited). Like +/// SWAPDB's broadcast, this is not atomic across shards: a concurrent read +/// can observe shard A flushed while shard B is not yet — Redis-cluster- +/// relaxed semantics. On any failed leg the caller receives an explicit +/// partial-flush error naming the shard, never a silent partial `+OK`. +pub(crate) async fn coordinate_flush_broadcast( + command: &Frame, + my_shard: usize, + num_shards: usize, + db_index: usize, + dispatch_tx: &Rc>>>, + spsc_notifiers: &[Arc], +) -> Result<(), Frame> { + let mut pending = Vec::with_capacity(num_shards.saturating_sub(1)); + for target in 0..num_shards { + if target == my_shard { + continue; + } + let (reply_tx, reply_rx) = channel::oneshot(); + let msg = ShardMessage::MultiExecute { + db_index, + // Keyless command: the routing key is unused by the flush arm. + commands: vec![(Bytes::new(), command.clone())], + reply_tx, + }; + let outcome = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + pending.push((target, reply_rx, outcome)); + } + + let mut failed: Option = None; + for (target, reply_rx, outcome) in pending { + if outcome != crate::shard::dispatch::PushOutcome::Pushed { + failed.get_or_insert(target); + continue; + } + match reply_rx.recv().await { + Ok(frames) if frames.iter().all(|f| !matches!(f, Frame::Error(_))) => {} + _ => { + failed.get_or_insert(target); + } + } + } + match failed { + None => Ok(()), + Some(target) => { + tracing::error!( + my_shard, + target, + "flush broadcast: remote leg failed — keyspace partially flushed" + ); + Err(Frame::Error(Bytes::from(format!( + "MOONERR FLUSH partial: shard {target} leg failed or unreachable — \ + local shard flushed; retry the FLUSH command" + )))) + } + } +} + /// Coordinate MGET across shards using VLL pattern. /// /// Groups keys by shard in a BTreeMap (ascending shard-ID order), executes diff --git a/tests/flush_cross_shard_scatter.rs b/tests/flush_cross_shard_scatter.rs new file mode 100644 index 00000000..db8be9f0 --- /dev/null +++ b/tests/flush_cross_shard_scatter.rs @@ -0,0 +1,221 @@ +//! D-2 red/green: FLUSHDB/FLUSHALL must clear the WHOLE keyspace on a +//! multi-shard server, not just the shard the issuing connection landed on. +//! +//! Before the fix, FLUSHDB/FLUSHALL were keyless → routed local-only, so on +//! `--shards 4` a client's FLUSHALL left ~3/4 of the keys intact (DBSIZE +//! scatter-gathers across shards, so it exposes the survivors directly). +//! +//! Spawns the release moon binary and shells out to `redis-cli`. Skips +//! gracefully when the binary or `redis-cli` is missing. +//! +//! Run with: +//! cargo test --release --test flush_cross_shard_scatter + +use std::net::TcpListener; +use std::process::{Child, Command, Stdio}; +use std::thread; +use std::time::{Duration, Instant}; + +fn free_port() -> u16 { + let l = TcpListener::bind("127.0.0.1:0").expect("bind 127.0.0.1:0"); + l.local_addr().unwrap().port() +} + +fn redis_cli_available() -> bool { + Command::new("redis-cli") + .arg("--version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|s| s.success()) + .unwrap_or(false) +} + +fn release_binary() -> std::path::PathBuf { + // MOON_BIN pin wins (VM-local target dirs); fall back to target/release. + if let Ok(p) = std::env::var("MOON_BIN") { + return std::path::PathBuf::from(p); + } + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("target/release/moon") +} + +struct Moon { + child: Child, + port: u16, + tmp_dir: std::path::PathBuf, +} + +impl Drop for Moon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_dir_all(&self.tmp_dir); + } +} + +/// Start moon with `--shards 4` on a fresh port + fresh dir. +fn spawn_moon_4shard() -> Option { + if !redis_cli_available() { + eprintln!("skipping: redis-cli not in PATH"); + return None; + } + let bin = release_binary(); + if !bin.exists() { + eprintln!( + "skipping: {} not built. Run `cargo build --release` first.", + bin.display() + ); + return None; + } + let port = free_port(); + let tmp_dir = std::env::temp_dir().join(format!("moon-flush-scatter-{port}")); + let _ = std::fs::create_dir_all(&tmp_dir); + let child = Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + "4", + "--admin-port", + "0", + "--appendonly", + "no", + "--dir", + tmp_dir.to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .ok()?; + let moon = Moon { + child, + port, + tmp_dir, + }; + + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if redis_cli(moon.port, &["PING"]) + .map(|out| out.trim() == "PONG") + .unwrap_or(false) + { + return Some(moon); + } + thread::sleep(Duration::from_millis(100)); + } + eprintln!("skipping: moon did not respond to PING within 10s on port {port}"); + None +} + +fn redis_cli(port: u16, args: &[&str]) -> Option { + let output = Command::new("redis-cli") + .args(["-p", &port.to_string()]) + .args(args) + .output() + .ok()?; + Some(String::from_utf8_lossy(&output.stdout).into_owned()) +} + +/// Populate `n` string keys with distinct names — enough spread that all 4 +/// shards own several (key→shard is a hash of the key bytes). +fn seed_keys(port: u16, prefix: &str, n: usize) { + for i in 0..n { + let key = format!("{prefix}:{i}"); + assert_eq!( + redis_cli(port, &["SET", &key, "v"]).unwrap().trim(), + "OK", + "seed SET {key} failed" + ); + } +} + +#[test] +fn flushall_clears_all_shards() { + let Some(m) = spawn_moon_4shard() else { return }; + + seed_keys(m.port, "fa", 64); + let before: i64 = redis_cli(m.port, &["DBSIZE"]) + .unwrap() + .trim() + .parse() + .unwrap(); + assert_eq!(before, 64, "seed must land 64 keys across the shards"); + + assert_eq!(redis_cli(m.port, &["FLUSHALL"]).unwrap().trim(), "OK"); + + // DBSIZE scatter-gathers across all shards — any surviving remote-shard + // key shows up here. Before the D-2 fix this reported ~48 (3/4 of 64). + let after: i64 = redis_cli(m.port, &["DBSIZE"]) + .unwrap() + .trim() + .parse() + .unwrap(); + assert_eq!( + after, 0, + "FLUSHALL must clear every shard's keyspace, {after} keys survived" + ); + + // Individual GETs on a sample confirm no ghost keys serve stale reads. + for i in [0usize, 17, 33, 63] { + let key = format!("fa:{i}"); + let got = redis_cli(m.port, &["GET", &key]).unwrap(); + assert_eq!(got.trim(), "", "ghost key {key} survived FLUSHALL"); + } +} + +#[test] +fn flushdb_clears_selected_db_on_all_shards() { + let Some(m) = spawn_moon_4shard() else { return }; + + seed_keys(m.port, "fd", 32); + assert_eq!(redis_cli(m.port, &["FLUSHDB"]).unwrap().trim(), "OK"); + let after: i64 = redis_cli(m.port, &["DBSIZE"]) + .unwrap() + .trim() + .parse() + .unwrap(); + assert_eq!( + after, 0, + "FLUSHDB must clear the selected db on every shard, {after} keys survived" + ); +} + +#[test] +fn flushdb_other_db_survives_scatter() { + let Some(m) = spawn_moon_4shard() else { return }; + + // Keys in db 0 must survive a FLUSHDB issued against db 1. + seed_keys(m.port, "keep", 16); + // Seed db 1 through the same connection sequence (-n selects the db). + for i in 0..16 { + let key = format!("gone:{i}"); + let out = Command::new("redis-cli") + .args(["-p", &m.port.to_string(), "-n", "1", "SET", &key, "v"]) + .output() + .unwrap(); + assert_eq!(String::from_utf8_lossy(&out.stdout).trim(), "OK"); + } + + // FLUSHDB on db 1: clears db 1 across shards, leaves db 0 intact. + let out = Command::new("redis-cli") + .args(["-p", &m.port.to_string(), "-n", "1", "FLUSHDB"]) + .output() + .unwrap(); + assert_eq!(String::from_utf8_lossy(&out.stdout).trim(), "OK"); + + let db1: String = { + let out = Command::new("redis-cli") + .args(["-p", &m.port.to_string(), "-n", "1", "DBSIZE"]) + .output() + .unwrap(); + String::from_utf8_lossy(&out.stdout).trim().to_string() + }; + assert_eq!(db1, "0", "db 1 must be empty on every shard"); + + let db0: i64 = redis_cli(m.port, &["DBSIZE"]) + .unwrap() + .trim() + .parse() + .unwrap(); + assert_eq!(db0, 16, "db 0 keys must survive a db-1 FLUSHDB"); +} From e4743130946b7eb89b86097b53ac4bc814f67121 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 11:00:42 +0700 Subject: [PATCH 5/6] chore(server): satisfy audit-unsafe 3-line SAFETY-marker rule Main (RSS/CPU wave 5) tightened scripts/audit-unsafe.sh: the literal `SAFETY:` token must sit within the 3 lines directly above the `unsafe {` line. Three Track B blocks had multi-line SAFETY comments whose marker line drifted out of range after the rebase; restructured the comments (context prose first, SAFETY invariant last) with no semantic change. audit-unsafe + audit-unwrap + fmt + check green on both runtimes. author: Tin Dang --- src/client_registry.rs | 7 ++++--- src/server/conn/handler_sharded/mod.rs | 5 ++--- src/shard/conn_accept.rs | 11 ++++++----- 3 files changed, 12 insertions(+), 11 deletions(-) diff --git a/src/client_registry.rs b/src/client_registry.rs index e5b438a4..b81d9036 100644 --- a/src/client_registry.rs +++ b/src/client_registry.rs @@ -250,11 +250,12 @@ pub fn kill_clients(filter: &KillFilter, self_id: Option) -> u64 { fn force_close_fd(fd: i32) { #[cfg(unix)] if fd >= 0 { + // Per the lock-ordering note in `kill_clients`, `fd` is a live socket + // for the duration of this call. We intentionally ignore the return + // value: an already-closed/half-closed socket is a benign no-op. // SAFETY: `libc::shutdown` is an FFI call that cannot cause memory // unsafety for any integer argument — an invalid fd merely returns - // `EBADF`. Per the lock-ordering note in `kill_clients`, `fd` is a live - // socket for the duration of this call. We intentionally ignore the - // return value: a already-closed/half-closed socket is a benign no-op. + // `EBADF`. unsafe { libc::shutdown(fd, libc::SHUT_RDWR); } diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 2645226d..61192e04 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -194,9 +194,8 @@ pub(crate) async fn handle_connection_sharded( // dropping the connection. Migration disabled so the // affinity tracker cannot loop. use std::os::unix::io::FromRawFd; - // SAFETY: fd is a valid, uniquely-owned file descriptor - // obtained from TcpStream::into_raw_fd() above; the push - // never succeeded so ownership stayed with us. + // SAFETY: fd is a valid, uniquely-owned descriptor from + // into_raw_fd(); the failed push left ownership with us. let std_stream = unsafe { std::net::TcpStream::from_raw_fd(payload.fd) }; match tokio::net::TcpStream::from_std(std_stream) { diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index 63f7331e..9f5d89e6 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -769,11 +769,12 @@ pub(crate) fn spawn_monoio_connection( use crate::shard::mesh::ChannelMesh; let raw_fd = stream.as_raw_fd(); - // SAFETY: raw_fd is a valid open socket fd from the monoio TcpStream. - // dup() creates a new, independent fd that we take ownership of. - // The ORIGINAL stream stays alive until the hand-off is confirmed - // (R-6 fail-open): if the push cannot be delivered we keep - // serving the client here instead of dropping it. + // The ORIGINAL stream stays alive until the hand-off is + // confirmed (R-6 fail-open): if the push cannot be delivered + // we keep serving the client here instead of dropping it. + // SAFETY: raw_fd is a valid open socket fd from the monoio + // TcpStream; dup() creates a new, independent fd that we + // take ownership of. let dup_fd = unsafe { libc::dup(raw_fd) }; if dup_fd < 0 { tracing::warn!("Shard {}: migration dup() failed: {} — keeping connection local", shard_id, std::io::Error::last_os_error()); From 842f17ce3b19a172cae6d95fbe517fa18846edd7 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 7 Jul 2026 11:10:59 +0700 Subject: [PATCH 6/6] test(server): add tcp_backlog to exhaustive ServerConfig test literals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit S-5 added `tcp_backlog` to ServerConfig; three integration-test files build the struct with exhaustive literals (no ..Default spread) and failed to compile in CI (E0063). Added `tcp_backlog: 1024` (the default) to the 4 literals. Verified with `cargo test --no-run` on both runtimes (0 errors) — the local gate previously compiled only --lib, which is how this slipped past. author: Tin Dang --- tests/mq_integration.rs | 1 + tests/txn_kv_wiring.rs | 1 + tests/workspace_integration.rs | 2 ++ 3 files changed, 4 insertions(+) diff --git a/tests/mq_integration.rs b/tests/mq_integration.rs index 9dac2eaa..c5f690d5 100644 --- a/tests/mq_integration.rs +++ b/tests/mq_integration.rs @@ -41,6 +41,7 @@ async fn start_mq_server(num_shards: usize) -> (u16, CancellationToken) { let config = ServerConfig { bind: "127.0.0.1".to_string(), port, + tcp_backlog: 1024, databases: 16, requirepass: None, appendonly: "no".to_string(), diff --git a/tests/txn_kv_wiring.rs b/tests/txn_kv_wiring.rs index 41d27b8d..a8bba67d 100644 --- a/tests/txn_kv_wiring.rs +++ b/tests/txn_kv_wiring.rs @@ -46,6 +46,7 @@ async fn start_txn_server(num_shards: usize, persistence_dir: &str) -> (u16, Can let config = ServerConfig { bind: "127.0.0.1".to_string(), port, + tcp_backlog: 1024, databases: 16, requirepass: None, appendonly, diff --git a/tests/workspace_integration.rs b/tests/workspace_integration.rs index 3817514a..2636ce73 100644 --- a/tests/workspace_integration.rs +++ b/tests/workspace_integration.rs @@ -34,6 +34,7 @@ async fn start_workspace_server(num_shards: usize) -> (u16, CancellationToken) { let config = ServerConfig { bind: "127.0.0.1".to_string(), port, + tcp_backlog: 1024, databases: 16, requirepass: None, appendonly: "no".to_string(), @@ -264,6 +265,7 @@ async fn start_workspace_server_with_auth( let config = ServerConfig { bind: "127.0.0.1".to_string(), port, + tcp_backlog: 1024, databases: 16, requirepass: Some(password), appendonly: "no".to_string(),