fix(conn): c10k cluster C — bound reply writes, query buffers, and maxclients rejections (+ allocator observability)

.config/nextest.toml

@@ -28,3 +28,13 @@ slow-timeout = { period = "60s", terminate-after = 8 }
 filter = 'test(/recall/)'
 retries = 0
 slow-timeout = { period = "120s", terminate-after = 5 }
+
+# The idle-reclaim probe is a MEASUREMENT, not just a pass/fail gate: on Apple
+# platforms it deliberately records whether the host reclaimed or retained
+# rather than failing, because jemalloc's background_thread is compiled out
+# there and no in-process remedy exists (see src/memory_ctl.rs). Captured
+# stdout on a pass would discard the only number the test exists to produce,
+# so surface it either way.
+[[profile.ci.overrides]]
+filter = 'test(/freed_memory_is_returned_to_the_os_when_idle/)'
+success-output = "final"

CHANGELOG.md

@@ -6,7 +6,96 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **`INFO memory` can now explain a `used_memory`-vs-RSS gap.** Build with
+  `--features jemalloc-stats` and `INFO memory` reports Redis's `allocator_*`
+  fields: `allocator_allocated`, `allocator_active`, `allocator_resident`,
+  `allocator_retained`, `allocator_frag_bytes`, `allocator_frag_ratio`, and
+  `allocator_unreturned_bytes`. `active - allocated` is fragmentation,
+  `resident - active` is dirty pages jemalloc holds but has not returned, and
+  anything the OS charges beyond `resident` belongs to something other than the
+  allocator. Motivated by a real instance reporting `used_memory` 2.43 GB while
+  the OS charged it 7.3 GB (7.2 GB of that swapped) with no way to tell which.
+  OFF by default — jemalloc's stats add bookkeeping to every allocation. Fields
+  are absent rather than zero-filled when not built in, because a zero would
+  read as "no fragmentation", which is worse than "not measured".
+
+### Known issues
+- **An idle moon may not return freed memory to the OS on Apple platforms, and
+  there is no in-process fix.** jemalloc's `background_thread` is compiled out
+  when `abi == macho` (`JEMALLOC_BACKGROUND_THREAD` is only defined otherwise),
+  so the `background_thread:true` moon bakes into its malloc conf is a silent
+  no-op there and decay runs only as a side effect of allocator activity. The
+  same 384 MiB churn-and-free reclaimed to a 3.7 MiB physical footprint on one
+  Apple Silicon machine and retained all 386 MiB indefinitely on a GitHub macOS
+  runner, so the behaviour varies by machine. A `--memory-decay-interval-ms`
+  timer calling `mallctl("arena.4096.decay")` — the same call jemalloc's own
+  background thread makes — was implemented and then **removed after it failed
+  to reclaim anything on the runner that reproduces the retention**, across 30
+  seconds of driving it and three independent retries, while the ctl itself
+  returned success. Production targets Linux, where the background thread is
+  compiled in and the retention has not been observed; on macOS, build with
+  `--features jemalloc-stats` and watch `allocator_unreturned_bytes` to tell
+  whether a given host is affected.
+
 ### Security
