diff --git a/CHANGELOG.md b/CHANGELOG.md index 6c649bb2..eb5297f2 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/scripts/test-commands.sh b/scripts/test-commands.sh index 289dd721..bba560f2 100755 --- a/scripts/test-commands.sh +++ b/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 # =========================================================================== diff --git a/scripts/test-consistency.sh b/scripts/test-consistency.sh index f4c657f5..d37ed6d0 100755 --- a/scripts/test-consistency.sh +++ b/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 diff --git a/src/command/mod.rs b/src/command/mod.rs index fce3c6f3..05fd536b 100644 --- a/src/command/mod.rs +++ b/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") { diff --git a/src/command/persistence.rs b/src/command/persistence.rs index dfb390ce..39f6e938 100644 --- a/src/command/persistence.rs +++ b/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 { + 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"))); + } } diff --git a/src/persistence/aof/writer_task.rs b/src/persistence/aof/writer_task.rs index ca36fdf9..b0aee809 100644 --- a/src/persistence/aof/writer_task.rs +++ b/src/persistence/aof/writer_task.rs @@ -428,25 +428,41 @@ pub async fn aof_writer_task( // Bounded wait: check the cancellation token each iteration and enforce // a hard timeout so the writer doesn't spin forever if main.rs fails to // create the manifest (e.g. disk full, permission error). + // + // task #27 (SHUTDOWN) fix: `AofManifest::load` runs BEFORE the + // cancellation/timeout checks, not after. A client can already be + // connected and writing (queuing `Append` messages into `rx`) while + // main.rs's recovery is still mid-flight creating this manifest on a + // separate thread -- that's the documented race this loop exists to + // wait out. The old ordering checked `cancel.is_cancelled()` FIRST: + // a graceful shutdown landing in that same window made the writer + // return immediately without ever loading the now-current manifest, + // silently discarding every already-queued Append (reproduced via + // SHUTDOWN arriving <50ms after boot -- `AOF writer: cancelled while + // waiting for manifest` with zero bytes ever written to the incr + // file). Trying the load first means a manifest that exists by the + // time we get here is never missed just because a shutdown signal + // happened to land in the same instant. let manifest_wait_start = Instant::now(); const MANIFEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); let mut manifest = loop { - if cancel.is_cancelled() { - info!("AOF writer: cancelled while waiting for manifest"); - return; - } - if manifest_wait_start.elapsed() > MANIFEST_TIMEOUT { - error!( - "AOF writer: manifest not found at {} after {:?}. Writer exiting; check recovery logs.", - base_dir.display(), - MANIFEST_TIMEOUT, - ); - return; - } match AofManifest::load(&base_dir) { Ok(Some(m)) => break m, Ok(None) => { - // main.rs recovery hasn't created the manifest yet — wait. + // main.rs recovery hasn't created the manifest yet — wait, + // unless we're being torn down or have waited too long. + if cancel.is_cancelled() { + info!("AOF writer: cancelled while waiting for manifest"); + return; + } + if manifest_wait_start.elapsed() > MANIFEST_TIMEOUT { + error!( + "AOF writer: manifest not found at {} after {:?}. Writer exiting; check recovery logs.", + base_dir.display(), + MANIFEST_TIMEOUT, + ); + return; + } std::thread::sleep(std::time::Duration::from_millis(50)); } Err(e) => { @@ -994,28 +1010,33 @@ pub async fn per_shard_aof_writer_task( use tokio::io::AsyncWriteExt; // Wait for main.rs recovery to create/load the manifest. + // + // task #27 fix: load-before-cancel-check, same rationale as the + // TopLevel loop above — a manifest that now exists must not be + // missed just because a shutdown signal landed in the same instant + // (see that loop's comment for the reproduction). let manifest_wait_start = Instant::now(); const MANIFEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); let manifest = loop { - if cancel.is_cancelled() { - info!( - "AOF writer shard {}: cancelled while waiting for manifest", - shard_id - ); - return; - } - if manifest_wait_start.elapsed() > MANIFEST_TIMEOUT { - error!( - "AOF writer shard {}: manifest not found at {} after {:?}. Writer exiting.", - shard_id, - base_dir.display(), - MANIFEST_TIMEOUT, - ); - return; - } match AofManifest::load(&base_dir) { Ok(Some(m)) => break m, Ok(None) => { + if cancel.is_cancelled() { + info!( + "AOF writer shard {}: cancelled while waiting for manifest", + shard_id + ); + return; + } + if manifest_wait_start.elapsed() > MANIFEST_TIMEOUT { + error!( + "AOF writer shard {}: manifest not found at {} after {:?}. Writer exiting.", + shard_id, + base_dir.display(), + MANIFEST_TIMEOUT, + ); + return; + } tokio::time::sleep(std::time::Duration::from_millis(50)).await; } Err(e) => { @@ -1385,28 +1406,30 @@ pub async fn per_shard_aof_writer_task( use crate::persistence::aof_manifest::{AofLayout, AofManifest}; use std::io::Write; + // task #27 fix: load-before-cancel-check, same rationale as the + // TopLevel loop above. let manifest_wait_start = Instant::now(); const MANIFEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); let manifest = loop { - if cancel.is_cancelled() { - info!( - "AOF writer shard {}: cancelled while waiting for manifest", - shard_id - ); - return; - } - if manifest_wait_start.elapsed() > MANIFEST_TIMEOUT { - error!( - "AOF writer shard {}: manifest not found at {} after {:?}. Writer exiting.", - shard_id, - base_dir.display(), - MANIFEST_TIMEOUT, - ); - return; - } match AofManifest::load(&base_dir) { Ok(Some(m)) => break m, Ok(None) => { + if cancel.is_cancelled() { + info!( + "AOF writer shard {}: cancelled while waiting for manifest", + shard_id + ); + return; + } + if manifest_wait_start.elapsed() > MANIFEST_TIMEOUT { + error!( + "AOF writer shard {}: manifest not found at {} after {:?}. Writer exiting.", + shard_id, + base_dir.display(), + MANIFEST_TIMEOUT, + ); + return; + } std::thread::sleep(std::time::Duration::from_millis(50)); } Err(e) => { diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 24be81ba..fffc58df 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -1121,6 +1121,83 @@ pub(super) fn try_handle_persistence( false } +/// Outcome of `try_handle_shutdown`. See the sharded-handler twin +/// (`handler_sharded::dispatch::ShutdownOutcome`) for the full rationale. +pub(super) enum ShutdownOutcome { + NotShutdown, + Rejected, + Exiting, +} + +/// Handle SHUTDOWN [NOSAVE|SAVE] on the monoio runtime. +/// +/// Mirrors `handler_sharded::dispatch::try_handle_shutdown`: a forced save +/// uses the cooperative per-shard BGSAVE snapshot (there is no synchronous +/// single-threaded SAVE in this mode) and polls for completion with a +/// bounded timeout via `monoio::time::sleep` (this handler's native runtime +/// -- `tokio::time::sleep` would not drive monoio's `!Send` reactor). +pub(super) async fn try_handle_shutdown( + cmd: &[u8], + cmd_args: &[Frame], + ctx: &ConnectionContext, + shutdown: &CancellationToken, + responses: &mut Vec, +) -> ShutdownOutcome { + if !cmd.eq_ignore_ascii_case(b"SHUTDOWN") { + return ShutdownOutcome::NotShutdown; + } + use crate::command::persistence::{self, ShutdownSaveMode}; + + let mode = match persistence::parse_shutdown_args(cmd_args) { + Ok(m) => m, + Err(e) => { + responses.push(e); + return ShutdownOutcome::Rejected; + } + }; + let should_save = match mode { + ShutdownSaveMode::Save => true, + ShutdownSaveMode::NoSave => false, + ShutdownSaveMode::Default => { + persistence::shutdown_default_should_save(ctx.config.save.as_deref()) + } + }; + if should_save { + match persistence::bgsave_start_sharded(&ctx.snapshot_trigger_tx, ctx.num_shards) { + Frame::Error(e) => { + responses.push(Frame::Error(e)); + return ShutdownOutcome::Rejected; + } + _ => {} + } + let start = std::time::Instant::now(); + loop { + if !persistence::SAVE_IN_PROGRESS.load(std::sync::atomic::Ordering::SeqCst) { + break; + } + if start.elapsed().as_millis() as u64 > persistence::SHUTDOWN_SAVE_TIMEOUT_MS { + responses.push(Frame::Error(Bytes::from_static( + b"ERR SHUTDOWN failed: background save timed out, check logs", + ))); + return ShutdownOutcome::Rejected; + } + monoio::time::sleep(std::time::Duration::from_millis( + persistence::SHUTDOWN_SAVE_POLL_MS, + )) + .await; + } + if !persistence::BGSAVE_LAST_STATUS.load(std::sync::atomic::Ordering::Relaxed) { + responses.push(Frame::Error(Bytes::from_static( + b"ERR SHUTDOWN failed: background save error, check logs", + ))); + return ShutdownOutcome::Rejected; + } + } + tracing::info!("SHUTDOWN command received -- initiating graceful shutdown"); + shutdown.cancel(); + ShutdownOutcome::Exiting +} + /// Handle SWAPDB — atomically exchange two databases across all shards. /// /// Validates arguments, enforces the BGREWRITEAOF guard, handles the same-index diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 236abed6..dade6ed6 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1040,6 +1040,18 @@ pub(crate) async fn handle_connection_sharded_monoio< if dispatch::try_handle_persistence(cmd, ctx, &mut responses) { continue; } + // --- SHUTDOWN [NOSAVE|SAVE] --- + match dispatch::try_handle_shutdown(cmd, cmd_args, ctx, &shutdown, &mut responses).await + { + dispatch::ShutdownOutcome::NotShutdown => {} + dispatch::ShutdownOutcome::Rejected => { + continue; + } + dispatch::ShutdownOutcome::Exiting => { + should_quit = true; + break; + } + } // ACL gate MUST run before any privileged intercept (SWAPDB included) // — otherwise unauthenticated clients can mutate cross-DB state. // handler_sharded already enforces this ordering; this matches it. diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index f939311b..57e8e9ee 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -397,6 +397,90 @@ pub(super) fn try_handle_persistence( false } +/// Outcome of `try_handle_shutdown`. +pub(super) enum ShutdownOutcome { + /// Not a SHUTDOWN command -- caller should keep dispatching. + NotShutdown, + /// SHUTDOWN was rejected (syntax error, forced SAVE failed) -- an error + /// frame was pushed onto `responses`; caller should `continue`. + Rejected, + /// SHUTDOWN succeeded and the graceful shutdown sequence has been + /// triggered -- Redis parity: no reply is sent, caller must close the + /// connection (`should_quit = true; break;`) without pushing anything + /// onto `responses`. + Exiting, +} + +/// Handle SHUTDOWN [NOSAVE|SAVE] in sharded mode. +/// +/// Sharded mode has no single-threaded synchronous SAVE path (see plain +/// SAVE's "not supported" rejection above), so a forced save here uses the +/// same cooperative per-shard BGSAVE snapshot (`bgsave_start_sharded`) and +/// polls for its completion with a bounded timeout -- a wedged shard must +/// not hang SHUTDOWN forever; timing out fails the command (server stays +/// up) rather than exiting with a torn snapshot. +pub(super) async fn try_handle_shutdown( + cmd: &[u8], + cmd_args: &[Frame], + ctx: &ConnectionContext, + shutdown: &crate::runtime::cancel::CancellationToken, + responses: &mut Vec, +) -> ShutdownOutcome { + if !cmd.eq_ignore_ascii_case(b"SHUTDOWN") { + return ShutdownOutcome::NotShutdown; + } + use crate::command::persistence::{self, ShutdownSaveMode}; + + let mode = match persistence::parse_shutdown_args(cmd_args) { + Ok(m) => m, + Err(e) => { + responses.push(e); + return ShutdownOutcome::Rejected; + } + }; + let should_save = match mode { + ShutdownSaveMode::Save => true, + ShutdownSaveMode::NoSave => false, + ShutdownSaveMode::Default => { + persistence::shutdown_default_should_save(ctx.config.save.as_deref()) + } + }; + if should_save { + match persistence::bgsave_start_sharded(&ctx.snapshot_trigger_tx, ctx.num_shards) { + Frame::Error(e) => { + responses.push(Frame::Error(e)); + return ShutdownOutcome::Rejected; + } + _ => {} + } + let start = std::time::Instant::now(); + loop { + if !persistence::SAVE_IN_PROGRESS.load(std::sync::atomic::Ordering::SeqCst) { + break; + } + if start.elapsed().as_millis() as u64 > persistence::SHUTDOWN_SAVE_TIMEOUT_MS { + responses.push(Frame::Error(Bytes::from_static( + b"ERR SHUTDOWN failed: background save timed out, check logs", + ))); + return ShutdownOutcome::Rejected; + } + tokio::time::sleep(std::time::Duration::from_millis( + persistence::SHUTDOWN_SAVE_POLL_MS, + )) + .await; + } + if !persistence::BGSAVE_LAST_STATUS.load(std::sync::atomic::Ordering::Relaxed) { + responses.push(Frame::Error(Bytes::from_static( + b"ERR SHUTDOWN failed: background save error, check logs", + ))); + return ShutdownOutcome::Rejected; + } + } + tracing::info!("SHUTDOWN command received -- initiating graceful shutdown"); + shutdown.cancel(); + ShutdownOutcome::Exiting +} + /// Handle cross-shard KEYS, SCAN, DBSIZE aggregation. /// Returns `true` if consumed. pub(super) async fn try_handle_cross_shard_scan( diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 03e63d36..b6c44dcf 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -1075,6 +1075,13 @@ pub(crate) async fn handle_connection_sharded_inner< continue; } + // --- SHUTDOWN [NOSAVE|SAVE] --- + match dispatch::try_handle_shutdown(cmd, cmd_args, ctx, &shutdown, &mut responses).await { + dispatch::ShutdownOutcome::NotShutdown => {} + dispatch::ShutdownOutcome::Rejected => { continue; } + dispatch::ShutdownOutcome::Exiting => { should_quit = true; break; } + } + // --- SWAPDB: handler-layer intercept (needs async + multi-db access) --- if dispatch::try_handle_swapdb(cmd, cmd_args, &conn, ctx, &mut responses).await { continue; diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 1b24da54..e871c5d0 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -708,6 +708,46 @@ pub async fn handle_connection( responses.push(crate::command::persistence::handle_lastsave()); continue; } + // SHUTDOWN [NOSAVE|SAVE] -- Redis parity: on success no reply + // is sent (the client observes the connection close as the + // server exits); on failure (bad syntax, or a forced SAVE + // that failed) an error is returned and the server stays up. + if cmd.eq_ignore_ascii_case(b"SHUTDOWN") { + match crate::command::persistence::parse_shutdown_args(cmd_args) { + Ok(mode) => { + use crate::command::persistence::ShutdownSaveMode; + let should_save = match mode { + ShutdownSaveMode::Save => true, + ShutdownSaveMode::NoSave => false, + ShutdownSaveMode::Default => { + crate::command::persistence::shutdown_default_should_save( + config.save.as_deref(), + ) + } + }; + if should_save + && let err @ Frame::Error(_) = crate::command::persistence::handle_save( + &db, + &config.dir, + &config.dbfilename, + ) + { + responses.push(err); + continue; + } + tracing::info!( + "SHUTDOWN command received -- initiating graceful shutdown" + ); + shutdown.cancel(); + should_quit = true; + break; + } + Err(e) => { + responses.push(e); + continue; + } + } + } // BGREWRITEAOF if cmd.eq_ignore_ascii_case(b"BGREWRITEAOF") { let response = if let Some(ref pool) = aof_pool { diff --git a/tests/shutdown_integration.rs b/tests/shutdown_integration.rs new file mode 100644 index 00000000..aded111a --- /dev/null +++ b/tests/shutdown_integration.rs @@ -0,0 +1,323 @@ +//! Task #27: SHUTDOWN [NOSAVE|SAVE] real-server integration tests. +//! +//! Covers the Redis-parity contract this command must honour end to end: +//! - `SHUTDOWN NOSAVE` exits the process promptly (no reply is sent; the +//! client observes the connection close). +//! - Under `--appendonly yes`, SHUTDOWN flushes durably -- a write made +//! just before SHUTDOWN survives a restart with no kill-9 tail loss. +//! - A forced `SHUTDOWN SAVE` that cannot complete (disk write failure) +//! replies with an error and the server stays up and reachable. +//! - Malformed arguments (`SHUTDOWN BOGUS`, conflicting modifiers) reply +//! `ERR syntax error` and the server stays up. +//! +//! Run with (release binary required): +//! cargo build --release +//! cargo test --release --test shutdown_integration -- --ignored --test-threads=1 + +#![allow(clippy::unwrap_used)] + +mod common; + +use std::io::{Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +use common::{find_moon_binary, sigkill, spawn_listening}; + +fn spawn_moon(dir: &std::path::Path, extra: &[&str]) -> (Child, u16) { + static GEN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let log_gen = GEN.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + spawn_listening(|port| { + let mut args: Vec = vec![ + "--port".into(), + port.to_string(), + "--dir".into(), + dir.to_string_lossy().into_owned(), + "--shards".into(), + "1".into(), + "--disk-free-min-pct".into(), + "0".into(), + ]; + for &e in extra { + args.push(e.into()); + } + Command::new(find_moon_binary()) + .args(&args) + .stdout( + std::fs::File::create(dir.join(format!("moon.stdout.{log_gen}.log"))) + .expect("create stdout log"), + ) + .stderr( + std::fs::File::create(dir.join(format!("moon.stderr.{log_gen}.log"))) + .expect("create stderr log"), + ) + .env("RUST_LOG", "moon=info") + .spawn() + .unwrap_or_else(|e| { + panic!( + "Failed to spawn moon binary at '{}': {e}. Build with \ + `cargo build --release` or set MOON_BIN.", + find_moon_binary().display() + ) + }) + }) +} + +/// RAII guard: SIGKILLs the server process when dropped, so a panicking +/// assertion never leaks a live server (which then poisons the next test's +/// port scan / holds the AOF dir open). +struct ServerGuard(Option); + +impl ServerGuard { + fn take(&mut self) -> Child { + self.0.take().expect("server already taken") + } +} + +impl Drop for ServerGuard { + fn drop(&mut self) { + if let Some(mut child) = self.0.take() { + sigkill(&mut child); + } + } +} + +fn connect(port: u16, deadline: Duration) -> TcpStream { + let addr = format!("127.0.0.1:{port}") + .to_socket_addrs() + .expect("parse addr") + .next() + .expect("one addr"); + let start = Instant::now(); + loop { + match TcpStream::connect_timeout(&addr, Duration::from_millis(200)) { + Ok(s) => { + // 20s: comfortably above SHUTDOWN_SAVE_TIMEOUT_MS (10s, see + // src/command/persistence.rs) so the forced-SAVE-failure test + // observes the server's own timeout error instead of racing + // its own socket read timeout. + s.set_read_timeout(Some(Duration::from_secs(20))).ok(); + s.set_write_timeout(Some(Duration::from_secs(20))).ok(); + return s; + } + Err(_) if start.elapsed() < deadline => { + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => panic!("server never accepted on port {port}: {e}"), + } + } +} + +/// Minimal inline-protocol client: enough for SET/GET/PING/SHUTDOWN. +struct Conn { + s: TcpStream, +} + +impl Conn { + fn open(port: u16) -> Self { + Self { + s: connect(port, Duration::from_secs(10)), + } + } + + /// Send a RESP array command and return the raw reply bytes read in one + /// `read()` call. Good enough for the simple status/error/bulk replies + /// these tests assert on (no pipelining, no partial-frame reassembly). + fn cmd(&mut self, parts: &[&str]) -> std::io::Result { + let mut req = Vec::with_capacity(64); + req.extend_from_slice(format!("*{}\r\n", parts.len()).as_bytes()); + for p in parts { + req.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + req.extend_from_slice(p.as_bytes()); + req.extend_from_slice(b"\r\n"); + } + self.s.write_all(&req)?; + let mut buf = [0u8; 4096]; + let n = self.s.read(&mut buf)?; + Ok(String::from_utf8_lossy(&buf[..n]).into_owned()) + } +} + +fn wait_ready(port: u16) { + let start = Instant::now(); + loop { + if let Ok(stream) = TcpStream::connect_timeout( + &std::net::SocketAddr::from(([127, 0, 0, 1], port)), + Duration::from_millis(200), + ) { + stream.set_read_timeout(Some(Duration::from_secs(2))).ok(); + stream.set_write_timeout(Some(Duration::from_secs(2))).ok(); + let mut c = Conn { s: stream }; + if let Ok(reply) = c.cmd(&["PING"]) { + if reply.contains("PONG") { + return; + } + } + } + assert!( + start.elapsed() < Duration::from_secs(30), + "server never answered PING on port {port}" + ); + std::thread::sleep(Duration::from_millis(100)); + } +} + +/// Poll until the child is no longer alive, or panic past `deadline`. +fn wait_exited(child: &mut Child, deadline: Duration) -> std::process::ExitStatus { + let start = Instant::now(); + loop { + if let Some(status) = child.try_wait().expect("try_wait") { + return status; + } + assert!( + start.elapsed() < deadline, + "server did not exit within {deadline:?} after SHUTDOWN" + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +// --------------------------------------------------------------------------- + +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn shutdown_nosave_exits_promptly() { + let dir = tempfile::tempdir().expect("tempdir"); + let (child, port) = spawn_moon(dir.path(), &["--appendonly", "no"]); + let mut guard = ServerGuard(Some(child)); + wait_ready(port); + + let mut c = Conn::open(port); + // Redis parity: on success SHUTDOWN sends no reply; the client just sees + // the connection close (a zero-length read, or a write/read error -- + // either is acceptable here, we only care that the process itself + // actually terminates below). + let _ = c.cmd(&["SHUTDOWN", "NOSAVE"]); + + let mut child = guard.take(); + let status = wait_exited(&mut child, Duration::from_secs(10)); + assert!( + status.success(), + "SHUTDOWN NOSAVE should exit cleanly (status 0), got {status:?}" + ); +} + +#[test] +#[ignore] +fn shutdown_appendonly_flushes_durably() { + let dir = tempfile::tempdir().expect("tempdir"); + let (child, port) = spawn_moon(dir.path(), &["--appendonly", "yes"]); + let mut guard = ServerGuard(Some(child)); + wait_ready(port); + + let mut c = Conn::open(port); + let set_reply = c.cmd(&["SET", "shutdown:durable", "hello"]).unwrap(); + assert!( + set_reply.contains("OK"), + "SET should succeed before SHUTDOWN: {set_reply}" + ); + let _ = c.cmd(&["SHUTDOWN", "NOSAVE"]); + + let mut child = guard.take(); + let status = wait_exited(&mut child, Duration::from_secs(10)); + assert!(status.success(), "SHUTDOWN should exit cleanly: {status:?}"); + + // Restart on the same --dir and confirm the AOF replayed the write -- + // a clean SHUTDOWN must not lose the tail the way a bare kill-9 can. + let (child2, port2) = spawn_moon(dir.path(), &["--appendonly", "yes"]); + let guard2 = ServerGuard(Some(child2)); + wait_ready(port2); + let mut c2 = Conn::open(port2); + let get_reply = c2.cmd(&["GET", "shutdown:durable"]).unwrap(); + assert!( + get_reply.contains("hello"), + "SHUTDOWN must flush AOF durably before exiting; got: {get_reply}" + ); + drop(guard2); +} + +#[test] +#[ignore] +fn shutdown_save_failure_keeps_server_up() { + let dir = tempfile::tempdir().expect("tempdir"); + let (child, port) = spawn_moon(dir.path(), &["--appendonly", "no"]); + let mut guard = ServerGuard(Some(child)); + wait_ready(port); + + // Make the data dir unwritable so the forced RDB save inside + // `SHUTDOWN SAVE` fails -- Redis parity: a save that fails must reply + // an error and NOT exit the process. + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o500)) + .expect("chmod read-only"); + } + + let mut c = Conn::open(port); + let reply = c.cmd(&["SHUTDOWN", "SAVE"]).unwrap(); + assert!( + reply.starts_with('-') || reply.to_ascii_uppercase().contains("ERR"), + "SHUTDOWN SAVE against an unwritable dir must reply an error, got: {reply}" + ); + + // Restore permissions before the guard's Drop tries to SIGKILL + the + // tempdir cleanup tries to remove files under it. + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700)) + .expect("chmod restore"); + } + + // Server must still be up and answering. + let pong = c.cmd(&["PING"]).unwrap(); + assert!( + pong.contains("PONG"), + "server must stay up after a failed SHUTDOWN SAVE, got: {pong}" + ); + + let mut child = guard.take(); + assert!( + child.try_wait().expect("try_wait").is_none(), + "server process must still be running after a failed SHUTDOWN SAVE" + ); + sigkill(&mut child); +} + +#[test] +#[ignore] +fn shutdown_syntax_error_keeps_server_up() { + let dir = tempfile::tempdir().expect("tempdir"); + let (child, port) = spawn_moon(dir.path(), &["--appendonly", "no"]); + let mut guard = ServerGuard(Some(child)); + wait_ready(port); + + let mut c = Conn::open(port); + let reply = c.cmd(&["SHUTDOWN", "BOGUS"]).unwrap(); + assert!( + reply.to_ascii_uppercase().contains("SYNTAX"), + "SHUTDOWN with an unknown modifier should be a syntax error, got: {reply}" + ); + + let reply2 = c.cmd(&["SHUTDOWN", "SAVE", "NOSAVE"]).unwrap(); + assert!( + reply2.to_ascii_uppercase().contains("SYNTAX"), + "SHUTDOWN with conflicting SAVE/NOSAVE modifiers should be a syntax \ + error, got: {reply2}" + ); + + let pong = c.cmd(&["PING"]).unwrap(); + assert!( + pong.contains("PONG"), + "server must stay up after SHUTDOWN syntax errors, got: {pong}" + ); + + let mut child = guard.take(); + assert!( + child.try_wait().expect("try_wait").is_none(), + "server process must still be running after SHUTDOWN syntax errors" + ); + sigkill(&mut child); +}