fix(hardening): production-hardening audit batches A–D (28-finding sweep)

CHANGELOG.md

@@ -6,6 +6,211 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed — Conn-plane: monoio central listener joins the SO_REUSEPORT group (PR #250)
+
+- Under the monoio runtime, each shard binds `bind:port` via `SO_REUSEPORT`, but
+  the central listener bound the same port with a plain
+  `monoio::net::TcpListener::bind` — unlike the tokio sibling, which uses
+  `create_reuseport_socket`. Linux requires every socket sharing a REUSEPORT port
+  to set the option, so depending on startup scheduling the central bind either
+  lost the race (EADDRINUSE propagated out of `run_sharded`, silently killing the
+  TLS listener + `conn_rx` fallback) or won it (forcing every shard's REUSEPORT
+  bind to fail and collapsing all traffic onto the single central accept loop) —
+  both silent and nondeterministic run-to-run. The monoio central listener now
+  binds via `create_reuseport_socket` + `from_std` like the tokio path (plain-bind
+  fallback on non-unix / unparseable addr). (audit finding 16)
+
+### Fixed — Durability: checkpoint Finalize backs off on repeated failure instead of flooding the WAL (PR #250)
+
+- When a checkpoint's Finalize step failed (`wal.wait_durable`, `manifest.commit`,
+  `graph_save`, or the control-file write), the state machine stayed `Finalizing`
+  and the next 1ms persistence tick re-ran finalize from step 1 — re-appending a
+  WAL Checkpoint record every single millisecond with no backoff. Under a
+  sustained failure (slow/degraded disk) this floods the WAL, and since
+  `last_checkpoint_lsn` never advances, `recycle_segments_before` never fires —
+  WAL disk usage grows fastest exactly during the disk-pressure incident it
+  should be backing off from. `CheckpointManager` now arms an exponential backoff
+  (50ms→5s cap) on each failed finalize; the tick checks `finalize_ready(now)`
+  before doing any finalize I/O, so retries are bounded instead of per-tick, and
+  a successful `complete()` clears it. (audit finding 13)
+
+### Fixed — Durability: legacy AOF replay stops at mid-stream corruption instead of resyncing (PR #250)
+
+- The default startup path `replay_aof` (single-file `appendonly.aof`, unframed
+  RESP with no per-record length/CRC) treated ANY mid-stream parse error by
+  discarding one byte, scanning forward to the next `*`, and resuming dispatch —
+  but a `*` appears freely inside binary value blobs, so the resync could land
+  mid-value and execute misaligned garbage as live SET/DEL/etc. against the
+  recovering keyspace (silent data corruption, WARN-only). The per-shard
+  `shard_replay` sibling already rejects exactly this. `replay_aof` now STOPS at
+  the first genuine mid-stream corruption (clean tail-truncation is still handled
+  gracefully), keeping only the valid prefix already applied and never
+  dispatching past the corruption point; it logs a loud error with the byte
+  offset and points at `redis-check-aof`. Operators who want the old best-effort
+  behavior can opt in with `MOON_AOF_BEST_EFFORT_RESYNC=1`. (audit finding 12)
+
+### Fixed — Xshard: bound cross-shard reply wait so a wedged shard can't hang a client (PR #250)
+
+- Every cross-shard leg (MGET/MSET/DEL/UNLINK/EXISTS/BITOP/COPY/MSETNX remote
+  legs, MULTI-on-owner, and the flush/scatter-gather loops) pushed a message via
+  the backoff-retried `spsc_send` and then did `reply_rx.recv().await` with **no
+  timeout**. `spsc_send`'s retry budget only bounds getting the message INTO the
+  target ring buffer; if the push succeeds but the target shard then stalls while
+  executing (wedged-disk fsync during snapshot/AOF, uninterruptible D-state I/O,
+  or a dead shard), the awaiting connection parked forever with no response and
+  no way to cancel. Added a `recv_reply_bounded` helper (races the receiver
+  against a runtime-agnostic 30s `XSHARD_REPLY_TIMEOUT` via `race2`) and applied
+  it to all 11 reply-await sites in the coordinator — a genuinely wedged shard
+  now surfaces the existing cross-shard-reply error instead of hanging. (audit finding 11)
+
+### Fixed — Hardening: bound TLS handshake + cluster-bus body reads against slow-loris (PR #250)
+
+- **TLS handshake has no timeout (#17):** after `try_accept_connection` consumed
+  a `maxclients` slot, `acceptor.accept(tcp_stream).await` ran with no deadline
+  on both the monoio and tokio TLS paths. A peer that completed the TCP handshake
+  on the TLS port and then sent no (or a partial) ClientHello pinned the task,
+  its fd, and its `maxclients` slot indefinitely at zero CPU cost — a classic
+  slow-loris that could exhaust the connection limit. Wrapped both accepts in a
+  10s `TLS_HANDSHAKE_TIMEOUT`, releasing the slot via `record_connection_closed()`
+  on expiry.
+- **Cluster-bus gossip body read has no timeout (#10):** `handle_cluster_peer`
+  guarded only the 4-byte length-prefix read with a shutdown-cancel select; once
+  a valid length was read, the body `read_exact` had no timeout and no
+  shutdown arm. A client could send a valid length header and never send the
+  body, blocking the spawned task forever (immune to graceful shutdown, socket
+  never reclaimed). Both runtimes now read the body under a
+  `GOSSIP_BODY_READ_TIMEOUT` (10s) + shutdown-cancel select.
+
+### Fixed — Txn: MULTI locality analysis honors SORT/GEORADIUS STORE destination (PR #250)
+
+- In a multi-shard deployment, `MULTI; SORT src STORE dst; EXEC` (or
+  `GEORADIUS ... STORE/STOREDIST dst`) where `src` and `dst` hash to different
+  shards was misclassified as single-shard and routed entirely to `src`'s
+  shard — writing `dst` into the WRONG shard's dataset, invisible to a later
+  normally-routed `GET dst`. The STORE/STOREDIST destination is positional
+  (the arg after the token) and so was not covered by the fixed command-metadata
+  key specs `analyze_txn_locality` consulted. It now detects the STORE clause
+  and forces a CrossShard classification, which the caller rejects with
+  CROSSSLOT instead of silently misrouting. (audit finding 15)
+
+### Fixed — Vector: DEL tombstones secondary fields + FT.INFO num_docs accuracy (PR #250)
+
+- **Multi-vector-field deletion resurrection (#20):** `DEL`/`UNLINK`/`HDEL` on a
+  document only tombstoned the *default* vector field. Secondary VECTOR fields
+  live in a separate `field_segments` map that `tombstone_key_in_index` never
+  iterated, so a subsequent `FT.SEARCH idx '@field2:[...]'` still returned the
+  deleted document (under a synthetic `vec:<id>` key, since the shared
+  key-hash→key map was cleared). Now every field's segments are tombstoned.
+- **FT.INFO `num_docs` over/under-count (#28):** the mutable segment's `len()`
+  counts tombstoned entries in place until compaction, inflating `num_docs` by
+  up to `compact_threshold` under DEL/HSET churn — switched to a new `live_len()`
+  that excludes tombstones. Separately, DiskANN cold-tier segments (experimental,
+  gated by `MOON_VEC_COLD_TIER`) were never summed at all even though cold docs
+  ARE returned by FT.SEARCH; added a `snap.cold` counting loop. (audit findings 20, 28)
+
+### Fixed — Vector: GraphUnion merge recall gate no longer lets a total collapse through (PR #250)
+
+- The background/manual GraphUnion merge recall gate read
+  `recall < tolerance && recall > 0.0`, so a merge whose verified recall was
+  **exactly 0.0** (total collapse) slipped past the guard and committed. Since
+  `verify_merge_recall` returns 1.0 (not 0.0) for the too-few-vectors cases,
+  0.0 only ever means a real measured collapse — the `&& recall > 0.0` clause
+  was removed so such a merge now aborts. (audit finding 21)
+
+### Fixed — Hardening: APPEND 512MB cap + reject FT.* inside MULTI on prod handlers (PR #250)
+
+- **APPEND** now enforces the 512MB max-string-size limit (matching
+  SETRANGE/SETBIT); repeated APPENDs previously grew a value past the documented
+  limit unbounded. (audit finding 22)
+- **FT.\*** commands are now explicitly rejected inside MULTI/EXEC on the
+  production sharded + monoio handlers (they aren't wired through the txn
+  execution path), matching the existing handler_single guard — a clear
+  "not supported inside MULTI/EXEC" error instead of an incidental one.
+  (audit finding 25)
+
+### Changed — Vector storage: fence experimental DiskANN cold tier + drop dead vector WAL (PR #250)
+
+- The DiskANN cold tier (WARM→COLD segment demotion + on-disk Vamana beam
+  search) was running by default (`--disk-offload` defaults on, `--segment-cold-after`
+  defaults 86400) despite being incomplete — **no cold-segment deletion**
+  (DEL/HDEL never removes cold docs), blocking per-hop uring I/O holding the
+  segment mutex, `FT.INFO num_docs` under-counting cold segments, unfinished
+  restart reload of PQ codebooks, and a library-code `panic!` in
+  `DiskAnnSegment::new`. It is now **fenced behind the experimental
+  `MOON_VEC_COLD_TIER=1` env gate** (default off); the WARM→COLD transition is
+  a no-op otherwise. Warm-tier mmap + LRU eviction (`--vec-warm-mmap-budget`)
+  continues to handle out-of-RAM indexes. (audit findings 19, 24, 28)
+- `DiskAnnSegment::new` now returns `io::Result` instead of panicking on a
+  file-open error, so a cold transition can never abort the shard.
+- Removed the dead vector WAL module (`vector/persistence/wal_record.rs`): it
+  was never on the live recovery path (superseded by manifest + segment +
+  keymap + dedup rescan) and carried a documented double-apply footgun.
+
+### Fixed — Cluster: gossip PING task leak + non-blocking accept send (PR #250)
+
+- The gossip PING ticker (both tokio + monoio) spawned an unbounded
+  connect+write+read task every 100ms with **no timeout**; against a dead or
+  partitioned peer the tasks piled up forever (task/FD/memory leak), and the
+  "random peer" it documented was always the first non-self node. Now a single
+  in-flight guard bounds outstanding probes to one, a probe timeout
+  (`node_timeout/2`, ≥100ms) cancels a hung connect/read, and targets rotate
+  across peers. (audit finding 7)
+- The monoio per-shard accept task used the **blocking** `flume::Sender::send()`
+  on a bounded channel; because the accept task shares the shard's single
+  monoio thread with the event-loop consumer, a full channel deadlocked the
+  loop that drains it under connection-storm churn. Fixed by `send_async().await`
+  (cooperative yield, keeps backpressure). (audit finding 8, Batch B)
+
+### Fixed — Security: bound client-controlled allocation counts (DoS class, PR #250)
+
+- Six wire-reachable command paths fed a client-supplied count straight into
+  `Vec::with_capacity` / `Vec::resize` with no upper bound. On a real host the
+  oversized request fails allocation and Rust's `handle_alloc_error` calls
+  `abort()` — uncatchable, crashing every shard and connection. A single
+  unauthenticated request was a full-process DoS. Fixed by clamping each count
+  before allocating: **FT.SEARCH HIGHLIGHT/SUMMARIZE FIELDS** (bound by
+  remaining tokens), **HSCAN COUNT** (`count.min(total)`),
+  **SRANDMEMBER / HRANDFIELD / ZRANDMEMBER** negative COUNT (absolute 2²⁰ cap
+  with a loud `ERR COUNT is out of range` beyond it — the first-cut relative
+  `len()*10` cap silently truncated legitimate requests on small collections,
+  breaking Redis's exact-|COUNT|-with-duplicates contract; caught in review and
+  fixed at all six sites including the pre-existing HRANDFIELD/ZRANDMEMBER
+  ones), **BITFIELD SET/INCRBY** offset (reject past the 512MB limit, matching
+  SETBIT; GET exempt), and **ZMPOP COUNT** (`pop_count.min(card)` — this last
+  site was missed by the audit finders and caught by the allocation sweep).
+  Each fix has a red/green test. From the production-hardening audit (Batch A).
+
+### Fixed — Review follow-ups on the hardening sweep (PR #250)
+
+- **Gossip PONG fragmentation (monoio):** the failure-detector's PING probe read
+  the 4-byte length and the PONG body with single `read()` calls, so a valid
+  frame arriving fragmented was silently dropped — leaving `pong_recv_ms` stale
+  and pushing a healthy peer toward PFAIL. Both reads now use the bus listener's
+  exact-read loop, and both runtimes' probes cap the peer-supplied PONG length
+  at the bus's 64KB frame bound before sizing the buffer (the tokio probe
+  previously allocated an unchecked `vec![0u8; pong_len]`).
+- **AOF corruption offsets are file-absolute:** replay's truncation/corruption
+  logs reported offsets relative to the RESP section; with an RDB preamble the
+  operator-facing repair guidance (redis-check-aof, truncation point) pointed at
+  the wrong file location. Offsets now include the preamble length.
+- **Cross-shard MSET no longer OKs an unconfirmed leg:** a timed-out, closed, or
+  errored remote `MultiExecute` ack was discarded and the client still got `OK`.
+  All legs are drained (each was already dispatched, and the local leg still
+  persists), but any failed leg now surfaces as an error instead of a false ack.
+- **`recv_reply_bounded` extended to the handler plane:** the coordinator's 30s
+  bounded reply-await now also covers the 11 remaining direct cross-shard acks
+  (WS.DROP cleanup, routed MQ commands, routed graph commands, TXN.COMMIT MQ
+  materialization acks, and the txn-abort remote rollback ack, on both
+  handler_monoio and handler_sharded) — a wedged owner shard can no longer park
+  those connections forever. MQ/graph/TXN owner execution is synchronous on the
+  shard thread (no blocking-wait commands route through these paths), so the
+  bound cannot cut off a legitimate long wait.
+- **Checkpoint finalize backoff arms from the failure time:** the backoff was
+  armed with a timestamp captured before the finalize I/O; when
+  `wait_durable`/`manifest.commit`/`graph_save`/control-write took longer than
+  the backoff window, the retry deadline was already in the past and the next
+  1ms tick retried immediately — exactly the flood the backoff exists to stop.
+
 ### Added — CLI/moon.conf defaults for vector + graph tuning knobs (PR #TBD)
 
 - **`--vector-ef-runtime` / `--vector-rerank-mult` / `--vector-exact-beam`**

src/cluster/bus.rs

@@ -27,6 +27,21 @@ use crate::cluster::gossip::{
 pub type SharedVoteTx =
     Arc<parking_lot::Mutex<Option<crate::runtime::channel::MpscSender<String>>>>;
 
+/// Maximum time to wait for a gossip message BODY after its 4-byte length
+/// prefix has been read (prod-hardening #10). Without this bound, any TCP
+/// client that can reach the cluster bus port can send a valid length header
+/// and then never send the body — the spawned `handle_cluster_peer` task
+/// blocks in `read_exact` forever, is immune to graceful shutdown (the
+/// cancellation token was only checked on the length read), and never releases
+/// its socket/fd. A handful of such connections exhaust the bus listener's
+/// accept capacity. On expiry we return an error, dropping the connection.
+const GOSSIP_BODY_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+/// Maximum accepted gossip frame length (alloc guard: the 4-byte length prefix
+/// is attacker-controlled, so it must never size an allocation unchecked).
+/// Shared with the PING/PONG probe path in `gossip.rs`.
+pub(crate) const MAX_GOSSIP_FRAME_LEN: usize = 64 * 1024;
+
 /// Run the cluster bus listener loop.
 ///
 /// Spawns a new task for each incoming peer connection.
@@ -97,13 +112,25 @@ async fn handle_cluster_peer(
             _ = shutdown.cancelled() => return Ok(()),
         }
         let msg_len = u32::from_be_bytes(len_buf) as usize;
-        if msg_len > 64 * 1024 {
+        if msg_len > MAX_GOSSIP_FRAME_LEN {
             anyhow::bail!("gossip message too large: {} bytes", msg_len);
         }
 
-        // Read message body
+        // Read message body — #10: bound with a timeout + shutdown-cancel so a
+        // client that sends the length prefix but stalls the body cannot pin
+        // the task/socket forever or dodge graceful shutdown.
         let mut buf = vec![0u8; msg_len];
-        stream.read_exact(&mut buf).await?;
+        tokio::select! {
+            result = stream.read_exact(&mut buf) => { result?; }
+            _ = shutdown.cancelled() => return Ok(()),
+            _ = tokio::time::sleep(GOSSIP_BODY_READ_TIMEOUT) => {
+                anyhow::bail!(
+                    "gossip body read from {} timed out after {}s",
+                    peer_addr,
+                    GOSSIP_BODY_READ_TIMEOUT.as_secs()
+                );
+            }
+        }
 
         // Deserialize
         let msg = match deserialize_gossip(&buf) {
@@ -179,7 +206,7 @@ async fn handle_cluster_peer(
 ///
 /// Monoio has no `read_exact` — we loop on `stream.read()` accumulating bytes.
 #[cfg(feature = "runtime-monoio")]
-async fn monoio_read_exact(
+pub(crate) async fn monoio_read_exact(
     stream: &mut monoio::net::TcpStream,
     total: usize,
 ) -> std::io::Result<Vec<u8>> {
@@ -267,14 +294,26 @@ async fn handle_cluster_peer(
         };
         let len_buf: [u8; 4] = len_data[..4].try_into().unwrap();
         let msg_len = u32::from_be_bytes(len_buf) as usize;
-        if msg_len > 64 * 1024 {
+        if msg_len > MAX_GOSSIP_FRAME_LEN {
             anyhow::bail!("gossip message too large: {} bytes", msg_len);
         }
 
-        // Read message body
-        let buf = monoio_read_exact(&mut stream, msg_len)
-            .await
-            .map_err(|e| anyhow::anyhow!(e))?;
+        // Read message body — #10: bound with a timeout + shutdown-cancel so a
+        // client that sends the length prefix but stalls the body cannot pin
+        // the task/socket forever or dodge graceful shutdown.
+        let buf = monoio::select! {
+            result = monoio_read_exact(&mut stream, msg_len) => {
+                result.map_err(|e| anyhow::anyhow!(e))?
+            }
+            _ = shutdown.cancelled() => return Ok(()),
+            _ = monoio::time::sleep(GOSSIP_BODY_READ_TIMEOUT) => {
+                anyhow::bail!(
+                    "gossip body read from {} timed out after {}s",
+                    peer_addr,
+                    GOSSIP_BODY_READ_TIMEOUT.as_secs()
+                );
+            }
+        };
 
         // Deserialize
         let msg = match deserialize_gossip(&buf) {