diff --git a/CHANGELOG.md b/CHANGELOG.md
index c17727d5..c9be9e31 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed — zombie/CPU hardening wave 1 (PR #TBD)
+
+- **Busy-poll spin idle-disengages** (`--io-busy-poll-us` / vendored monoio
+  legacy driver): an idle server no longer burns the spin budget on every
+  park forever (measured ~2.4s CPU per 3s wall on an idle 4-shard server at
+  200µs). The driver tracks the last real event per thread and skips the
+  spin window after `MOON_SPIN_IDLE_DISENGAGE_US` (default 10ms; 0 = old
+  always-spin) of quiet; the next event re-arms it via the normal wake path,
+  so steady traffic — the GCE p=1 win path — never disengages.
+- **Shard-thread panics abort the whole process** instead of leaving an
+  N-1-shard server silently answering with a dead shard's keys gone; the
+  shutdown join loop logs instead of swallowing panics. New `DEBUG PANIC`
+  subcommand (Redis parity) as the crash-handling test aid.
+- **SIGTERM shutdown regression matrix**: shards × held-connections ×
+  busy-poll × 16-writer AOF write-storm (7 cases, both runtimes, 3
+  platforms). The documented SIGTERM+SO_REUSEPORT bench hang did NOT
+  reproduce at HEAD in any composition; the matrix stays as the guard and
+  the detached monoio accept-task change is deferred until a reproducer
+  exists.
+- **`wait_ready` test harness hardening**: readiness probes retry with a
+  fresh connection on mid-startup connection resets (CI flake: per-shard
+  SO_REUSEPORT listeners accept-then-reset during init).
+
 ### Added — int8 symmetric ADC for SQ8 vector search (task #13)
 
 - New per-candidate integer dot-product path for SQ8 asymmetric distance
diff --git a/src/command/server_admin.rs b/src/command/server_admin.rs
index c6166847..845d5dc7 100644
--- a/src/command/server_admin.rs
+++ b/src/command/server_admin.rs
@@ -152,6 +152,7 @@ pub fn debug(db: &mut Database, args: &[Frame]) -> Frame {
     match classify_debug(args) {
         Ok(DebugCall::Object(rest)) => debug_object(db, rest),
         Ok(DebugCall::Sleep(rest)) => debug_sleep(rest),
+        Ok(DebugCall::Panic) => debug_panic(),
         Ok(DebugCall::Help) => debug_help(),
         Err(e) => e,
     }
@@ -168,6 +169,7 @@ pub fn debug_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame {
     match classify_debug(args) {
         Ok(DebugCall::Object(rest)) => debug_object_readonly(db, rest, now_ms),
         Ok(DebugCall::Sleep(rest)) => debug_sleep(rest),
+        Ok(DebugCall::Panic) => debug_panic(),
         Ok(DebugCall::Help) => debug_help(),
         Err(e) => e,
     }
@@ -176,6 +178,7 @@ pub fn debug_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame {
 enum DebugCall<'a> {
     Object(&'a [Frame]),
     Sleep(&'a [Frame]),
+    Panic,
     Help,
 }
 
@@ -191,6 +194,8 @@ fn classify_debug(args: &[Frame]) -> Result<DebugCall<'_>, Frame> {
         Ok(DebugCall::Object(&args[1..]))
     } else if sub.eq_ignore_ascii_case(b"SLEEP") {
         Ok(DebugCall::Sleep(&args[1..]))
+    } else if sub.eq_ignore_ascii_case(b"PANIC") {
+        Ok(DebugCall::Panic)
     } else if sub.eq_ignore_ascii_case(b"HELP") {
         Ok(DebugCall::Help)
     } else {
@@ -211,6 +216,10 @@ fn debug_help() -> Frame {
         Frame::BulkString(Bytes::from_static(
             b"  Stall this shard for <seconds> (float, capped at 30).",
         )),
+        Frame::BulkString(Bytes::from_static(b"DEBUG PANIC")),
+        Frame::BulkString(Bytes::from_static(
+            b"  Panic this shard thread (crash-handling test aid, as in Redis).",
+        )),
         Frame::BulkString(Bytes::from_static(b"DEBUG HELP")),
         Frame::BulkString(Bytes::from_static(b"  Return subcommand help.")),
     ])
@@ -256,6 +265,14 @@ fn debug_object_reply(entry: &Entry) -> Frame {
     Frame::SimpleString(Bytes::from(body))
 }
 
+/// `DEBUG PANIC` — deliberately panic the executing shard thread (Redis
+/// parity: a crash-handling test aid). The process-level panic policy
+/// (fail-fast abort, installed in main) is what a client observes; the
+/// return type exists only for the signature.
+fn debug_panic() -> Frame {
+    panic!("DEBUG PANIC requested by client");
+}
+
 fn debug_sleep(args: &[Frame]) -> Frame {
     if args.len() != 1 {
         return err_wrong_args("DEBUG SLEEP");
diff --git a/src/main.rs b/src/main.rs
index 2675e876..12f8779a 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -93,6 +93,29 @@ fn main() -> anyhow::Result<()> {
         )
         .init();
 
+    // Fail-fast on shard-thread panics. Shard threads are joined only at
+    // shutdown, so an unwinding panic during normal operation would leave an
+    // N-1-shard server listening and answering while every key owned by the
+    // dead shard is silently gone — worse than a crash for a database. The
+    // default hook runs first so the panic message + backtrace still print.
+    // (Guarded by thread-name prefix: panics on connection/test/aux threads
+    // keep their normal unwind behavior.)
+    {
+        let default_hook = std::panic::take_hook();
+        std::panic::set_hook(Box::new(move |info| {
+            default_hook(info);
+            let thread = std::thread::current();
+            let name = thread.name().unwrap_or("");
+            if name.starts_with("shard-") {
+                eprintln!(
+                    "FATAL: shard thread '{name}' panicked; aborting the whole \
+                     process rather than serving with a dead shard"
+                );
+                std::process::abort();
+            }
+        }));
+    }
+
     // ── moon.conf integration: build merged argv before clap parse ──────────
     // `merge_conf_argv` scans argv for a positional conf path (argv[1] not
     // starting with `-`) or `--config FILE`.  When found it loads and tokenises
@@ -1684,7 +1707,12 @@ fn main() -> anyhow::Result<()> {
     }
     cancel_token.cancel();
     for handle in shard_handles {
-        let _ = handle.join();
+        // A shard panic normally aborts via the panic hook above; if one
+        // still surfaces here (hook replaced, non-unwind edge), say so
+        // instead of silently swallowing it.
+        if let Err(e) = handle.join() {
+            tracing::error!("shard thread panicked during shutdown: {e:?}");
+        }
     }
 
     info!("Server shut down");
diff --git a/tests/busy_poll_idle.rs b/tests/busy_poll_idle.rs
new file mode 100644
index 00000000..f689596a
--- /dev/null
+++ b/tests/busy_poll_idle.rs
@@ -0,0 +1,179 @@
+//! `--io-busy-poll-us` must not burn CPU on an IDLE server.
+//!
+//! RSS/CPU/OOM review item 2 (CPU lens): the poll-mode park spins its full
+//! budget on EVERY park even with zero connections and zero work — an idle
+//! server burns ~budget/park-interval CPU per shard thread, forever (the
+//! "zombie CPU eater" when such a server is orphaned). The spin exists to
+//! delete wake latency UNDER TRAFFIC; after an idle window it must disengage
+//! and fall back to plain blocking parks, re-arming on the next real event.
+//!
+//! RED before the fix: an idle 4-shard server with `--io-busy-poll-us 200`
+//! accumulates CPU time at roughly 0.8 cores (200µs spin per ~1ms timer
+//! park × 4 shard threads); the 3s idle window below measures well over
+//! 500ms of CPU. GREEN after: background ticks only, well under the bound.
+//!
+//! Linux-only (reads /proc/<pid>/stat). The busy-poll CLI flag forces the
+//! epoll/kqueue LegacyDriver, so this exercises the patched spin path on
+//! both io_uring-capable and MOON_NO_URING environments.
+//!
+//! monoio-only: under the tokio runtime `--io-busy-poll-us` is an explicit
+//! no-op (main.rs warns and skips the spin wiring), so there is no spin to
+//! measure — and a debug-build tokio server's unrelated idle-tick cost
+//! (~0.9s/3s at 4 shards on CI) would fail the bound spuriously.
+
+#![cfg(all(unix, target_os = "linux", feature = "runtime-monoio"))]
+#![allow(clippy::unwrap_used)]
+
+use std::io::{Read, Write};
+use std::net::TcpStream;
+use std::process::{Child, Command};
+use std::time::{Duration, Instant};
+
+fn moon_binary() -> std::path::PathBuf {
+    std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon"))
+}
+
+fn free_port() -> u16 {
+    let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0");
+    let p = l.local_addr().expect("local_addr").port();
+    drop(l);
+    p
+}
+
+struct ServerGuard(Child);
+
+impl Drop for ServerGuard {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+
+fn wait_for_ready(port: u16, deadline: Duration) -> bool {
+    let start = Instant::now();
+    while start.elapsed() < deadline {
+        let addr = format!("127.0.0.1:{port}");
+        if let Ok(mut s) =
+            TcpStream::connect_timeout(&addr.parse().expect("addr"), Duration::from_millis(200))
+        {
+            s.set_read_timeout(Some(Duration::from_millis(500))).ok();
+            s.set_write_timeout(Some(Duration::from_millis(500))).ok();
+            if s.write_all(b"PING\r\n").is_ok() {
+                let mut buf = [0u8; 16];
+                if let Ok(n) = s.read(&mut buf) {
+                    if buf[..n].windows(4).any(|w| w == b"PONG") {
+                        return true;
+                    }
+                }
+            }
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+    false
+}
+
+/// Process CPU time (user + system) in seconds from /proc/<pid>/stat.
+fn cpu_seconds(pid: u32) -> f64 {
+    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).expect("read /proc stat");
+    // comm can contain spaces — split after the closing paren.
+    let after = stat.rsplit_once(')').expect("stat paren").1;
+    let fields: Vec<&str> = after.split_whitespace().collect();
+    // After ')': field[0]=state ... utime is the 12th, stime the 13th.
+    let utime: u64 = fields[11].parse().expect("utime");
+    let stime: u64 = fields[12].parse().expect("stime");
+    // SAFETY: sysconf(_SC_CLK_TCK) reads a static kernel constant; it takes
+    // no pointers and touches no shared state.
+    let hz = unsafe { libc::sysconf(libc::_SC_CLK_TCK) } as f64;
+    (utime + stime) as f64 / hz
+}
+
+#[test]
+fn busy_poll_idle_server_does_not_burn_cpu() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let port = free_port();
+    let child = Command::new(moon_binary())
+        .args([
+            "--port",
+            &port.to_string(),
+            "--dir",
+            &dir.path().to_string_lossy(),
+            "--shards",
+            "4",
+            "--appendonly",
+            "no",
+            "--io-busy-poll-us",
+            "200",
+        ])
+        .stdout(std::fs::File::create(dir.path().join("moon.stdout.log")).expect("stdout log"))
+        .stderr(std::fs::File::create(dir.path().join("moon.stderr.log")).expect("stderr log"))
+        .spawn()
+        .expect("spawn moon");
+    let guard = ServerGuard(child);
+    let pid = guard.0.id();
+
+    assert!(
+        wait_for_ready(port, Duration::from_secs(15)),
+        "server never ready"
+    );
+
+    // Settle past startup (recovery scan, first ticks) AND past the idle
+    // disengage window, then measure a fully idle interval.
+    std::thread::sleep(Duration::from_secs(1));
+    let before = cpu_seconds(pid);
+    std::thread::sleep(Duration::from_secs(3));
+    let burned = cpu_seconds(pid) - before;
+
+    // RED (spin always engaged): ~200µs per ~1ms park × 4 shards ≈ 2.4s here.
+    // GREEN (idle disengage): background ticks only — comfortably < 0.5s
+    // even on a loaded CI runner.
+    assert!(
+        burned < 0.5,
+        "idle 4-shard server with --io-busy-poll-us 200 burned {burned:.2}s \
+         CPU in 3s — spin never disengages when idle"
+    );
+}
+
+/// Traffic after an idle period must still be served (the disengaged spin
+/// re-arms via the normal wake path — correctness, not latency, asserted).
+#[test]
+fn busy_poll_recovers_after_idle() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let port = free_port();
+    let child = Command::new(moon_binary())
+        .args([
+            "--port",
+            &port.to_string(),
+            "--dir",
+            &dir.path().to_string_lossy(),
+            "--shards",
+            "2",
+            "--appendonly",
+            "no",
+            "--io-busy-poll-us",
+            "40",
+        ])
+        .stdout(std::fs::File::create(dir.path().join("moon.stdout.log")).expect("stdout log"))
+        .stderr(std::fs::File::create(dir.path().join("moon.stderr.log")).expect("stderr log"))
+        .spawn()
+        .expect("spawn moon");
+    let guard = ServerGuard(child);
+
+    assert!(
+        wait_for_ready(port, Duration::from_secs(15)),
+        "server never ready"
+    );
+    // Idle well past any disengage window, then talk to every shard's path.
+    std::thread::sleep(Duration::from_secs(2));
+    let addr = format!("127.0.0.1:{port}");
+    let mut s = TcpStream::connect_timeout(&addr.parse().expect("addr"), Duration::from_secs(2))
+        .expect("post-idle connect");
+    s.set_read_timeout(Some(Duration::from_secs(5))).ok();
+    for i in 0..100 {
+        s.write_all(format!("SET pk:{i} v{i}\r\nGET pk:{i}\r\n").as_bytes())
+            .expect("post-idle write");
+        let mut buf = [0u8; 128];
+        let n = s.read(&mut buf).expect("post-idle read");
+        assert!(n > 0, "no reply after idle re-engage (op {i})");
+    }
+    drop(guard);
+}
diff --git a/tests/shard_panic_abort.rs b/tests/shard_panic_abort.rs
new file mode 100644
index 00000000..d690b390
--- /dev/null
+++ b/tests/shard_panic_abort.rs
@@ -0,0 +1,119 @@
+//! A panicked shard thread must take the whole process down (fail-fast),
+//! never leave an N-1-shard zombie silently serving partial data.
+//!
+//! RSS/CPU/OOM review item 7 (zombie lens): shard threads are joined only at
+//! shutdown and the join result was discarded (`let _ = handle.join()`), so a
+//! shard panic during normal operation left the server listening and
+//! answering for the surviving shards while every key owned by the dead
+//! shard was silently gone — worse than a crash for a database.
+//!
+//! RED before the fix: `DEBUG PANIC` kills one shard thread, the process
+//! stays alive past the timeout. GREEN: a shard-thread panic hook aborts.
+
+#![cfg(unix)]
+#![allow(clippy::unwrap_used)]
+
+use std::io::{Read, Write};
+use std::net::TcpStream;
+use std::process::{Child, Command};
+use std::time::{Duration, Instant};
+
+fn moon_binary() -> std::path::PathBuf {
+    std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon"))
+}
+
+fn free_port() -> u16 {
+    let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0");
+    let p = l.local_addr().expect("local_addr").port();
+    drop(l);
+    p
+}
+
+struct ServerGuard(Child);
+
+impl Drop for ServerGuard {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+
+fn wait_for_ready(port: u16, deadline: Duration) -> bool {
+    let start = Instant::now();
+    while start.elapsed() < deadline {
+        let addr = format!("127.0.0.1:{port}");
+        if let Ok(mut s) =
+            TcpStream::connect_timeout(&addr.parse().expect("addr"), Duration::from_millis(200))
+        {
+            s.set_read_timeout(Some(Duration::from_millis(500))).ok();
+            s.set_write_timeout(Some(Duration::from_millis(500))).ok();
+            if s.write_all(b"PING\r\n").is_ok() {
+                let mut buf = [0u8; 16];
+                if let Ok(n) = s.read(&mut buf) {
+                    if buf[..n].windows(4).any(|w| w == b"PONG") {
+                        return true;
+                    }
+                }
+            }
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+    false
+}
+
+fn assert_panic_aborts_process(shards: u32, label: &str) {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let port = free_port();
+    let child = Command::new(moon_binary())
+        .args([
+            "--port",
+            &port.to_string(),
+            "--dir",
+            &dir.path().to_string_lossy(),
+            "--shards",
+            &shards.to_string(),
+            "--appendonly",
+            "no",
+        ])
+        .stdout(std::fs::File::create(dir.path().join("moon.stdout.log")).expect("stdout log"))
+        .stderr(std::fs::File::create(dir.path().join("moon.stderr.log")).expect("stderr log"))
+        .spawn()
+        .expect("spawn moon");
+    let mut guard = ServerGuard(child);
+
+    assert!(
+        wait_for_ready(port, Duration::from_secs(15)),
+        "[{label}] server never ready"
+    );
+
+    let addr = format!("127.0.0.1:{port}");
+    let mut s = TcpStream::connect_timeout(&addr.parse().expect("addr"), Duration::from_secs(2))
+        .expect("connect");
+    s.set_read_timeout(Some(Duration::from_secs(2))).ok();
+    s.write_all(b"DEBUG PANIC\r\n").expect("send DEBUG PANIC");
+    // The connection is expected to die; the reply (if any) is irrelevant.
+    let mut buf = [0u8; 64];
+    let _ = s.read(&mut buf);
+
+    let start = Instant::now();
+    while start.elapsed() < Duration::from_secs(5) {
+        if guard.0.try_wait().expect("try_wait").is_some() {
+            return; // process died fail-fast — GREEN
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+    panic!(
+        "[{label}] process still alive 5s after a shard panic — N-1-shard \
+         zombie serving partial data (see moon.stderr.log in the temp dir)"
+    );
+}
+
+#[test]
+fn shard_panic_aborts_process_shards_1() {
+    assert_panic_aborts_process(1, "s1");
+}
+
+#[test]
+fn shard_panic_aborts_process_shards_4() {
+    assert_panic_aborts_process(4, "s4");
+}
diff --git a/tests/sigterm_shutdown.rs b/tests/sigterm_shutdown.rs
index e56694d8..0384813c 100644
--- a/tests/sigterm_shutdown.rs
+++ b/tests/sigterm_shutdown.rs
@@ -76,6 +76,26 @@ fn send_sigterm(pid: u32) {
 /// `extra_args` is appended to the base arg list so individual cases can enable
 /// optional subsystems (e.g. `--admin-port`) that spawn threads earlier in main().
 fn assert_sigterm_clean_exit(label: &str, extra_args: &[&str]) {
+    assert_sigterm_clean_exit_shards(label, 1, extra_args, ConnAtSigterm::None)
+}
+
+/// Connection state to hold when SIGTERM lands — the zombie review
+/// (tmp/RSS-CPU-REVIEW.md item 1) implicates the per-shard accept path, and
+/// the bench-harness hangs happened with recent/live connections, so the
+/// matrix covers both.
+enum ConnAtSigterm {
+    /// No client connected (beyond the readiness probe, already closed).
+    None,
+    /// One client connection held OPEN across the SIGTERM.
+    HeldOpen,
+}
+
+fn assert_sigterm_clean_exit_shards(
+    label: &str,
+    shards: u32,
+    extra_args: &[&str],
+    conn: ConnAtSigterm,
+) {
     let port = free_port();
     let dir = std::env::temp_dir().join(format!(
         "moon-sigterm-{}-{}-{}",
@@ -88,6 +108,7 @@ fn assert_sigterm_clean_exit(label: &str, extra_args: &[&str]) {
     // Bind to locals so the &str slices in base_args are not temporaries.
     let port_str = port.to_string();
     let dir_str = dir.to_string_lossy().into_owned();
+    let shards_str = shards.to_string();
 
     let mut base_args: Vec<&str> = vec![
         "--port",
@@ -97,7 +118,7 @@ fn assert_sigterm_clean_exit(label: &str, extra_args: &[&str]) {
         "--appendonly",
         "no",
         "--shards",
-        "1",
+        &shards_str,
     ];
     base_args.extend_from_slice(extra_args);
 
@@ -121,6 +142,24 @@ fn assert_sigterm_clean_exit(label: &str, extra_args: &[&str]) {
         );
     }
 
+    // Optionally hold a live client connection across the SIGTERM (verified
+    // alive with a PING round-trip so the conn is fully registered on a shard).
+    let held = match conn {
+        ConnAtSigterm::None => None,
+        ConnAtSigterm::HeldOpen => {
+            let addr = format!("127.0.0.1:{port}");
+            let mut s =
+                TcpStream::connect_timeout(&addr.parse().expect("addr"), Duration::from_secs(2))
+                    .expect("held conn connect");
+            s.set_read_timeout(Some(Duration::from_secs(2))).ok();
+            s.write_all(b"PING\r\n").expect("held conn ping");
+            let mut buf = [0u8; 16];
+            let n = s.read(&mut buf).expect("held conn pong");
+            assert!(buf[..n].windows(4).any(|w| w == b"PONG"), "held conn PONG");
+            Some(s)
+        }
+    };
+
     // Send SIGTERM and wait for the process to exit.
     send_sigterm(pid);
 
@@ -147,6 +186,7 @@ fn assert_sigterm_clean_exit(label: &str, extra_args: &[&str]) {
         label,
         status
     );
+    drop(held);
 }
 
 #[test]
@@ -168,3 +208,141 @@ fn sigterm_clean_exit_with_admin_port() {
     let admin_port_str = admin_port.to_string();
     assert_sigterm_clean_exit("admin", &["--admin-port", &admin_port_str]);
 }
+
+// ── Zombie-review matrix (tmp/RSS-CPU-REVIEW.md item 1) ─────────────────────
+//
+// The detached monoio per-shard accept task never observes the shutdown
+// token; the documented bench-harness hangs involved multi-shard SO_REUSEPORT
+// servers and live/recent connections. Each ingredient gets a case.
+
+#[test]
+fn sigterm_clean_exit_shards_4_idle() {
+    assert_sigterm_clean_exit_shards("s4-idle", 4, &[], ConnAtSigterm::None);
+}
+
+#[test]
+fn sigterm_clean_exit_shards_4_held_conn() {
+    assert_sigterm_clean_exit_shards("s4-held", 4, &[], ConnAtSigterm::HeldOpen);
+}
+
+#[test]
+fn sigterm_clean_exit_shards_1_held_conn() {
+    assert_sigterm_clean_exit_shards("s1-held", 1, &[], ConnAtSigterm::HeldOpen);
+}
+
+/// The zombie-CPU-eater composition: busy-poll spin configured AND SIGTERM'd.
+#[test]
+fn sigterm_clean_exit_busy_poll() {
+    assert_sigterm_clean_exit_shards(
+        "s2-busypoll",
+        2,
+        &["--io-busy-poll-us", "40"],
+        ConnAtSigterm::HeldOpen,
+    );
+}
+
+/// Bench-harness shape: SIGTERM lands while many clients hammer pipelined
+/// writes (the documented GCE hang always involved live benchmark traffic).
+/// Writer threads tolerate every I/O error — after SIGTERM their connections
+/// are expected to die; the only assertion is that the SERVER exits cleanly.
+#[test]
+fn sigterm_clean_exit_under_write_storm() {
+    let port = free_port();
+    let dir = std::env::temp_dir().join(format!(
+        "moon-sigterm-{}-{}-storm",
+        std::process::id(),
+        port
+    ));
+    std::fs::create_dir_all(&dir).expect("mk tmpdir");
+    let port_str = port.to_string();
+    let dir_str = dir.to_string_lossy().into_owned();
+
+    let mut child = Command::new(moon_binary())
+        .args([
+            "--port",
+            &port_str,
+            "--dir",
+            &dir_str,
+            "--appendonly",
+            "yes",
+            "--shards",
+            "4",
+        ])
+        .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log"))
+        .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log"))
+        .spawn()
+        .expect("spawn moon");
+    let pid = child.id();
+
+    if !wait_for_ready(port, Duration::from_secs(15)) {
+        let _ = child.kill();
+        let _ = child.wait();
+        panic!("[storm] server did not become ready within 15s on port {port}");
+    }
+
+    // 16 writer threads, each pipelining SETs in a tight loop until the
+    // connection dies or the stop flag flips.
+    let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let writers: Vec<_> = (0..16)
+        .map(|t| {
+            let stop = stop.clone();
+            let addr = format!("127.0.0.1:{port}");
+            std::thread::spawn(move || {
+                let Ok(mut s) = TcpStream::connect_timeout(
+                    &addr.parse().expect("addr"),
+                    Duration::from_secs(2),
+                ) else {
+                    return;
+                };
+                s.set_write_timeout(Some(Duration::from_millis(500))).ok();
+                s.set_read_timeout(Some(Duration::from_millis(500))).ok();
+                let mut n = 0u64;
+                let mut buf = [0u8; 4096];
+                while !stop.load(std::sync::atomic::Ordering::Relaxed) {
+                    // 8-deep pipeline of SETs; ignore all errors (the server
+                    // is being killed under us — that's the point).
+                    let mut pipe = Vec::new();
+                    for i in 0..8 {
+                        pipe.extend_from_slice(
+                            format!("SET k:{t}:{}:{i} v{n}\r\n", n % 512).as_bytes(),
+                        );
+                    }
+                    if s.write_all(&pipe).is_err() {
+                        break;
+                    }
+                    let _ = s.read(&mut buf);
+                    n += 1;
+                }
+            })
+        })
+        .collect();
+
+    // Let the storm establish, then SIGTERM mid-flight.
+    std::thread::sleep(Duration::from_millis(500));
+    send_sigterm(pid);
+
+    let deadline = Instant::now();
+    let status = loop {
+        match child.try_wait().expect("try_wait") {
+            Some(s) => break s,
+            None => {
+                if deadline.elapsed() >= Duration::from_secs(10) {
+                    stop.store(true, std::sync::atomic::Ordering::Relaxed);
+                    let _ = child.kill();
+                    let _ = child.wait();
+                    panic!("[storm] server did not exit within 10s after SIGTERM");
+                }
+                std::thread::sleep(Duration::from_millis(100));
+            }
+        }
+    };
+    stop.store(true, std::sync::atomic::Ordering::Relaxed);
+    for w in writers {
+        let _ = w.join();
+    }
+    assert_eq!(
+        status.code(),
+        Some(0),
+        "[storm] expected exit code 0 after SIGTERM under load, got {status:?}"
+    );
+}
diff --git a/tests/vector_del_unindex.rs b/tests/vector_del_unindex.rs
index 4335ac56..4ebe74fd 100644
--- a/tests/vector_del_unindex.rs
+++ b/tests/vector_del_unindex.rs
@@ -196,19 +196,32 @@ impl Client {
         self.writer.write_all(&buf).expect("send pipeline");
         cmds.iter().map(|_| self.parse()).collect()
     }
+
+    /// Fallible PING for the readiness probe: a connection accepted while the
+    /// server is still bringing up its per-shard SO_REUSEPORT listeners can be
+    /// RESET mid-read — that must retry with a fresh connection, not panic.
+    fn try_ping(&mut self) -> std::io::Result<bool> {
+        self.writer.write_all(b"*1\r\n$4\r\nPING\r\n")?;
+        let mut buf = [0u8; 7];
+        self.reader.read_exact(&mut buf)?;
+        Ok(&buf == b"+PONG\r\n")
+    }
 }
 
 fn wait_ready(port: u16) -> Client {
-    let mut c = Client::connect(port);
     let start = Instant::now();
     loop {
-        match c.cmd(&[b"PING"]) {
-            V::Simple(s) if s == "PONG" => return c,
-            _ if start.elapsed() < Duration::from_secs(30) => {
-                std::thread::sleep(Duration::from_millis(100));
-            }
-            other => panic!("server never answered PING: {other:?}"),
+        let mut c = Client::connect(port);
+        // Any I/O error (or a non-PONG answer, which would desync the framing)
+        // drops this connection and probes again on a new one.
+        if let Ok(true) = c.try_ping() {
+            return c;
         }
+        assert!(
+            start.elapsed() < Duration::from_secs(30),
+            "server never answered PING on port {port}"
+        );
+        std::thread::sleep(Duration::from_millis(100));
     }
 }
 
diff --git a/vendor/monoio/src/driver/legacy/mod.rs b/vendor/monoio/src/driver/legacy/mod.rs
index 92c45487..74315fad 100644
--- a/vendor/monoio/src/driver/legacy/mod.rs
+++ b/vendor/monoio/src/driver/legacy/mod.rs
@@ -127,6 +127,50 @@ fn moon_spin_probe() -> bool {
     })
 }
 
+// moon patch: idle disengage for the spin-poll. The spin exists to delete
+// wake latency UNDER TRAFFIC; without a gate an idle thread burns the full
+// budget on EVERY park forever (~budget/park-interval CPU per shard thread,
+// e.g. ~4%/core at 40µs vs 1ms timer parks — the "zombie CPU eater" when a
+// stray server is orphaned). Once this thread has seen no readiness events
+// and no host work for MOON_SPIN_IDLE_DISENGAGE_US (default 10ms; 0 = never
+// disengage), parks fall back to plain blocking polls. The next real event
+// still arrives via the normal wake path (a one-time ~µs penalty) and
+// re-arms the spin. Timer-only wakeups (empty poll after timeout) do NOT
+// count as events, so a quiescent server stays disengaged.
+thread_local! {
+    static MOON_SPIN_LAST_EVENT: std::cell::Cell<Option<std::time::Instant>> =
+        const { std::cell::Cell::new(None) };
+}
+
+fn moon_spin_idle_disengage_us() -> u64 {
+    static V: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
+    *V.get_or_init(|| {
+        std::env::var("MOON_SPIN_IDLE_DISENGAGE_US")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(10_000)
+    })
+}
+
+fn moon_spin_engaged() -> bool {
+    let win = moon_spin_idle_disengage_us();
+    if win == 0 {
+        return true;
+    }
+    MOON_SPIN_LAST_EVENT.with(|c| match c.get() {
+        // First park on this thread: start the window now (covers startup).
+        None => {
+            c.set(Some(std::time::Instant::now()));
+            true
+        }
+        Some(t) => t.elapsed() < std::time::Duration::from_micros(win),
+    })
+}
+
+fn moon_spin_note_event() {
+    MOON_SPIN_LAST_EVENT.with(|c| c.set(Some(std::time::Instant::now())));
+}
+
 #[allow(dead_code)]
 impl LegacyDriver {
     const DEFAULT_ENTRIES: u32 = 1024;
@@ -235,7 +279,9 @@ impl LegacyDriver {
         #[cfg(unix)]
         if need_wait {
             let spin_us = moon_epoll_spin_budget_us();
-            if spin_us > 0 {
+            // Idle disengage (see MOON_SPIN_LAST_EVENT above): only pay the
+            // spin budget while events have arrived recently.
+            if spin_us > 0 && moon_spin_engaged() {
                 let budget = match timeout {
                     Some(d) => d.min(Duration::from_micros(spin_us)),
                     None => Duration::from_micros(spin_us),
@@ -262,6 +308,7 @@ impl LegacyDriver {
                     }
                     if !events.is_empty() {
                         polled = true;
+                        moon_spin_note_event();
                         break;
                     }
                     if hooks && moon_spin_probe() {
@@ -291,6 +338,7 @@ impl LegacyDriver {
                     // Host work is pending and a local wake was delivered —
                     // return to the executor instead of blocking.
                     polled = true;
+                    moon_spin_note_event();
                 }
             }
         }
@@ -302,6 +350,12 @@ impl LegacyDriver {
                 Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
                 Err(e) => return Err(e),
             }
+            // moon patch: real fd/waker events (not bare timer expiry)
+            // re-arm the idle-disengaged spin for the next park.
+            #[cfg(unix)]
+            if events.iter().next().is_some() {
+                moon_spin_note_event();
+            }
         }
         #[cfg(unix)]
         let iter = events.iter();