fix(conn): one byte no longer hides a connection from the idle sweep (c10k D2)

CHANGELOG.md

@@ -39,6 +39,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   whether a given host is affected.
 
 ### Security
+- **One byte made a connection invisible to the idle sweep (c10k D2).** The
+  stage-2 park predicate required an EMPTY `read_buf`, so a single `*` — the
+  first character of every RESP array — kept it non-empty permanently, made the
+  connection unparkable, and dropped it into the handler's UNREGISTERED plain
+  read. Nothing on that path carries a sweep handle, so the connection held its
+  full stage-2 working set for as long as the attacker left the socket open,
+  with no authentication and no further traffic. One byte and one socket per
+  connection; at 1M connections roughly 10-15 GB that no amount of idle time
+  reclaims. Unparsed input no longer blocks a park: the remainder is carried in
+  `read_buf_remainder` and re-parsed on resume, exactly as a migrating
+  connection already did. Deliberately NOT capped — `read_buf` is already
+  bounded by `client_query_buffer_limit`, and a second threshold would only
+  move the attack past it (send 513 bytes instead of 1). A pending `write_buf`
+  still blocks the park, because a reply the client is owed is carried nowhere
+  and would be silently dropped.
+
 - **Reply writes were unbounded — a client that stops reading held the whole
   reply forever (c10k C1).** `write_all` on a socket whose receive window is
   closed never returns. A client could pipeline a large response and then

src/server/conn/handler_monoio/mod.rs

@@ -865,8 +865,18 @@ pub(crate) async fn handle_connection_sharded_monoio<
             let parkable = park.can_park
                 && S::SUPPORTS_TASK_PARK
                 && idle_park::park_after_ms() > 0
-                && read_buf.is_empty()
-                && write_buf.is_empty()
+                // c10k D2: unparsed input must NOT block the park. Requiring an
+                // EMPTY read_buf here let one byte (`*`) pin a connection into
+                // the unregistered plain read below, out of the sweep's reach
+                // forever. The remainder rides along in `read_buf_remainder`
+                // and is re-parsed on resume; its size is already bounded by
+                // `client_query_buffer_limit` upstream, so no second cap
+                // belongs here (one would just be an escape hatch to size
+                // past — see `park_policy`).
+                && crate::server::conn::park_policy::remainder_allows_park(
+                    read_buf.len(),
+                    write_buf.len(),
+                )
                 && !conn.in_multi
                 && conn.command_queue.is_empty()
                 && conn.active_cross_txn.is_none()
