feat(command): real SHUTDOWN [NOSAVE|SAVE] + clean-shutdown AOF data-loss fix (task #27)

CHANGELOG.md

@@ -6,6 +6,56 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added — real SHUTDOWN [NOSAVE|SAVE] (task #27)
+
+`SHUTDOWN` was previously a stub that always replied `ERR Errors trying to
+SHUTDOWN. Check logs.` and never terminated the process. It is now handled
+at the connection-handler level (like BGSAVE/ACL, alongside all three
+dispatch paths: `handler_single`, `handler_sharded`, `handler_monoio`):
+
+- Parses the optional `NOSAVE` / `SAVE` modifier (Redis parity: bare
+  `SHUTDOWN` forces a save iff RDB save points are configured; `NOSAVE`
+  always skips it; `SAVE` always forces it). `ABORT` is rejected (Moon's
+  SHUTDOWN runs synchronously to completion, so there is never an
+  in-progress shutdown); `FORCE`/`NOW` are accepted no-ops.
+- A forced save that fails replies an error and the server stays up
+  (single-shard: synchronous `SAVE`; sharded/monoio: the cooperative
+  per-shard BGSAVE snapshot, polled with a bounded timeout).
+- On success, triggers the exact same `CancellationToken`-driven graceful
+  shutdown sequence already used for SIGTERM — per-shard WAL/AOF flush +
+  fsync across every plane, accept-loop stop, and clean connection
+  teardown — rather than reimplementing it. No reply is sent (Redis
+  parity: the client observes the connection close).
+- ACL category was already `admin/dangerous` (`DNG`) in the command
+  registry; unchanged.
+
+Coverage: `tests/shutdown_integration.rs` (NOSAVE prompt exit + waitpid
+status, AOF durability across a SHUTDOWN→restart round trip, a failed
+forced SAVE keeps the server up, syntax errors keep the server up) plus a
+cross-shard durability smoke section in `scripts/test-consistency.sh`.
+
+### Fixed — AOF writer manifest-wait/cancellation data-loss race (found via task #27)
+
+Writing the SHUTDOWN durability test surfaced a real, pre-existing bug: on
+first boot, main.rs's recovery creates the AOF manifest on a separate
+thread/task from the one that starts accepting connections, so a client can
+already be writing (queuing `Append` messages) while the AOF writer thread
+is still polling for that manifest to appear (`src/persistence/aof/writer_task.rs`,
+three writer-loop variants). That polling loop checked
+`cancel.is_cancelled()` *before* attempting `AofManifest::load` each
+iteration — a graceful shutdown landing in the same ~50ms poll window made
+the writer return immediately without ever loading the (by-then-current)
+manifest, silently discarding every already-queued `Append`. Reproduced
+reliably (>80% of runs) by sending `SHUTDOWN` within tens of milliseconds of
+boot: `AOF writer: cancelled while waiting for manifest` in the log, and a
+0-byte incr AOF file. Fixed by trying the load first in all three affected
+loops (TopLevel-monoio, PerShard-tokio, PerShard-monoio); a manifest that
+exists by the time the writer gets there is never missed just because a
+shutdown signal happened to land in the same instant. In practice this
+window was rarely hit via SIGTERM (real signal delivery has enough latency
+to clear it), which is presumably why it went unnoticed until a command
+that can call `cancel()` synchronously, milliseconds after the connection
+that issued the write, existed.
 ### Changed — CI pipeline optimization (round 2: cargo-nextest)
 
 - CI test steps (Linux/macOS/Windows) now run under `cargo nextest run

scripts/test-commands.sh

@@ -784,6 +784,13 @@ if should_run "persistence"; then
 
     assert_moon "BGSAVE"               "Background saving started" BGSAVE
     sleep 1
+
+    # SHUTDOWN [NOSAVE|SAVE] is intentionally NOT exercised in this section:
+    # it terminates the server process this whole script shares across every
+    # other category, which would abort the run. Coverage lives in
+    # tests/shutdown_integration.rs (spawns its own throwaway server per
+    # case) and scripts/test-consistency.sh (cross-shard durability smoke
+    # check against its own dedicated instance).
 fi
 
 # ===========================================================================

scripts/test-consistency.sh

@@ -1450,6 +1450,83 @@ fi
 # MEMORY DOCTOR: Moon-specific schema, not parity-tested against Redis.
 # Coverage: integration test tests/memory_doctor_response.rs + test-commands.sh.
 
+# ===========================================================================
+# SHUTDOWN [NOSAVE|SAVE] -- task #27
+#
+# Destructive by nature (the command exits the server), so it cannot share
+# $RUST_PID with the sections above -- it runs against its own throwaway
+# instance on a dedicated port/dir and cleans up after itself. Full
+# correctness/edge-case coverage (syntax errors, forced-SAVE failure keeps
+# the server up, etc.) lives in the Rust integration test
+# tests/shutdown_integration.rs; this section is the cross-shard durability
+# smoke check the new-command convention asks for.
+# ===========================================================================
+echo ""
+echo "=== SHUTDOWN [NOSAVE|SAVE] ==="
+
+PORT_SHUTDOWN=$((PORT_RUST + 500))
+SHUTDOWN_DIR=$(mktemp -d /tmp/moon-shutdown-dir.XXXXXX)
+
+# NOSAVE: exits promptly, appendonly=no so no durability is expected -- this
+# only checks the process actually terminates instead of erroring forever.
+"$RUST_BINARY" --port "$PORT_SHUTDOWN" --shards 1 --dir "$SHUTDOWN_DIR" \
+    --appendonly no --disk-free-min-pct 0 >/dev/null 2>&1 &
+SHUTDOWN_PID=$!
+for _ in $(seq 1 50); do
+    redis-cli -p "$PORT_SHUTDOWN" PING >/dev/null 2>&1 && break
+    sleep 0.1
+done
+redis-cli -p "$PORT_SHUTDOWN" SHUTDOWN NOSAVE >/dev/null 2>&1 || true
+SHUTDOWN_EXITED=false
+for _ in $(seq 1 50); do
+    kill -0 "$SHUTDOWN_PID" 2>/dev/null || { SHUTDOWN_EXITED=true; break; }
+    sleep 0.1
+done
+if $SHUTDOWN_EXITED; then
+    PASS=$((PASS + 1)); echo "  PASS: SHUTDOWN NOSAVE exits promptly"
+else
+    FAIL=$((FAIL + 1)); echo "  FAIL: SHUTDOWN NOSAVE did not exit within 5s"
+    kill -9 "$SHUTDOWN_PID" 2>/dev/null || true
+fi
+wait "$SHUTDOWN_PID" 2>/dev/null || true
+rm -rf "$SHUTDOWN_DIR"
+
+# appendonly=yes: SHUTDOWN must flush the AOF durably -- write, shut down,
+# restart, and confirm the key survived (no kill-9 tail loss on a clean exit).
+SHUTDOWN_DIR=$(mktemp -d /tmp/moon-shutdown-dir.XXXXXX)
+"$RUST_BINARY" --port "$PORT_SHUTDOWN" --shards 1 --dir "$SHUTDOWN_DIR" \
+    --appendonly yes --disk-free-min-pct 0 >/dev/null 2>&1 &
+SHUTDOWN_PID=$!
+for _ in $(seq 1 50); do
+    redis-cli -p "$PORT_SHUTDOWN" PING >/dev/null 2>&1 && break
+    sleep 0.1
+done
+redis-cli -p "$PORT_SHUTDOWN" SET shutdown:durable hello >/dev/null 2>&1
+redis-cli -p "$PORT_SHUTDOWN" SHUTDOWN NOSAVE >/dev/null 2>&1 || true
+for _ in $(seq 1 50); do
+    kill -0 "$SHUTDOWN_PID" 2>/dev/null || break
+    sleep 0.1
+done
+wait "$SHUTDOWN_PID" 2>/dev/null || true
+
+"$RUST_BINARY" --port "$PORT_SHUTDOWN" --shards 1 --dir "$SHUTDOWN_DIR" \
+    --appendonly yes --disk-free-min-pct 0 >/dev/null 2>&1 &
+SHUTDOWN_PID=$!
+for _ in $(seq 1 50); do
+    redis-cli -p "$PORT_SHUTDOWN" PING >/dev/null 2>&1 && break
+    sleep 0.1
+done
+SHUTDOWN_RESTORED=$(redis-cli -p "$PORT_SHUTDOWN" GET shutdown:durable 2>&1)
+if [[ "$SHUTDOWN_RESTORED" == "hello" ]]; then
+    PASS=$((PASS + 1)); echo "  PASS: SHUTDOWN flushes AOF durably (appendonly=yes survives restart)"
+else
+    FAIL=$((FAIL + 1)); echo "  FAIL: SHUTDOWN did not persist AOF durably: got '$SHUTDOWN_RESTORED'"
+fi
+kill "$SHUTDOWN_PID" 2>/dev/null || true
+wait "$SHUTDOWN_PID" 2>/dev/null || true
+pkill -f "moon.*${PORT_SHUTDOWN}" 2>/dev/null || true
+rm -rf "$SHUTDOWN_DIR"
+
 # Restart moon with the originally-requested shard count so summary works.
 start_moon_with_shards "$SHARDS" || true
 

src/command/mod.rs

@@ -765,9 +765,19 @@ fn dispatch_inner(
         (8, b's') => {
             // SMEMBERS SETRANGE SHUTDOWN
             if cmd.eq_ignore_ascii_case(b"SHUTDOWN") {
-                // Acknowledge but don't kill — actual shutdown is handled by the server
+                // Real SHUTDOWN handling (parse NOSAVE/SAVE, force a durable
+                // save, then trigger the same graceful shutdown sequence used
+                // by SIGTERM) is intercepted at the connection-handler level
+                // (handler_single.rs / handler_sharded / handler_monoio) --
+                // like BGSAVE/ACL, it needs access to the full database
+                // handle and the shard's shutdown CancellationToken, neither
+                // of which this single-`Database` dispatch path has. This
+                // arm is reached only when SHUTDOWN slips past that
+                // intercept (e.g. queued inside MULTI/EXEC), where Redis
+                // itself also refuses admin commands -- fail closed rather
+                // than silently no-op.
                 return resp(Frame::Error(Bytes::from_static(
-                    b"ERR Errors trying to SHUTDOWN. Check logs.",
+                    b"ERR SHUTDOWN is not allowed in this context",
                 )));
             }
             if cmd.eq_ignore_ascii_case(b"SMEMBERS") {

src/command/persistence.rs

@@ -389,6 +389,79 @@ pub fn handle_save(db: &SharedDatabases, dir: &str, dbfilename: &str) -> Frame {
     }
 }
 
+/// Modifier parsed from SHUTDOWN's optional argument.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ShutdownSaveMode {
+    /// `SHUTDOWN SAVE` — force a synchronous save regardless of configured
+    /// save points.
+    Save,
+    /// `SHUTDOWN NOSAVE` — skip any extra RDB save (Redis parity: whatever
+    /// AOF/WAL already made durable is all that survives).
+    NoSave,
+    /// Bare `SHUTDOWN` — save iff RDB save points are configured (Redis
+    /// parity).
+    Default,
+}
+
+/// Parse SHUTDOWN's argument list.
+///
+/// Redis SHUTDOWN accepts an optional single modifier: `NOSAVE` or `SAVE`.
+/// `ABORT` is rejected here — Moon's SHUTDOWN runs synchronously to
+/// completion inside the command itself, so there is never an in-progress
+/// shutdown to abort. The `FORCE`/`NOW` timeout-override modifiers Redis
+/// added later are accepted as no-ops for client compatibility: Moon's
+/// shutdown path already flushes durably via the same bounded sequence used
+/// for SIGTERM and has no separate "hung shutdown" timeout to override.
+pub fn parse_shutdown_args(args: &[Frame]) -> Result<ShutdownSaveMode, Frame> {
+    let mut mode = None;
+    for a in args {
+        let bytes: &[u8] = match a {
+            Frame::BulkString(b) => b.as_ref(),
+            Frame::SimpleString(b) => b.as_ref(),
+            _ => return Err(Frame::Error(Bytes::from_static(b"ERR syntax error"))),
+        };
+        if bytes.eq_ignore_ascii_case(b"NOSAVE") {
+            if mode.is_some() {
+                return Err(Frame::Error(Bytes::from_static(b"ERR syntax error")));
+            }
+            mode = Some(ShutdownSaveMode::NoSave);
+        } else if bytes.eq_ignore_ascii_case(b"SAVE") {
+            if mode.is_some() {
+                return Err(Frame::Error(Bytes::from_static(b"ERR syntax error")));
+            }
+            mode = Some(ShutdownSaveMode::Save);
+        } else if bytes.eq_ignore_ascii_case(b"FORCE") || bytes.eq_ignore_ascii_case(b"NOW") {
+            continue;
+        } else if bytes.eq_ignore_ascii_case(b"ABORT") {
+            return Err(Frame::Error(Bytes::from_static(
+                b"ERR No shutdown in progress",
+            )));
+        } else {
+            return Err(Frame::Error(Bytes::from_static(b"ERR syntax error")));
+        }
+    }
+    Ok(mode.unwrap_or(ShutdownSaveMode::Default))
+}
+
+/// Poll interval used by SHUTDOWN SAVE in sharded/monoio mode while waiting
+/// for the cooperative per-shard BGSAVE snapshot it triggers to complete.
+pub const SHUTDOWN_SAVE_POLL_MS: u64 = 5;
+
+/// Bound on how long SHUTDOWN SAVE waits for that snapshot. A wedged shard
+/// must not hang SHUTDOWN forever; exceeding this fails the command (server
+/// stays up, Redis parity: a save that doesn't complete blocks shutdown)
+/// rather than exiting against a torn snapshot. 10s: long enough for a real
+/// (large) snapshot under normal disk I/O, short enough that an operator
+/// (or a client with a bounded read timeout) isn't left hanging.
+pub const SHUTDOWN_SAVE_TIMEOUT_MS: u64 = 10_000;
+
+/// Decide whether SHUTDOWN's `Default` mode should perform a synchronous
+/// save, mirroring Redis: save iff at least one RDB save point is
+/// configured.
+pub fn shutdown_default_should_save(save_points: Option<&str>) -> bool {
+    save_points.is_some_and(|s| !s.trim().is_empty())
+}
+
 /// LASTSAVE command: returns Unix timestamp of last successful save.
 pub fn handle_lastsave() -> Frame {
     let ts = LAST_SAVE_TIME.load(Ordering::Relaxed);
@@ -542,4 +615,76 @@ mod tests {
             other => panic!("expected Frame::Error when gate is ON, got {other:?}"),
         }
     }
+
+    // ── SHUTDOWN argument parsing (task #27) ────────────────────────────
+
+    #[test]
+    fn test_parse_shutdown_args_bare() {
+        assert_eq!(parse_shutdown_args(&[]).unwrap(), ShutdownSaveMode::Default);
+    }
+
+    #[test]
+    fn test_parse_shutdown_args_nosave() {
+        let args = [Frame::BulkString(Bytes::from_static(b"NOSAVE"))];
+        assert_eq!(
+            parse_shutdown_args(&args).unwrap(),
+            ShutdownSaveMode::NoSave
+        );
+        // Case-insensitive, matching Redis.
+        let args = [Frame::BulkString(Bytes::from_static(b"nosave"))];
+        assert_eq!(
+            parse_shutdown_args(&args).unwrap(),
+            ShutdownSaveMode::NoSave
+        );
+    }
+
+    #[test]
+    fn test_parse_shutdown_args_save() {
+        let args = [Frame::BulkString(Bytes::from_static(b"SAVE"))];
+        assert_eq!(parse_shutdown_args(&args).unwrap(), ShutdownSaveMode::Save);
+    }
+
+    #[test]
+    fn test_parse_shutdown_args_save_force_noop() {
+        // SAVE FORCE / NOW are accepted no-ops for client compatibility.
+        let args = [
+            Frame::BulkString(Bytes::from_static(b"SAVE")),
+            Frame::BulkString(Bytes::from_static(b"FORCE")),
+        ];
+        assert_eq!(parse_shutdown_args(&args).unwrap(), ShutdownSaveMode::Save);
+    }
+
+    #[test]
+    fn test_parse_shutdown_args_conflicting_modifiers_reject() {
+        let args = [
+            Frame::BulkString(Bytes::from_static(b"SAVE")),
+            Frame::BulkString(Bytes::from_static(b"NOSAVE")),
+        ];
+        assert!(matches!(parse_shutdown_args(&args), Err(Frame::Error(_))));
+    }
+
+    #[test]
+    fn test_parse_shutdown_args_abort_rejected() {
+        let args = [Frame::BulkString(Bytes::from_static(b"ABORT"))];
+        match parse_shutdown_args(&args) {
+            Err(Frame::Error(msg)) => {
+                assert!(std::str::from_utf8(&msg).unwrap().contains("No shutdown"));
+            }
+            other => panic!("expected ABORT rejection, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn test_parse_shutdown_args_garbage_rejected() {
+        let args = [Frame::BulkString(Bytes::from_static(b"BOGUS"))];
+        assert!(matches!(parse_shutdown_args(&args), Err(Frame::Error(_))));
+    }
+
+    #[test]
+    fn test_shutdown_default_should_save() {
+        assert!(!shutdown_default_should_save(None));
+        assert!(!shutdown_default_should_save(Some("")));
+        assert!(!shutdown_default_should_save(Some("   ")));
+        assert!(shutdown_default_should_save(Some("3600 1 300 100")));
+    }
 }