+- **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
+  simply stop reading, parking the handler inside that write while it held the
+  ENTIRE serialized reply — the monoio handler coalesces a whole batch into one
+  `Bytes` before its single write syscall, so a deep pipeline is hundreds of MB
+  — plus its `maxclients` slot, for as long as the attacker cared to wait. N
+  such clients is an OOM that costs the attacker nothing: no reads, no CPU,
+  just a TCP window it refuses to open. New `--client-write-timeout-ms`
+  (default `60000`, `0` = the previous wait-forever behaviour) bounds every
+  reply-carrying write in all three handlers — monoio top-level, sharded, and
+  the tokio single-shard `Framed` path — closing the connection and dropping
+  the reply when a write makes no progress for that long. 60s is far beyond any
+  healthy client's stall and replication does not use this path (PSYNC hijacks
+  the connection before it), but operators streaming very large replies over
+  very slow links should raise it: the budget covers the whole write call, not
+  each byte. Verified at 1 and 4 shards on both runtimes, with the `0` case as
+  a differential control proving the mechanism rather than the harness
+  (`tests/write_timeout.rs`). **Unverified on Windows**: the tests need the
+  server's write to actually block, and two attempts to force that under
+  Winsock failed (a 25 MB reply was absorbed by send/receive autotuning, and
+  clamping the victim's `SO_RCVBUF` to 8 KiB did not change it), so they are
+  skipped there rather than weakened until they pass. The code path is shared
+  and compiles on Windows, but nothing proves the timeout fires; Windows is not
+  a target platform (Linux and macOS are).
+- **`CLIENT LIST` reported `obl=0 oll=0 omem=0` unconditionally (c10k C1).**
+  The held-output counters were hardcoded, so an operator watching a
+  slow-client output-buffer OOM in progress saw every client reporting zero
+  bytes held — the attack above left no trace anywhere. `obl`/`omem` now report
+  the reply bytes a connection has in an in-flight write, and `tot-net-out`
+  counts reply bytes that actually reached the peer. `oll` stays 0: moon has
+  one contiguous output buffer, not Redis's static-buffer + reply-list pair, so
+  there is no list whose length it could report. Wired in the two handlers that
+  own a client-registry entry (monoio top-level and sharded); the legacy tokio
+  `handler_single` path bounds its writes but does not register, so it has
+  nothing to report through.
+- **No size ceiling on a reply (c10k C1).** Adds
+  `--client-output-buffer-limit-normal`, Redis's
+  `client-output-buffer-limit normal <hard>`, and ships it at **256 MiB rather
+  than Redis's unlimited default** — that default is exactly why an unread
+  socket can OOM Redis too. A reply exceeding the cap is refused and the
+  connection closed instead of buffering it. Consequence, intended and worth
+  knowing: the cap covers the whole serialized reply, so a single value larger
+  than the cap is undeliverable, not merely a long pipeline. No pub/sub-class
+  knob ships: moon's only subscriber write is already hard-capped at 64 KiB by
+  `MAX_COALESCE_BYTES`, so such a knob could never fire, and the real pub/sub
+  backpressure is the 4096-slot bounded channel (`CONN_CHANNEL_CAPACITY`) —
+  which bounds message COUNT, not bytes, and is a genuine follow-up.
+- **The write watchdog no longer arms on the hot path.** A timer per batch
+  flush would land once per command at pipeline depth 1, the path this project
+  spent a milestone winning against Redis. A reply that fits in the socket
+  buffer cannot block, so only writes ≥ 256 KiB arm the watchdog now
+  (`util::arm_write_timeout`). Residual, stated rather than hidden: a small
+  write to a genuinely wedged socket is still unbounded — it holds at most
+  256 KiB instead of hundreds of MB, but keeps its `maxclients` slot. It is now
+  visible (`omem` > 0) and killable (`CLIENT KILL` shuts the fd down), which it
+  was not before.
 - **Privileged commands ran BEFORE the ACL permission check (c10k B1).**
   Both connection handlers intercepted `EVAL`/`EVALSHA`/`SCRIPT`, `ACL`, and
   `CLUSTER` above the ACL gate, and each intercept `continue`s on a match, so

Cargo.toml

@@ -69,6 +69,10 @@ http-body-util = "0.1"
 
 tikv-jemallocator = { version = "0.6", optional = true }
 tikv-jemalloc-ctl = { version = "0.6", optional = true }
+# Direct dep for the raw `mallctl` symbol: `arena.<i>.decay` is a
+# NEITHER_READ_NOR_WRITE ctl (every pointer NULL, newlen 0), which the
+# safe tikv-jemalloc-ctl wrappers cannot express. See src/memory_decay.rs.
+tikv-jemalloc-sys = { version = "0.6", optional = true }
 monoio = { version = "0.2", optional = true, features = ["sync", "bytes"] }
 cudarc = { version = "0.19", optional = true, default-features = false, features = ["cuda-version-from-build-system"] }
 slotmap = "1"
@@ -99,6 +103,14 @@ levenshtein_automata = { version = "0.2", features = ["fst_automaton"], optional
 #   cargo build --no-default-features --features runtime-monoio,jemalloc  # force Monoio
 default = ["runtime-monoio", "jemalloc", "graph", "text-index"]
 jemalloc = ["dep:tikv-jemallocator", "dep:tikv-jemalloc-ctl"]
+# Compile jemalloc with --enable-stats and expose its counters through
+# `INFO memory` (Redis-compatible `allocator_*` field names). OFF by default:
+# jemalloc's stats add bookkeeping to every allocation, which this project
+# will not pay on the hot path unconditionally. Build with it when you need
+# to answer "why is RSS N times used_memory" — see src/memory_ctl.rs.
+# Only `jemalloc-stats` needs the -sys crate: no Rust code calls it directly,
+# it is here solely to turn on jemalloc's `stats` build option.
+jemalloc-stats = ["jemalloc", "dep:tikv-jemalloc-sys", "tikv-jemalloc-ctl/stats", "tikv-jemalloc-sys/stats"]
 # Opt-in alternate allocator. Mutually exclusive with `jemalloc`.
 # Usage:
 #   cargo build --no-default-features --features runtime-monoio,mimalloc-alt,graph,text-index

src/client_registry.rs

@@ -113,6 +113,18 @@ pub struct ClientLiveState {
     /// 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,
+    /// Bytes of reply currently held in an in-flight write (c10k C1).
+    ///
+    /// This is the memory a client that stops reading is pinning: the handler
+    /// serializes a whole batch into one buffer before its single write
+    /// syscall, so a deep pipeline against a closed receive window parks
+    /// hundreds of MB here. `CLIENT LIST` reported `obl=0 oll=0 omem=0`
+    /// unconditionally, which made the attack invisible while it happened.
+    /// Reported as `obl`/`omem` — moon has one contiguous output buffer, not
+    /// Redis's static-buffer + reply-list pair, so `oll` stays 0.
+    pub pending_out_bytes: AtomicU64,
+    /// Cumulative reply bytes successfully written — `tot-net-out`.
+    pub tot_net_out: AtomicU64,
 }
 
 impl ClientLiveState {
@@ -132,6 +144,25 @@ impl ClientLiveState {
         self.flags.store(flags.to_bits(), Ordering::Relaxed);
     }
 
+    /// Mark a reply write as in flight. One relaxed store, on a path that is
+    /// about to make a syscall — the cost is not measurable there.
+    #[inline]
+    pub fn begin_write(&self, bytes: usize) {
+        self.pending_out_bytes
+            .store(bytes as u64, Ordering::Relaxed);
+    }
+
+    /// Clear the in-flight marker. `completed` distinguishes a delivered reply
+    /// from one abandoned by `--client-write-timeout-ms` or a socket error, so
+    /// `tot-net-out` counts only bytes that actually reached the peer.
+    #[inline]
+    pub fn end_write(&self, bytes: usize, completed: bool) {
+        self.pending_out_bytes.store(0, Ordering::Relaxed);
+        if completed {
+            self.tot_net_out.fetch_add(bytes as u64, Ordering::Relaxed);
+        }
+    }
+
     /// Lock-free CLIENT KILL check for the connection's own loop.
     #[inline]
     pub fn is_killed(&self) -> bool {
@@ -204,6 +235,8 @@ pub fn register(
         connected_at_epoch_ms: crate::storage::entry::current_time_ms(),
         db: AtomicUsize::new(0),
         last_cmd_ms: AtomicU64::new(0),
+        pending_out_bytes: AtomicU64::new(0),
+        tot_net_out: AtomicU64::new(0),
         flags: AtomicU8::new(ClientFlags::default().to_bits()),
         kill_flag: AtomicBool::new(false),
         kill_fd,
@@ -449,13 +482,21 @@ fn format_client_line(buf: &mut String, entry: &ClientEntry, now: Instant) {
     let name = entry.name.as_deref().unwrap_or("");
     let flags = ClientFlags::from_bits(live.flags.load(Ordering::Relaxed)).to_flag_str();
     let db = live.db.load(Ordering::Relaxed);
+    // c10k C1: `obl`/`omem` report the reply bytes this connection is holding
+    // in an in-flight write. A client that stops reading pins that memory for
+    // as long as `--client-write-timeout-ms` allows, and with these hardcoded
+    // to 0 there was no way to see it happening. `oll` stays 0: moon has one
+    // contiguous output buffer, not Redis's static-buffer + reply-list pair,
+    // so there is no list whose length it could report.
+    let omem = live.pending_out_bytes.load(Ordering::Relaxed);
+    let tot_net_out = live.tot_net_out.load(Ordering::Relaxed);
     let _ = writeln!(
         buf,
         "id={} addr={} laddr=127.0.0.1:0 fd=0 name={} age={} idle={} flags={} db={} \
          sub=0 psub=0 ssub=0 multi=-1 watch=0 qbuf=0 qbuf-free=0 argv-mem=0 multi-mem=0 \
-         tot-net-in=0 tot-net-out=0 rbs=1024 rbp=0 obl=0 oll=0 omem=0 tot-mem=0 events=r \
+         tot-net-in=0 tot-net-out={} rbs=1024 rbp=0 obl={} oll=0 omem={} tot-mem={} events=r \
          cmd=NULL user={} redir=-1 resp=2 lib-name= lib-ver=",
-        entry.id, entry.addr, name, age, idle, flags, db, entry.user,
+        entry.id, entry.addr, name, age, idle, flags, db, tot_net_out, omem, omem, omem, entry.user,
     );
 }
 

src/command/connection.rs

@@ -231,6 +231,41 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
         allocator_overhead_bytes = allocator_overhead_bytes,
         pagecache_bytes = pagecache_bytes,
     );
+
+    // Allocator counters, Redis's `allocator_*` field names so existing
+    // dashboards and exporters read them without translation.
+    //
+    // These are the fields that make a used_memory-vs-RSS gap diagnosable
+    // rather than a guess: `allocator_frag_bytes` is space lost to size-class
+    // rounding, `allocator_unreturned_bytes` is dirty pages jemalloc is
+    // holding instead of giving back, and whatever the OS charges beyond
+    // `allocator_resident` belongs to something other than the allocator
+    // (mmap'd segments, thread stacks, the binary image).
+    //
+    // Only present on `--features jemalloc-stats`; jemalloc's stats cost
+    // bookkeeping on every allocation, so the default build does not pay it.
+    // Absent rather than zero-filled: a zero here would read as "no
+    // fragmentation", which is a worse answer than "not measured".
+    #[cfg(feature = "jemalloc-stats")]
+    if let Some(st) = crate::memory_ctl::jemalloc_stats() {
+        let _ = write!(
+            sections,
+            "allocator_allocated:{}\r\n\
+             allocator_active:{}\r\n\
+             allocator_resident:{}\r\n\
+             allocator_retained:{}\r\n\
+             allocator_frag_bytes:{}\r\n\
+             allocator_frag_ratio:{:.2}\r\n\
+             allocator_unreturned_bytes:{}\r\n",
+            st.allocated,
+            st.active,
+            st.resident,
+            st.retained,
+            st.frag_bytes(),
+            st.frag_ratio(),
+            st.unreturned_bytes(),
+        );
+    }
     sections.push_str("\r\n");
 
     sections.push_str("# Persistence\r\n");