@@ -895,8 +905,11 @@ pub(crate) async fn handle_connection_sharded_monoio<
                     Err(ref e) if !idle_park::is_sweep_cancel(e) => break,
                     Err(_) => {
                         // Cancelled by the stage-2 sweep: exit the task.
-                        // read_buf is empty (predicate), so no partial frame
-                        // is at risk.
+                        // read_buf holds at most MAX_PARKED_REMAINDER bytes
+                        // (predicate) and `read_buf.split()` below carries
+                        // them into the parked state, so a partial frame
+                        // resumes exactly where it left off rather than being
+                        // dropped or pinning the connection awake (D2).
                         let state = Box::new(MigratedConnectionState {
                             selected_db: conn.selected_db,
                             authenticated: conn.authenticated,

src/server/conn/mod.rs

@@ -7,6 +7,7 @@ pub mod handler_monoio;
 pub mod handler_sharded;
 #[cfg(feature = "runtime-tokio")]
 pub mod handler_single;
+pub mod park_policy;
 pub mod shared;
 #[cfg(all(test, feature = "runtime-monoio"))]
 mod tests;

src/server/conn/park_policy.rs

@@ -0,0 +1,108 @@
+//! When may an idle connection be parked? (c10k D2)
+//!
+//! The park itself is monoio-only (`handler_monoio::idle_park`), but the
+//! *policy* is pure logic with no runtime dependency, and it lives here
+//! deliberately so it is testable under both feature sets. Every CI test job
+//! builds `--no-default-features --features runtime-tokio`, so a test placed
+//! inside the `#[cfg(feature = "runtime-monoio")]` handler tree never runs in
+//! CI — a security predicate guarded by invisible tests is guarded by nothing.
+
+/// Does the connection's buffer state permit a stage-2 park?
+///
+/// This used to be `read_buf.is_empty() && write_buf.is_empty()` inline in the
+/// monoio handler, and the `read_buf` half was an unauthenticated DoS: one byte
+/// — `*`, the first character of every RESP array — kept `read_buf` non-empty
+/// permanently, which made the connection unparkable, which dropped it into the
+/// handler's UNREGISTERED plain read. Nothing on that path carries a sweep
+/// handle, so the connection became invisible to the idle sweep for as long as
+/// the attacker left it open, holding its full stage-2 working set. Cost to the
+/// attacker: one byte and one socket. At 1M connections, 10-15 GB that no
+/// amount of idle time reclaims.
+///
+/// # Why there is no cap on the remainder
+///
+/// The obvious fix is "park only if the remainder is small" — and it is wrong.
+/// Any threshold simply moves the attack: with a 512-byte cap, an attacker
+/// sends 513 bytes of an incomplete frame instead of 1 and the connection is
+/// invisible again, 513× more expensive and still free. A partial fix here
+/// reads as a fix while leaving the vector open.
+///
+/// An unbounded remainder is safe because `read_buf` is *already* bounded, one
+/// layer up: `client_query_buffer_limit` (and the smaller pre-auth ceiling) is
+/// enforced after every read arm and ahead of both parse paths, precisely
+/// because an incomplete frame is what makes `read_buf` grow. So the remainder
+/// a park can carry is capped by an existing, configurable limit — adding a
+/// second, arbitrary one would only reopen the hole between them.
+///
+/// Carrying the remainder is also strictly better than the alternative: it
+/// moves those bytes from a ~10-15 KB live handler into the ~3.3 KB parked
+/// state, where the ordinary sweep can reach them.
+///
+/// # Why `write_buf` is still strict
+///
+/// The two buffers are not symmetric. `read_buf` is unparsed *input*: it is
+/// carried across the park in `MigratedConnectionState::read_buf_remainder`
+/// and re-parsed on resume, so the partial frame continues exactly where it
+/// left off. `write_buf` is a *reply the client is owed* and is carried
+/// nowhere — parking with it non-empty would silently drop bytes the client is
+/// waiting for, so it stays a strict emptiness check.
+pub fn remainder_allows_park(read_buf_len: usize, write_buf_len: usize) -> bool {
+    // `read_buf_len` is deliberately unused: see "Why there is no cap on the
+    // remainder" above. It stays in the signature because the question "does
+    // unparsed input block a park?" is exactly what this predicate answers,
+    // and the answer being "no" is the fix.
+    let _ = read_buf_len;
+    write_buf_len == 0
+}
+
+#[cfg(test)]
+mod park_policy_tests {
+    use super::*;
+
+    /// The D2 attack itself: one byte must not pin a connection awake.
+    #[test]
+    fn one_byte_partial_frame_still_parks() {
+        assert!(
+            remainder_allows_park(1, 0),
+            "a 1-byte partial frame is the D2 attack: it MUST NOT pin the \
+             connection out of the sweep's reach"
+        );
+    }
+
+    /// The ordinary case must keep parking.
+    #[test]
+    fn empty_read_buf_still_parks() {
+        assert!(remainder_allows_park(0, 0));
+    }
+
+    /// A cap on the remainder would only move the attack past the cap. This
+    /// pins the decision: no size of unparsed input may block a park, because
+    /// `client_query_buffer_limit` already bounds `read_buf` upstream.
+    ///
+    /// If someone later reintroduces a threshold here, this fails — which is
+    /// the point.
+    #[test]
+    fn no_remainder_size_blocks_a_park() {
+        for len in [512, 513, 8192, 64 * 1024, 512 * 1024, usize::MAX] {
+            assert!(
+                remainder_allows_park(len, 0),
+                "a {len}-byte remainder must still park: any cap here is an \
+                 escape hatch an attacker just sizes past"
+            );
+        }
+    }
+
+    /// A pending reply must NEVER park: unlike unparsed input, `write_buf` is
+    /// not carried in `MigratedConnectionState`, so parking with it non-empty
+    /// would silently drop bytes the client is owed.
+    #[test]
+    fn pending_reply_never_parks() {
+        assert!(
+            !remainder_allows_park(0, 1),
+            "an unwritten reply must keep the connection awake — it is not \
+             carried across the park"
+        );
+        assert!(!remainder_allows_park(1, 1));
+        assert!(!remainder_allows_park(usize::MAX, 1));
+    }
+}

tests/parked_idle_parity.rs

@@ -344,3 +344,55 @@ fn active_sibling_undisturbed_by_parking() {
     let _ = child.kill();
     let _ = child.wait();
 }
+
+/// c10k D2: a connection holding a PARTIAL frame must park, and the fragment
+/// must survive the park intact.
+///
+/// Before the fix the stage-2 park predicate required an empty `read_buf`, so
+/// this connection could never park at all — one byte pinned it into the
+/// handler's unregistered plain read, invisible to the idle sweep, holding its
+/// full working set for as long as the attacker cared to wait. The wire
+/// behaviour was identical either way, which is exactly why this needs an
+/// explicit test: the bug was invisible from the client's side.
+///
+/// What this proves is the half that *is* observable — correctness of the
+/// carry. The remainder rides in `MigratedConnectionState::read_buf_remainder`
+/// and is re-parsed on resume, so a command split across a park must complete
+/// normally. If the fragment were dropped or double-counted, the reply would
+/// never arrive or would be garbage.
+#[test]
+fn partial_frame_survives_a_park() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let (mut child, port) = common::spawn_listening(|p| spawn_moon(dir.path(), p));
+
+    let mut conn = TcpStream::connect(("127.0.0.1", port)).expect("connect");
+    conn.set_nodelay(true).ok();
+    ping(&mut conn);
+
+    // Send ONE byte of a RESP array and stop. This is the D2 attack byte: it
+    // parses to nothing, so it sits in read_buf indefinitely.
+    conn.write_all(b"*").expect("write partial frame");
+    conn.flush().ok();
+
+    // Long enough for downshift + park threshold + sweep. The connection must
+    // park here rather than pinning its working set awake.
+    std::thread::sleep(PARK_WAIT);
+
+    // Now complete the frame: `*2\r\n$3\r\nGET\r\n$1\r\nk\r\n` minus the `*`
+    // already sent. The carried fragment must join these bytes seamlessly.
+    conn.write_all(b"2\r\n$3\r\nGET\r\n$1\r\nk\r\n")
+        .expect("write frame tail");
+    let r = read_exact_deadline(&mut conn, 5);
+    assert_eq!(
+        &r, b"$-1\r\n",
+        "a command split across a park must complete: the carried fragment \
+         and the tail must reassemble into exactly one GET"
+    );
+
+    // And the connection must still be fully usable afterwards — the carry
+    // must not have left stray bytes in the buffer.
+    ping(&mut conn);
+
+    let _ = child.kill();
+    let _ = child.wait();
+}