diff --git a/.add/milestones/v3-3-vector-kv-polish/MILESTONE.md b/.add/milestones/v3-3-vector-kv-polish/MILESTONE.md index a3ef0538..89d5a4c0 100644 --- a/.add/milestones/v3-3-vector-kv-polish/MILESTONE.md +++ b/.add/milestones/v3-3-vector-kv-polish/MILESTONE.md @@ -18,8 +18,13 @@ In: - FT.SEARCH avoids the ~3.2 MB `key_hash_to_key` clone per query (borrow / `Arc` the map). - KV `INCR`/`DECR` write the integer via `itoa` to a buffer — no per-op `String` alloc on the hot path. `src/command/string/string_write.rs:312,317`. -- The command-dispatch path uses `parking_lot::RwLock`, not `std::sync::RwLock`, for the ACL table. - `src/shard/event_loop.rs:65`, `src/command/connection.rs:374,460`. +- The command-dispatch path uses `parking_lot::RwLock`, not `std::sync::RwLock`, for the ACL table + (`src/shard/event_loop.rs:65`, `src/command/connection.rs:374,460`) **and for the replica-role check + that gates inline dispatch** on the p=1 read hot path (`ctx.repl_state.try_read()`, + `src/server/conn/handler_monoio/mod.rs:524` — cache as `AtomicBool` flipped on role change, or `parking_lot`). +- KV inline `GET` hit writes the value **once**: frame the `$len\r\n…\r\n` reply straight from the borrowed + `&[u8]` into `write_buf`, dropping the intermediate `val.to_vec()` heap copy. + `src/server/conn/blocking.rs:1275` (on the p=1 read hot path; the wasted copy scales with value size). Out: - Re-quantization / new vector codecs — only the existing SQ8 decode length is in scope, not new @@ -51,8 +56,10 @@ Out: FT.SEARCH (borrow / `Arc`). - [ ] kv-incr-itoa depends-on: none — `INCR`/`DECR` via `itoa`-to-buffer; no `String` alloc on the hot path. -- [ ] kv-dispatch-lock-discipline depends-on: none — ACL / dispatch `std::sync::RwLock` -> - `parking_lot::RwLock`. +- [ ] kv-dispatch-lock-discipline depends-on: none — ACL / dispatch + inline-gate replica check + `std::sync::RwLock` -> `parking_lot::RwLock` / `AtomicBool` (incl. `handler_monoio/mod.rs:524`). +- [ ] kv-inline-get-nocopy depends-on: none — inline `GET` hit writes value straight from the + borrow into `write_buf`; drop the `val.to_vec()` double-copy (`blocking.rs:1275`, scales w/ value size). ## Exit criteria (observable; map each to the task that delivers it) - [ ] An SQ8 immutable segment decodes vectors at the correct length — recall parity with the @@ -61,5 +68,7 @@ Out: - [ ] FT.SEARCH performs no per-query 3 MB `key_hash` clone — allocation probe / throughput test shows the clone gone. (← vector-search-keyhash-noclone) - [ ] `INCR`/`DECR` allocate no `String` on the hot path — `itoa` path asserted (test / no-alloc check). (← kv-incr-itoa) -- [ ] No `std::sync::RwLock` remains on the command-dispatch path — ACL table is `parking_lot`; - audit/grep + test. (← kv-dispatch-lock-discipline) +- [ ] No `std::sync::RwLock` remains on the command-dispatch path — ACL table + inline-gate replica + check are `parking_lot`/atomic; audit/grep + test. (← kv-dispatch-lock-discipline) +- [ ] Inline `GET` performs no intermediate `Vec` copy of the value — the `$len…` reply is framed from + the borrowed slice; allocation probe / large-value throughput shows the copy gone. (← kv-inline-get-nocopy) diff --git a/.add/milestones/v3-4-kv-correctness/MILESTONE.md b/.add/milestones/v3-4-kv-correctness/MILESTONE.md new file mode 100644 index 00000000..82f3067b --- /dev/null +++ b/.add/milestones/v3-4-kv-correctness/MILESTONE.md @@ -0,0 +1,147 @@ +# MILESTONE: KV Write Correctness & Data-Integrity Parity + +goal: Moon's KV write commands honor Redis data-integrity contracts: no wrong-type command destroys data, integer commands reject overflow instead of panicking or wrapping, expire-in-the-past deletes the key, and MSETNX is atomic (cross-shard spans rejected). +rationale: new-major (split 3/3) — the KV-correctness sibling of the v3 "secondary-engine correctness & parity" theme. v3-1 (FTS) and v3-2 (graph) closed the search/graph defects; the 2026-07-02 KV deep review (`tmp/KV-DEEP-REVIEW.md`) surfaced a cluster of P0 data-integrity bugs in the *primary* KV engine (silent data loss on wrong-type GETDEL/GETSET/SET..GET, i64 overflow panics/wraps in DECRBY + the expire family, and a missing MSETNX). Correctness only — the p=1 throughput question from the same review is platform-driven (confound closed) and is NOT in scope here. +stage: production · status: active · created: 2026-07-02 + +> SDD living doc for this milestone. Keep it THIN: breadth, shared decisions, and +> exit criteria only — per-task detail lives in each `.add/tasks//TASK.md`, +> written just-in-time. Update this doc whenever a task reveals a milestone gap. + +## Scope +In: +- **Wrong-type data loss (P0):** GETDEL, GETSET, and `SET key val GET` must return WRONGTYPE and + leave the key intact when it holds a non-string — never delete/overwrite it. (`check-before-mutate` + parity with Redis.) `src/command/string/string_read.rs` (GETDEL), `string_write.rs` (SET..GET, GETSET). +- **Integer overflow (P0):** `DECRBY key i64::MIN` must not panic on `checked_neg`; the expire family + (`SET EX/EXAT`, `SETEX`, `EXPIRE`/`EXPIREAT`/`PEXPIRE`) must reject seconds/ms that overflow + `i64 * 1000 + now` with an "invalid expire time" error instead of wrapping to a bogus TTL. +- **Expire-in-the-past semantics (P0):** `EXPIRE`/`PEXPIRE`/`EXPIREAT` with a non-positive / already-past + time DELETES the key and returns 1 (Redis parity), and that deletion PROPAGATES through command-based + WAL replay to replicas (`DispatchReplayEngine::replay_command` re-dispatches the raw command). +- **MSETNX (P0, missing command):** add MSETNX as an atomic multi-key write (set all iff none exist). + Single-shard: two-phase check-then-set with no await (atomic). Multi-shard: the coordinator REJECTS a + cross-shard span with CROSSSLOT (by design — no 2PC), runs co-located keys atomically on the owner. + +Out: +- **P1 semantics polish (future milestone):** SET-option mutual-exclusion validation (EX+PX, NX+XX), + SCAN delete-stability under rehash, TOUCH access-time bump, stricter integer parse (leading `+`, + whitespace), TTL rounding parity. Advisor-endorsed P0/P1 split — these are correctness *refinements*, + not data-loss/crash bugs. +- **p=1 throughput vs Redis:** the KV deep review CLOSED this as platform/environment-driven (same + build wins p=1 on OrbStack, loses on GCloud shared-vCPU); not a KV-code defect, not in scope. +- **Cross-shard MSETNX atomicity via 2PC:** explicitly rejected (CROSSSLOT) rather than half-built. + +## Shared decisions & glossary deltas (living — every task must honor these) +- **check-before-mutate:** any command that could destroy data on a type mismatch must peek the entry + type and return WRONGTYPE BEFORE the mutation — never remove/overwrite first. +- **TDD red/green (CLAUDE.md Rule 3):** each defect lands a FAILING test first, then the fix. The + cross-shard MSETNX reject was red/green-proven by neutralizing only the CROSSSLOT branch. +- **Overflow is an error, not a panic/wrap:** integer/time arithmetic on the command path uses + `checked_*` and returns a `Frame::Error`, never `unwrap()`/silent wrap (CLAUDE.md error-handling). +- **Replay consistency:** semantic changes to write commands must be verified to propagate through + command-based WAL replay (a replay test), so master and replica converge. +- **New commands carry script coverage:** MSETNX added to `scripts/test-consistency.sh` (hash-tagged + for 1/4/12-shard parity with Redis) + `scripts/test-commands.sh` (CLAUDE.md New Commands rule). + +## Shared / risky contracts (freeze these first) +- **MSETNX cross-shard disposition** — reject (CROSSSLOT) vs best-effort scatter vs 2PC. Chosen: REJECT + (user decision). A wrong choice here re-does the coordinator + the command's whole test surface. + -> owning task `kv-msetnx-atomic` +- **Expire-past delete + replay** — deleting on past-time must replay identically or replicas diverge. + -> owning task `kv-expire-past-deletes` + +## Tasks (breadth-first decomposition; detail lives in each TASK.md) +- [x] kv-wrongtype-guard depends-on: none — GETDEL / GETSET / SET..GET check type + before mutate; wrong-type returns WRONGTYPE and preserves the key (no silent data loss). +- [x] kv-integer-overflow-guard depends-on: none — DECRBY i64::MIN + the expire family + (SET EX/EXAT, SETEX, EXPIRE/EXPIREAT/PEXPIRE) reject overflow with an error instead of panic/wrap. +- [x] kv-expire-past-deletes depends-on: none — EXPIRE/PEXPIRE/EXPIREAT with a past/ + non-positive time deletes the key (returns 1); verified to propagate through WAL replay. +- [x] kv-msetnx-atomic depends-on: none — add MSETNX; atomic on one shard, + CROSSSLOT-reject across shards, co-located keys atomic on the owner. + +## Exit criteria (observable; map each to the task that delivers it) +- [x] `GETDEL`/`GETSET`/`SET k v GET` on a key holding a list/hash returns WRONGTYPE and the key still + exists afterward (no data loss) — unit tests assert key preserved. (← kv-wrongtype-guard) +- [x] `DECRBY k -9223372036854775808` and `SETEX`/`SET … EX ` return an error, not a panic or a + wrapped TTL — unit tests for min-overflow + expire overflow. (← kv-integer-overflow-guard) +- [x] `EXPIRE k -1` (and past `EXPIREAT`) deletes k and returns 1; a WAL-replay test proves the delete + re-applies on replay (master/replica consistent). (← kv-expire-past-deletes) +- [x] Co-located `MSETNX` is atomic (all-new→1, any-exists→0 with no partial write); a cross-shard + `MSETNX` returns CROSSSLOT and writes nothing — unit + `--shards 4` integration test. (← kv-msetnx-atomic) + +## Close — ship review (AI fills when every task is done — the evidence behind the engine gate, read before the boxes are checked) +> Whole-milestone, cross-task review the AI fills in. It is the evidence behind the EXISTING engine +> gate (milestone-done / checking the Exit-criteria boxes) — NOT a new approval. Tool-agnostic. + +### Ship by domain (what changed, per bounded context) +- command : `src/command/string/{string_read,string_write,mod}.rs` (GETDEL/GETSET/SET..GET/DECRBY/ + SETEX/SET-EX guards + `msetnx` handler + unit tests), `src/command/key.rs` (expire-past delete + + overflow guards + tests), `src/command/metadata.rs` (MSETNX phf entry), `src/command/mod.rs` + ((6,'m') MSETNX dispatch). +- shard : `src/shard/coordinator.rs` (`coordinate_msetnx` — CROSSSLOT reject / owner-atomic), + `src/server/conn/shared.rs` (`is_multi_key_command` MSETNX arm). +- persistence : `src/persistence/replay.rs` (expire-past-delete replay propagation tests). +- scripts : `scripts/test-consistency.sh` + `scripts/test-commands.sh` (MSETNX entries). +- tests : `tests/msetnx_cross_shard_reject.rs` (new `--shards 4` regression suite). + +### Cross-task evidence (one row per task) +- kv-wrongtype-guard : gate=PASS · commits 4a2245b + 5044487 · tests=3 unit (key-preserved) green +- kv-integer-overflow-guard : gate=PASS · commits 56008df + 9a915d3 + d6b4136 (i64-domain bound — review Finding 2/3) · + tests=3 unit (min-overflow, SETEX, SET EX) + 9 unit (i64-bound reject + extreme-negative preserve) green +- kv-expire-past-deletes : gate=PASS · commit 56008df · tests=unit + 2 replay-propagation green +- kv-msetnx-atomic : gate=PASS · commit 3b2c7dc (atomicity) + coordinator local-leg AOF durability fix + (Finding 1) · tests=3 unit + 2 integration (--shards 4, red/green-proven) + 3 crash-recovery + (coordinator_local_leg_durability, red/green on monoio+tokio) green + +### Adversarial code review (whole-diff, senior-rust-engineer agent — **SHIP** verdict) +The five milestone commits deliver their stated fixes with no new regressions or reachable +panics. Three findings surfaced (all independently re-verified against source before acting): +- **Finding 2/3 (MEDIUM/LOW) — FIXED in `d6b4136`.** The overflow guards bounded against + `u64::MAX`, not Redis's effective `i64::MAX`; a huge-but-sub-`u64` TTL (e.g. `EXPIRE k + 15000000000000000`) was accepted and then read back as a **negative** `PTTL`/`PEXPIRETIME` + on a live key, and an extreme-negative `EXPIRE` deleted instead of erroring. Now bounded to + the i64 domain at all eight expiry setters via a shared `expiry_ms_in_range` helper (red/green, + 9 tests). This makes the "not a wrapped TTL" exit criterion hold on the READ path too. +- **Finding 1 (HIGH) — PRE-EXISTING, FIXED for MSET/MSETNX (this milestone).** The cross-shard + coordinator's LOCAL leg (`coordinate_mset` fast-path + scatter local slice; `coordinate_msetnx` + local branch) executed co-located multi-key writes in memory but never appended them to the owning + shard's AOF — while the REMOTE leg (`MultiExecute` in `spsc_handler`) does (`wal_append_and_fanout`). + So a co-located `MSET`/`MSETNX` was durable when the owner was a *remote* shard but silently + **non-durable** when the owner was the connection's *own* shard. **Blast radius:** multi-shard + (`--shards >1`) + appendonly + keys hashing to the connection's own shard; single-shard (the + recommended default) was unaffected (the coordinator is bypassed for `num_shards<=1` and normal + dispatch persists). **Not introduced by MSETNX** (it copied the existing MSET coordinator pattern; + MSETNX's own exit criterion is *atomicity*, delivered — not durability). + **Fix:** the local leg now persists to the owning shard's AOF via a new `persist_local_leg` helper — + the same `AofWriterPool::issue_append_lsn` + `try_send_append_durable` path **every local single-key + write already uses** (matching the local-write contract, NOT the SPSC remote path; `ChannelMesh` has + no self-send slot, so a local leg cannot route through `wal_append_and_fanout`). Granularity: the + **whole command** for a co-located owner (MSETNX; MSET fast path), and a **synthesized MSET over only + the local keys** for a scattered MSET's local slice (never the full scattered command — `my_shard` + does not own the remote keys, and replay re-dispatches raw commands). On AOF failure the leg returns + `AOF_FSYNC_ERR` instead of a false `+OK` (design-for-failure). Red/green crash-recovery TDD in + `tests/coordinator_local_leg_durability.rs` (`--shards 4 --appendonly yes`; one co-located group per + shard so exactly one is the local leg regardless of SO_REUSEPORT landing; SIGKILL → restart → + reload): RED before (the connection's-shard group vanished), GREEN after, on **both monoio and + tokio**. The fix strictly *adds* durability and changes nothing about replication (live fan-out via + `replica_txs` is SPSC-only and untouched — not independently re-verified for local writes here). + **Remaining (tracked follow-up — same mechanism, out of this KV milestone's command scope):** BITOP / + COPY (via `run_on_owner`'s local branch) and DEL / UNLINK (via `coordinate_multi_del_or_exists`) + carry the *identical* local-leg non-durability and are NOT yet fixed; a focused follow-up should + route their local legs through `persist_local_leg` too. + +### Goal met? (map the evidence back to this milestone's Exit criteria — read before the Exit-criteria boxes are checked) +- [x] each Exit criterion above is satisfied by a Cross-task evidence row or a Ship-by-domain change (cited inline) +- goal: KV writes now honor Redis data-integrity contracts — full default lib suite 3633 green + tokio + feature set green + both clippy gates + fmt clean; no wrong-type deletes, no overflow panics/wraps, + past-expire deletes (and replays), MSETNX atomic with cross-shard reject. + +## Release steps (AI-DEFINED — fill the ordered steps to ship this milestone; engine records, human gate) +> The AI writes the release steps for THIS milestone here (hints, not engine commands). MERGE is one +> small step among them. These feed the release scope (release.md) when the cut is bundled. +- [ ] Add a CHANGELOG `[Unreleased]` entry (### Fixed — KV data-integrity; ### Added — MSETNX) — CI Lint gate. +- [ ] Open a PR from `fix/v3-4-kv-correctness` using the Close ship-review above; human reviews + merges. +- [ ] `add.py milestone-done v3-4-kv-correctness` once the Exit-criteria boxes are verified. +- [ ] Fold milestone deltas into the foundation on merge (per project convention); bundle into the next release cut. diff --git a/.add/state.json b/.add/state.json index 1c7a8617..818339a3 100644 --- a/.add/state.json +++ b/.add/state.json @@ -2,7 +2,7 @@ "project": "moon", "stage": "production", "active_task": null, - "active_milestone": null, + "active_milestone": "v3-4-kv-correctness", "tasks": { "hotpath-lock-quickwins": { "title": "Eliminate per-command global locks & syscall-level quick wins", @@ -235,10 +235,18 @@ "status": "planned", "created": "2026-06-16T04:42:42+00:00", "updated": "2026-06-16T04:42:42+00:00" + }, + "v3-4-kv-correctness": { + "title": "KV Write Correctness & Data-Integrity Parity", + "goal": "Moon's KV write commands honor Redis data-integrity contracts: no wrong-type command destroys data, integer commands reject overflow instead of panicking or wrapping, expire-in-the-past deletes the key, and MSETNX is atomic (cross-shard spans rejected).", + "stage": "production", + "status": "active", + "created": "2026-07-02T13:45:23+00:00", + "updated": "2026-07-02T13:45:23+00:00" } }, "created": "2026-06-11T03:18:21+00:00", - "updated": "2026-06-23T08:25:16+00:00", + "updated": "2026-07-02T13:45:23+00:00", "setup": { "locked": true, "locked_at": "2026-06-11T03:28:00+00:00", diff --git a/BENCHMARK.md b/BENCHMARK.md index 1b61bb0d..7df971cc 100644 --- a/BENCHMARK.md +++ b/BENCHMARK.md @@ -499,6 +499,8 @@ non-pipelined workloads; add shards only for pipelined / AOF / hash-tag-co-locat degrades with shard count (0.93×→0.61× as 1→12) as its keys scatter cross-shard — `{hash-tag}` co-location restores it. Detail: `docs/reviews/2026-06-17/WIDER-BENCH.md`. +> **Clarification (2026-07-02 KV deep review).** The **0.46–0.51×** p=1 figure above is from `bench-production.sh`, which changes three things at once *besides* shard count: distributed keys (`-r`, real cross-shard scatter — vs `bench-compare.sh`'s single hot `__rand_key__`), larger values (512B–4KB), and higher client counts (`-c 100/200` on the INCR rows). It is a **throughput artifact of {distributed keys × high concurrency × value size × multi-key scatter}, not a clean shard-count signal** — the cross-shard hop itself is ~10µs (v2-2 bare-metal), ~2% of the ~460µs p=1 baseline. The controlled `bench-compare.sh` sweep in the table above (only shard count varies) is **flat at p=1** (0.79→0.82×). Read 0.46× as "production-shaped multi-key under concurrency," not "the cross-shard hop costs 2×." + --- ## 7. Persistence (AOF) Performance diff --git a/CHANGELOG.md b/CHANGELOG.md index 680606b6..3f85e510 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,41 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +Milestone **v3-4 KV Write Correctness & Data-Integrity Parity** — the primary KV +engine now honors Redis's data-integrity contracts (no wrong-type data loss, no +integer-overflow panics/wraps, past-expire deletes, atomic MSETNX). + +### Added + +- **MSETNX** — atomic multi-key string write (set all pairs iff none of the keys exist; returns 1/0). + Single-shard atomic (two-phase check-then-set, no await between phases). The cross-shard coordinator + rejects a key span with `CROSSSLOT` (by design — no two-phase commit) and runs co-located `{hash-tag}` + keys atomically on the owning shard. New `--shards 4` regression suite `tests/msetnx_cross_shard_reject.rs`. + +### Fixed + +- **KV data-integrity (P0):** wrong-type `GETDEL` / `GETSET` / `SET … GET` no longer silently delete or + overwrite a key holding a non-string — they return `WRONGTYPE` and preserve the value (check-before-mutate). +- **Integer overflow (P0):** `DECRBY key -9223372036854775808` no longer panics; every expiry-setting + command (`SET EX/PX/EXAT`, `SETEX`, `PSETEX`, `EXPIRE`/`EXPIREAT`/`PEXPIRE`) now rejects a time whose + absolute expiry falls outside the `i64` millisecond domain with an "invalid expire time" error — + matching Redis (`when > LLONG_MAX/1000`). This closes a latent wrap where an accepted-but-huge TTL + (e.g. `EXPIRE k 15000000000000000`) surfaced as a **negative** `PTTL`/`PEXPIRETIME` on a live key, and + makes an extreme-negative `EXPIRE`/`EXPIREAT` (`< i64::MIN/1000`) error instead of deleting. +- **Expire-in-the-past semantics (P0):** `EXPIRE`/`PEXPIRE`/`EXPIREAT` with a non-positive or already-past + time now deletes the key and returns 1 (Redis parity); verified to propagate through command-based WAL + replay so replicas stay consistent. +- **Cross-shard coordinator local-leg durability (P0, pre-existing):** a co-located `MSET`/`MSETNX` whose + keys hash to the **connection's own shard** now appends to that shard's AOF. The coordinator's local + leg previously executed the write in memory but never persisted it (the remote `MultiExecute` leg + always did), so with `--shards >1 --appendonly yes` an own-shard co-located `MSET`/`MSETNX` could be + lost on crash. It now persists via the same append path every local single-key write uses — the whole + command for a co-located owner, and a synthesized `MSET` over **only the local keys** for a scattered + `MSET`'s local slice — returning an AOF error instead of a false `+OK` on append failure. + Crash-recovery verified on both monoio and tokio (`tests/coordinator_local_leg_durability.rs`). + Single-shard (the default) was never affected. The same local-leg gap remains for coordinator + `BITOP` / `COPY` / `DEL` / `UNLINK` (tracked follow-up). + ## [0.4.1] — 2026-06-23 Measure-only validation release — **no server behavior change**. Bundles the closed diff --git a/CLAUDE.md b/CLAUDE.md index 7a96d621..b84868af 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -140,7 +140,7 @@ orb run -m moon-dev bash -c 'sudo apt-get update -qq && sudo apt-get install -y - Never hold a lock across `.await` points. - Replace `.read().unwrap()` / `.write().unwrap()` with `.read()` / `.write()` (parking_lot doesn't poison). - Per-shard locks only — no global locks on the write path. -- **monoio cross-thread wakers:** `monoio::spawn` creates `!Send` tasks; `Waker::wake()` from another OS thread does NOT reach them. Use the `pending_wakers: Rc>>` relay — connection handler registers its waker, event loop drains and wakes locally after SPSC processing. For cross-thread signalling, prefer `flume::bounded(1)` over custom atomic oneshots. +- **monoio cross-thread wakers:** `monoio::spawn` creates `!Send` tasks; `Waker::wake()` from another OS thread does NOT reach them. The cross-shard **reply** path therefore has the connection **await a `flume` oneshot directly** (the target shard sends the reply on it after executing). A `pending_wakers: Rc>>` relay is still swept each event-loop iteration (pub/sub + backpressure paths thread it), but it is **no longer the reply-wake mechanism** — the old "register your waker, event loop drains it after SPSC" design was retired when test `swf0` disproved its premise (see `event_loop.rs` ~1730). For cross-thread signalling, prefer `flume::bounded(1)` over custom atomic oneshots. ### Error Handling - All command errors return `Frame::Error(Bytes)` — no `Result` types in dispatch paths. diff --git a/scripts/test-commands.sh b/scripts/test-commands.sh index 2b865fcf..e8297aff 100755 --- a/scripts/test-commands.sh +++ b/scripts/test-commands.sh @@ -344,6 +344,8 @@ if should_run "string"; then assert_moon "INCRBYFLOAT" "6.5" INCRBYFLOAT str:cnt1 0.5 assert_moon "MSET" "OK" MSET str:m1 a str:m2 b str:m3 c assert_moon_ok "MGET" MGET str:m1 str:m2 str:m3 + assert_moon "MSETNX (new)" "(integer) 1" MSETNX "{mcm}n1" a "{mcm}n2" b + assert_moon "MSETNX (exists)" "(integer) 0" MSETNX "{mcm}n2" x "{mcm}n3" c assert_moon_ok "GETEX with EX" GETEX str:m1 EX 100 else assert_match "SET basic" SET str:k1 hello @@ -369,6 +371,8 @@ if should_run "string"; then assert_match "INCRBYFLOAT" INCRBYFLOAT str:cnt1 0.5 assert_match "MSET" MSET str:m1 a str:m2 b str:m3 c assert_match "MGET" MGET str:m1 str:m2 str:m3 + assert_match "MSETNX (new)" MSETNX "{mcc}n1" a "{mcc}n2" b + assert_match "MSETNX (exists)" MSETNX "{mcc}n2" x "{mcc}n3" c assert_match "GETEX with EX" GETEX str:m1 EX 100 fi fi diff --git a/scripts/test-consistency.sh b/scripts/test-consistency.sh index 67bf6613..aca83d5b 100755 --- a/scripts/test-consistency.sh +++ b/scripts/test-consistency.sh @@ -247,6 +247,13 @@ both MSET mk1 "val1" mk2 "val2" mk3 "val3" assert_both "MGET 3 keys" MGET mk1 mk2 mk3 assert_both "MGET with missing" MGET mk1 nonexistent mk3 +# MSETNX: hash-tagged ({mn}) so all keys co-locate on one shard -> atomic under +# Moon's 1/4/12 shard configs (cross-shard MSETNX is rejected CROSSSLOT by design). +assert_both "MSETNX all new" MSETNX "{mn}k1" "v1" "{mn}k2" "v2" +assert_both "MGET after MSETNX" MGET "{mn}k1" "{mn}k2" +assert_both "MSETNX one exists (0)" MSETNX "{mn}k2" "new2" "{mn}k3" "v3" +assert_both "MSETNX no partial write" GET "{mn}k3" + # =========================================================================== # 4. SET with options (EX, PX, NX, XX, KEEPTTL, GET) # =========================================================================== diff --git a/src/command/helpers.rs b/src/command/helpers.rs index f55f93dd..84392836 100644 --- a/src/command/helpers.rs +++ b/src/command/helpers.rs @@ -27,3 +27,16 @@ pub fn ok() -> Frame { pub fn err(msg: &str) -> Frame { Frame::Error(Bytes::from(msg.to_string())) } + +/// Whether an absolute expiry (unix millis) is representable without a +/// client-visible wrap. +/// +/// Expiry is stored as `u64` millis, but `PTTL`/`PEXPIRETIME` cast it back to +/// `i64` — an expiry past `i64::MAX` surfaces as a NEGATIVE TTL on a key that is +/// very much alive. Redis rejects such out-of-range expiries outright +/// (`when > LLONG_MAX / 1000` → "invalid expire time"), so every expiry-setting +/// command must too. Returns `true` when `expires_at_ms` is safe to store. +#[inline] +pub fn expiry_ms_in_range(expires_at_ms: u64) -> bool { + expires_at_ms <= i64::MAX as u64 +} diff --git a/src/command/key.rs b/src/command/key.rs index 6c1b22bc..685da0ca 100644 --- a/src/command/key.rs +++ b/src/command/key.rs @@ -6,7 +6,7 @@ use crate::storage::Database; use crate::storage::compact_key::CompactKey; use crate::storage::entry::current_time_ms; -use super::helpers::err_wrong_args; +use super::helpers::{err_wrong_args, expiry_ms_in_range}; /// Extract a key as &[u8] from a Frame argument. pub(crate) fn extract_key(frame: &Frame) -> Option<&[u8]> { @@ -64,8 +64,9 @@ pub fn exists(db: &mut Database, args: &[Frame]) -> Frame { /// EXPIRE key seconds /// -/// Set a timeout on key. Returns 1 if timeout was set, 0 if key does not exist. -/// Negative or zero seconds returns an error (modern Redis 7+ behavior). +/// Set a timeout on key. Returns 1 if the timeout was set (or the key was +/// deleted because of a non-positive/past TTL), 0 if the key does not exist. +/// A non-positive TTL deletes the key immediately (Redis past-time semantics). pub fn expire(db: &mut Database, args: &[Frame]) -> Frame { if args.len() != 2 { return err_wrong_args("EXPIRE"); @@ -82,12 +83,37 @@ pub fn expire(db: &mut Database, args: &[Frame]) -> Frame { )); } }; - if seconds <= 0 { + // Redis rejects an out-of-i64-range expiry (`seconds < LLONG_MIN/1000`) BEFORE + // the past-time delete, so an extreme negative errors rather than deleting. + if seconds < i64::MIN / 1000 { return Frame::Error(Bytes::from_static( b"ERR invalid expire time in 'EXPIRE' command", )); } - let expires_at_ms = current_time_ms() + (seconds as u64) * 1000; + // Redis parity: a non-positive TTL is a past-time expiry -> delete the key now + // (return 1 if it existed, 0 otherwise) rather than erroring. Mirrors EXPIREAT. + if seconds <= 0 { + return if db.remove(key).is_some() { + Frame::Integer(1) + } else { + Frame::Integer(0) + }; + } + // Guard the u64 arithmetic (seconds*1000 + now_ms can overflow) AND bound the + // result to the i64 domain so PTTL — which casts the stored u64 back to i64 — + // never wraps negative on a live key. Redis rejects an out-of-range expiry. + let expires_at_ms = match (seconds as u64) + .checked_mul(1000) + .and_then(|delta| current_time_ms().checked_add(delta)) + .filter(|ms| expiry_ms_in_range(*ms)) + { + Some(ms) => ms, + None => { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'EXPIRE' command", + )); + } + }; if db.set_expiry(key, expires_at_ms) { Frame::Integer(1) } else { @@ -97,7 +123,8 @@ pub fn expire(db: &mut Database, args: &[Frame]) -> Frame { /// PEXPIRE key milliseconds /// -/// Like EXPIRE but the timeout is specified in milliseconds. +/// Like EXPIRE but the timeout is specified in milliseconds. A non-positive TTL +/// deletes the key immediately (Redis past-time semantics). pub fn pexpire(db: &mut Database, args: &[Frame]) -> Frame { if args.len() != 2 { return err_wrong_args("PEXPIRE"); @@ -114,12 +141,27 @@ pub fn pexpire(db: &mut Database, args: &[Frame]) -> Frame { )); } }; + // Redis parity: a non-positive TTL is a past-time expiry -> delete the key now. if millis <= 0 { - return Frame::Error(Bytes::from_static( - b"ERR invalid expire time in 'PEXPIRE' command", - )); + return if db.remove(key).is_some() { + Frame::Integer(1) + } else { + Frame::Integer(0) + }; } - let expires_at_ms = current_time_ms() + millis as u64; + // Guard the u64 arithmetic against overflow AND bound the result to the i64 + // domain so PTTL never wraps negative on a live key (consistent with EXPIRE). + let expires_at_ms = match current_time_ms() + .checked_add(millis as u64) + .filter(|ms| expiry_ms_in_range(*ms)) + { + Some(ms) => ms, + None => { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'PEXPIRE' command", + )); + } + }; if db.set_expiry(key, expires_at_ms) { Frame::Integer(1) } else { @@ -236,6 +278,13 @@ pub fn expireat(db: &mut Database, args: &[Frame]) -> Frame { )); } }; + // Redis rejects an out-of-i64-range timestamp (`< LLONG_MIN/1000`) before the + // past-time delete, so an extreme negative errors rather than deleting. + if timestamp < i64::MIN / 1000 { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'EXPIREAT' command", + )); + } // Redis accepts 0 and negative timestamps as past-time expiry (deletes key immediately) if timestamp <= 0 { return if db.remove(key).is_some() { @@ -244,7 +293,19 @@ pub fn expireat(db: &mut Database, args: &[Frame]) -> Frame { Frame::Integer(0) }; } - let expires_at_ms = (timestamp as u64) * 1000; + // Guard the *1000 conversion against u64 overflow AND bound the result to the + // i64 domain so PEXPIRETIME never wraps negative on a live key. + let expires_at_ms = match (timestamp as u64) + .checked_mul(1000) + .filter(|ms| expiry_ms_in_range(*ms)) + { + Some(ms) => ms, + None => { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'EXPIREAT' command", + )); + } + }; if db.set_expiry(key, expires_at_ms) { Frame::Integer(1) } else { @@ -1383,10 +1444,130 @@ mod tests { } #[test] - fn test_expire_negative() { + fn test_expire_nonpositive_deletes() { + // Redis parity: EXPIRE with a non-positive TTL deletes the key immediately + // (past-time expiry) and returns 1 -- it must NOT error and leave the key. + let mut db = setup_db_with_key(b"foo", b"bar"); + assert_eq!(expire(&mut db, &[bs(b"foo"), bs(b"-1")]), Frame::Integer(1)); + assert!(!db.exists(b"foo"), "EXPIRE foo -1 must delete the key"); + + let mut db2 = setup_db_with_key(b"foo", b"bar"); + assert_eq!(expire(&mut db2, &[bs(b"foo"), bs(b"0")]), Frame::Integer(1)); + assert!(!db2.exists(b"foo"), "EXPIRE foo 0 must delete the key"); + } + + #[test] + fn test_expire_nonpositive_missing_key() { + let mut db = Database::new(); + assert_eq!( + expire(&mut db, &[bs(b"nope"), bs(b"-1")]), + Frame::Integer(0) + ); + } + + #[test] + fn test_expire_overflow_rejected() { + // now_ms + seconds*1000 overflows u64 -> error, no silent wrap; key untouched. + let mut db = setup_db_with_key(b"foo", b"bar"); + let huge = Frame::BulkString(Bytes::from(i64::MAX.to_string())); + let result = expire(&mut db, &[bs(b"foo"), huge]); + assert!( + matches!(result, Frame::Error(ref s) if s.starts_with(b"ERR invalid expire")), + "overflowing EXPIRE must error, got {result:?}" + ); + assert!( + db.exists(b"foo"), + "rejected EXPIRE must not disturb the key" + ); + } + + // --- i64-domain expiry bound (Finding 2/3): an absolute expiry past i64::MAX + // would surface as a NEGATIVE TTL on a live key, because PTTL/PEXPIRETIME cast + // the stored u64 back to i64. Redis rejects such expiries outright + // (`when > LLONG_MAX/1000` -> "invalid expire time"); Moon must too. --- + + #[test] + fn test_expire_i64_bound_rejected() { + // seconds*1000 fits u64 but now+that exceeds i64::MAX. Without the i64 bound + // Moon accepts it and PTTL wraps negative. Must error and leave the key's TTL + // untouched (no expiry set -> PTTL == -1, never a bogus large-negative). + let mut db = setup_db_with_key(b"foo", b"bar"); + let over = Frame::BulkString(Bytes::from("15000000000000000")); // 1.5e16 s + let result = expire(&mut db, &[bs(b"foo"), over]); + assert!( + matches!(result, Frame::Error(ref s) if s.starts_with(b"ERR invalid expire")), + "EXPIRE past i64::MAX must error, got {result:?}" + ); + assert!( + db.exists(b"foo"), + "rejected EXPIRE must not disturb the key" + ); + assert_eq!( + pttl(&mut db, &[bs(b"foo")]), + Frame::Integer(-1), + "rejected EXPIRE must leave the key with no expiry (PTTL -1, never wrapped-negative)" + ); + } + + #[test] + fn test_expire_extreme_negative_errors_not_deletes() { + // Redis rejects |seconds| > LLONG_MAX/1000 BEFORE the past-time delete + // (when < LLONG_MIN/1000). i64::MIN must ERROR and PRESERVE the key, unlike a + // normal small negative which deletes. + let mut db = setup_db_with_key(b"foo", b"bar"); + let min = Frame::BulkString(Bytes::from(i64::MIN.to_string())); + let result = expire(&mut db, &[bs(b"foo"), min]); + assert!( + matches!(result, Frame::Error(ref s) if s.starts_with(b"ERR invalid expire")), + "EXPIRE i64::MIN must error, got {result:?}" + ); + assert!( + db.exists(b"foo"), + "rejected extreme-negative EXPIRE must not delete the key" + ); + } + + #[test] + fn test_pexpire_i64_bound_rejected() { + // now_ms + i64::MAX ms exceeds i64::MAX -> reject (else PTTL wraps negative). let mut db = setup_db_with_key(b"foo", b"bar"); - let result = expire(&mut db, &[bs(b"foo"), bs(b"-1")]); - assert!(matches!(result, Frame::Error(ref s) if s.starts_with(b"ERR invalid expire"))); + let over = Frame::BulkString(Bytes::from(i64::MAX.to_string())); + let result = pexpire(&mut db, &[bs(b"foo"), over]); + assert!( + matches!(result, Frame::Error(ref s) if s.starts_with(b"ERR invalid expire")), + "PEXPIRE past i64::MAX must error, got {result:?}" + ); + assert!(db.exists(b"foo")); + assert_eq!(pttl(&mut db, &[bs(b"foo")]), Frame::Integer(-1)); + } + + #[test] + fn test_expireat_i64_bound_rejected() { + // absolute seconds*1000 exceeds i64::MAX -> reject (else PEXPIRETIME wraps negative). + let mut db = setup_db_with_key(b"foo", b"bar"); + let over = Frame::BulkString(Bytes::from("15000000000000000")); // 1.5e16 s + let result = expireat(&mut db, &[bs(b"foo"), over]); + assert!( + matches!(result, Frame::Error(ref s) if s.starts_with(b"ERR invalid expire")), + "EXPIREAT past i64::MAX must error, got {result:?}" + ); + assert!(db.exists(b"foo")); + assert_eq!(pexpiretime(&mut db, &[bs(b"foo")]), Frame::Integer(-1)); + } + + #[test] + fn test_expireat_extreme_negative_errors_not_deletes() { + let mut db = setup_db_with_key(b"foo", b"bar"); + let min = Frame::BulkString(Bytes::from(i64::MIN.to_string())); + let result = expireat(&mut db, &[bs(b"foo"), min]); + assert!( + matches!(result, Frame::Error(ref s) if s.starts_with(b"ERR invalid expire")), + "EXPIREAT i64::MIN must error, got {result:?}" + ); + assert!( + db.exists(b"foo"), + "rejected extreme-negative EXPIREAT must not delete the key" + ); } // --- PEXPIRE tests --- @@ -1403,6 +1584,17 @@ mod tests { } } + #[test] + fn test_pexpire_nonpositive_deletes() { + // Redis parity: PEXPIRE with a non-positive TTL deletes the key and returns 1. + let mut db = setup_db_with_key(b"foo", b"bar"); + assert_eq!( + pexpire(&mut db, &[bs(b"foo"), bs(b"-1")]), + Frame::Integer(1) + ); + assert!(!db.exists(b"foo"), "PEXPIRE foo -1 must delete the key"); + } + // --- TTL tests --- #[test] @@ -1881,6 +2073,22 @@ mod tests { assert!(db.get(b"k").unwrap().has_expiry()); } + #[test] + fn test_expireat_overflow_rejected() { + // (timestamp * 1000) overflows u64 for huge timestamps -> error, key untouched. + let mut db = setup_db_with_key(b"k", b"v"); + let huge = Frame::BulkString(Bytes::from(i64::MAX.to_string())); + let result = expireat(&mut db, &[bs(b"k"), huge]); + assert!( + matches!(result, Frame::Error(ref e) if e.starts_with(b"ERR invalid expire")), + "overflowing EXPIREAT must error, got {result:?}" + ); + assert!( + db.exists(b"k"), + "rejected EXPIREAT must not disturb the key" + ); + } + #[test] fn test_expireat_missing() { let mut db = Database::new(); diff --git a/src/command/metadata.rs b/src/command/metadata.rs index 935d2744..3923f108 100644 --- a/src/command/metadata.rs +++ b/src/command/metadata.rs @@ -142,6 +142,7 @@ pub static COMMAND_META: phf::Map<&'static str, CommandMeta> = phf_map! { "SET" => CommandMeta { name: "SET", arity: -3, flags: WF, first_key: 1, last_key: 1, step: 1, acl_categories: STR }, "MGET" => CommandMeta { name: "MGET", arity: -2, flags: RF, first_key: 1, last_key: -1, step: 1, acl_categories: STR }, "MSET" => CommandMeta { name: "MSET", arity: -3, flags: WF, first_key: 1, last_key: -1, step: 2, acl_categories: STR }, + "MSETNX" => CommandMeta { name: "MSETNX", arity: -3, flags: WF, first_key: 1, last_key: -1, step: 2, acl_categories: STR }, "SETNX" => CommandMeta { name: "SETNX", arity: 3, flags: WF, first_key: 1, last_key: 1, step: 1, acl_categories: STR }, "SETEX" => CommandMeta { name: "SETEX", arity: 4, flags: WF, first_key: 1, last_key: 1, step: 1, acl_categories: STR }, "PSETEX" => CommandMeta { name: "PSETEX", arity: 4, flags: WF, first_key: 1, last_key: 1, step: 1, acl_categories: STR }, diff --git a/src/command/mod.rs b/src/command/mod.rs index ecface75..8d79fb60 100644 --- a/src/command/mod.rs +++ b/src/command/mod.rs @@ -506,10 +506,13 @@ fn dispatch_inner( } } (6, b'm') => { - // MEMORY (USAGE, STATS, DOCTOR, HELP) + // MEMORY (USAGE, STATS, DOCTOR, HELP), MSETNX if cmd.eq_ignore_ascii_case(b"MEMORY") { return resp(server_admin::memory(db, args)); } + if cmd.eq_ignore_ascii_case(b"MSETNX") { + return resp(string::msetnx(db, args)); + } } (6, b'o') => { // OBJECT diff --git a/src/command/string/mod.rs b/src/command/string/mod.rs index c573f87e..97d699fc 100644 --- a/src/command/string/mod.rs +++ b/src/command/string/mod.rs @@ -266,6 +266,44 @@ mod tests { assert!(matches!(result, Frame::Error(_))); } + #[test] + fn test_msetnx_all_new() { + // All keys absent -> set all, return 1. + let mut db = make_db(); + let r = msetnx(&mut db, &[bs(b"a"), bs(b"1"), bs(b"b"), bs(b"2")]); + assert_eq!(r, Frame::Integer(1)); + assert_eq!( + get(&mut db, &[bs(b"a")]), + Frame::BulkString(Bytes::from_static(b"1")) + ); + assert_eq!( + get(&mut db, &[bs(b"b")]), + Frame::BulkString(Bytes::from_static(b"2")) + ); + } + + #[test] + fn test_msetnx_one_exists_sets_none() { + // Atomic all-or-nothing: if ANY key exists, set NOTHING, return 0. + let mut db = make_db(); + db.set_string(Bytes::from_static(b"b"), Bytes::from_static(b"old")); + let r = msetnx(&mut db, &[bs(b"a"), bs(b"1"), bs(b"b"), bs(b"2")]); + assert_eq!(r, Frame::Integer(0)); + assert_eq!(get(&mut db, &[bs(b"a")]), Frame::Null, "a must not be set"); + assert_eq!( + get(&mut db, &[bs(b"b")]), + Frame::BulkString(Bytes::from_static(b"old")), + "b must be unchanged" + ); + } + + #[test] + fn test_msetnx_odd_args() { + let mut db = make_db(); + let r = msetnx(&mut db, &[bs(b"a"), bs(b"1"), bs(b"b")]); + assert!(matches!(r, Frame::Error(_))); + } + // --- INCR/DECR tests --- #[test] @@ -323,6 +361,80 @@ mod tests { assert_eq!(result, Frame::Integer(7)); } + #[test] + fn test_decrby_min_overflow() { + // DECRBY key i64::MIN negates to +2^63, which is unrepresentable -> must + // return an error, not panic (debug) or wrap (release). + let mut db = make_db(); + db.set_string(Bytes::from_static(b"n"), Bytes::from_static(b"0")); + let arg = Frame::BulkString(Bytes::from(i64::MIN.to_string())); + let result = decrby(&mut db, &[bs(b"n"), arg]); + match result { + Frame::Error(e) => assert!(e.ends_with(b"overflow"), "got {e:?}"), + other => panic!("expected overflow error, got {other:?}"), + } + } + + #[test] + fn test_setex_overflow_rejected() { + // now_ms + seconds*1000 overflows for huge seconds -> error, key not created. + let mut db = make_db(); + let huge = Frame::BulkString(Bytes::from(i64::MAX.to_string())); + let result = setex(&mut db, &[bs(b"k"), huge, bs(b"v")]); + assert!(matches!(result, Frame::Error(ref e) if e.starts_with(b"ERR invalid expire"))); + assert!(!db.exists(b"k"), "rejected SETEX must not create the key"); + } + + #[test] + fn test_set_ex_overflow_rejected() { + let mut db = make_db(); + let huge = Frame::BulkString(Bytes::from(i64::MAX.to_string())); + let result = set(&mut db, &[bs(b"k"), bs(b"v"), bs(b"EX"), huge]); + assert!(matches!(result, Frame::Error(ref e) if e.starts_with(b"ERR invalid expire"))); + assert!(!db.exists(b"k"), "rejected SET EX must not create the key"); + } + + // --- i64-domain expiry bound (Finding 2): reject expiries past i64::MAX so + // PTTL/PEXPIRETIME never wrap negative on a live key (Redis parity). --- + + #[test] + fn test_setex_i64_bound_rejected() { + // seconds*1000 fits u64 but exceeds i64::MAX -> reject (else PTTL wraps negative). + let mut db = make_db(); + let over = Frame::BulkString(Bytes::from("15000000000000000")); // 1.5e16 s + let result = setex(&mut db, &[bs(b"k"), over, bs(b"v")]); + assert!(matches!(result, Frame::Error(ref e) if e.starts_with(b"ERR invalid expire"))); + assert!(!db.exists(b"k"), "rejected SETEX must not create the key"); + } + + #[test] + fn test_set_ex_i64_bound_rejected() { + let mut db = make_db(); + let over = Frame::BulkString(Bytes::from("15000000000000000")); + let result = set(&mut db, &[bs(b"k"), bs(b"v"), bs(b"EX"), over]); + assert!(matches!(result, Frame::Error(ref e) if e.starts_with(b"ERR invalid expire"))); + assert!(!db.exists(b"k"), "rejected SET EX must not create the key"); + } + + #[test] + fn test_set_px_i64_bound_rejected() { + // now_ms + i64::MAX ms exceeds i64::MAX -> reject. + let mut db = make_db(); + let over = Frame::BulkString(Bytes::from(i64::MAX.to_string())); + let result = set(&mut db, &[bs(b"k"), bs(b"v"), bs(b"PX"), over]); + assert!(matches!(result, Frame::Error(ref e) if e.starts_with(b"ERR invalid expire"))); + assert!(!db.exists(b"k"), "rejected SET PX must not create the key"); + } + + #[test] + fn test_psetex_i64_bound_rejected() { + let mut db = make_db(); + let over = Frame::BulkString(Bytes::from(i64::MAX.to_string())); + let result = psetex(&mut db, &[bs(b"k"), over, bs(b"v")]); + assert!(matches!(result, Frame::Error(ref e) if e.starts_with(b"ERR invalid expire"))); + assert!(!db.exists(b"k"), "rejected PSETEX must not create the key"); + } + #[test] fn test_incrby() { let mut db = make_db(); @@ -557,6 +669,42 @@ mod tests { assert!(!entry.has_expiry()); } + #[test] + fn test_getset_wrongtype_preserves_key() { + // Regression (data-loss): GETSET on a wrong-type key must return WRONGTYPE + // and MUST NOT overwrite it. Previously db.set_string ran unconditionally. + let mut db = make_db(); + db.set(Bytes::from_static(b"myhash"), Entry::new_hash()); + let result = getset(&mut db, &[bs(b"myhash"), bs(b"newval")]); + match result { + Frame::Error(e) => assert!(e.starts_with(b"WRONGTYPE")), + other => panic!("Expected WRONGTYPE error, got {other:?}"), + } + // The hash must be intact: a plain GET still reports WRONGTYPE (not the new string). + match get(&mut db, &[bs(b"myhash")]) { + Frame::Error(e) => assert!(e.starts_with(b"WRONGTYPE")), + other => panic!("GETSET must not overwrite a wrong-type key, got {other:?}"), + } + } + + #[test] + fn test_set_get_option_wrongtype_preserves_key() { + // Regression (data-loss): SET key val GET on a wrong-type key must return + // WRONGTYPE and perform NO write (precedence over NX/XX). Previously db.set + // ran unconditionally, destroying the wrong-type value. + let mut db = make_db(); + db.set(Bytes::from_static(b"myhash"), Entry::new_hash()); + let result = set(&mut db, &[bs(b"myhash"), bs(b"newval"), bs(b"GET")]); + match result { + Frame::Error(e) => assert!(e.starts_with(b"WRONGTYPE")), + other => panic!("Expected WRONGTYPE error, got {other:?}"), + } + match get(&mut db, &[bs(b"myhash")]) { + Frame::Error(e) => assert!(e.starts_with(b"WRONGTYPE")), + other => panic!("SET..GET must not overwrite a wrong-type key, got {other:?}"), + } + } + // --- GETDEL tests --- #[test] @@ -575,6 +723,24 @@ mod tests { assert_eq!(result, Frame::Null); } + #[test] + fn test_getdel_wrongtype_preserves_key() { + // Regression (data-loss): GETDEL on a wrong-type key must return WRONGTYPE + // and MUST NOT delete the key. Previously db.remove() fired before the type + // check, destroying the key and then returning WRONGTYPE as if nothing happened. + let mut db = make_db(); + db.set(Bytes::from_static(b"myhash"), Entry::new_hash()); + let result = getdel(&mut db, &[bs(b"myhash")]); + match result { + Frame::Error(e) => assert!(e.starts_with(b"WRONGTYPE")), + other => panic!("Expected WRONGTYPE error, got {other:?}"), + } + assert!( + db.exists(b"myhash"), + "GETDEL on a wrong-type key must not delete it (data-loss regression)" + ); + } + // --- GETEX tests --- #[test] diff --git a/src/command/string/string_read.rs b/src/command/string/string_read.rs index 4cc4780a..ca818782 100644 --- a/src/command/string/string_read.rs +++ b/src/command/string/string_read.rs @@ -189,12 +189,25 @@ pub fn getdel(db: &mut Database, args: &[Frame]) -> Frame { Some(k) => k, None => return err_wrong_args("GETDEL"), }; + // Check-before-mutate (Redis parity): peek the type via a cheap borrow BEFORE + // removing. Removing a wrong-type key and only then returning WRONGTYPE would + // destroy the key (data-loss). as_bytes() borrows without cloning; the borrow + // ends before db.remove() so the mutable re-borrow is sound. + match db.get(key) { + None => return Frame::Null, + Some(entry) => { + if entry.value.as_bytes().is_none() { + return Frame::Error(Bytes::from_static( + b"WRONGTYPE Operation against a key holding the wrong kind of value", + )); + } + } + } + // Confirmed string — safe to remove and return its bytes. match db.remove(key) { Some(entry) => match entry.value.as_bytes_owned() { Some(v) => Frame::BulkString(v), - None => Frame::Error(Bytes::from_static( - b"WRONGTYPE Operation against a key holding the wrong kind of value", - )), + None => Frame::Null, }, None => Frame::Null, } diff --git a/src/command/string/string_write.rs b/src/command/string/string_write.rs index fab0f691..8f35c33b 100644 --- a/src/command/string/string_write.rs +++ b/src/command/string/string_write.rs @@ -5,7 +5,7 @@ use crate::storage::Database; use crate::storage::entry::{Entry, current_time_ms}; use super::{format_float, parse_f64, parse_i64, parse_positive_i64}; -use crate::command::helpers::{err_wrong_args, extract_bytes, ok}; +use crate::command::helpers::{err_wrong_args, expiry_ms_in_range, extract_bytes, ok}; /// SET command handler with EX/PX/EXAT/PXAT/NX/XX/KEEPTTL/GET options. pub fn set(db: &mut Database, args: &[Frame]) -> Frame { @@ -51,7 +51,20 @@ pub fn set(db: &mut Database, args: &[Frame]) -> Frame { return Frame::Error(Bytes::from_static(b"ERR syntax error")); } match parse_positive_i64(&args[i]) { - Some(secs) => expires_at_ms = current_time_ms() + (secs as u64) * 1000, + Some(secs) => { + expires_at_ms = match (secs as u64) + .checked_mul(1000) + .and_then(|d| current_time_ms().checked_add(d)) + .filter(|ms| expiry_ms_in_range(*ms)) + { + Some(ms) => ms, + None => { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'SET' command", + )); + } + } + } None => { return Frame::Error(Bytes::from_static( b"ERR value is not an integer or out of range", @@ -64,7 +77,19 @@ pub fn set(db: &mut Database, args: &[Frame]) -> Frame { return Frame::Error(Bytes::from_static(b"ERR syntax error")); } match parse_positive_i64(&args[i]) { - Some(ms) => expires_at_ms = current_time_ms() + ms as u64, + Some(ms) => { + expires_at_ms = match current_time_ms() + .checked_add(ms as u64) + .filter(|v| expiry_ms_in_range(*v)) + { + Some(v) => v, + None => { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'SET' command", + )); + } + }; + } None => { return Frame::Error(Bytes::from_static( b"ERR value is not an integer or out of range", @@ -78,7 +103,17 @@ pub fn set(db: &mut Database, args: &[Frame]) -> Frame { } match parse_positive_i64(&args[i]) { Some(ts) => { - expires_at_ms = (ts as u64) * 1000; + expires_at_ms = match (ts as u64) + .checked_mul(1000) + .filter(|ms| expiry_ms_in_range(*ms)) + { + Some(ms) => ms, + None => { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'SET' command", + )); + } + }; } None => { return Frame::Error(Bytes::from_static( @@ -127,6 +162,13 @@ pub fn set(db: &mut Database, args: &[Frame]) -> Frame { None }; + // Check-before-mutate (Redis parity): SET ... GET on a wrong-type key returns + // WRONGTYPE and performs NO write. This takes precedence over NX/XX handling. + // Previously db.set(...) ran unconditionally, overwriting (destroying) the value. + if matches!(old_value, Some(Frame::Error(_))) { + return old_value.unwrap_or(Frame::Null); + } + // NX + XX both set: contradictory, return nil (or old value if GET) if nx && xx { return if get_old { @@ -197,6 +239,43 @@ pub fn mset(db: &mut Database, args: &[Frame]) -> Frame { ok() } +/// MSETNX command handler (single-shard atomic). +/// +/// Sets all key/value pairs only if NONE of the keys already exist. Returns 1 if +/// all were set, 0 if at least one key already existed (and nothing was set). +/// +/// Atomic on a single shard/database (the two phases run without any await point). +/// Cross-shard atomicity is enforced by the coordinator, which rejects MSETNX when +/// keys span shards (CROSSSLOT); see `coordinate_msetnx`. +pub fn msetnx(db: &mut Database, args: &[Frame]) -> Frame { + if args.is_empty() || !args.len().is_multiple_of(2) { + return err_wrong_args("MSETNX"); + } + // Phase 1: verify NONE of the keys exist. + for pair in args.chunks(2) { + let key = match extract_bytes(&pair[0]) { + Some(k) => k, + None => return err_wrong_args("MSETNX"), + }; + if db.exists(key) { + return Frame::Integer(0); + } + } + // Phase 2: all keys absent -> set them all. + for pair in args.chunks(2) { + let key = match extract_bytes(&pair[0]) { + Some(k) => k.clone(), + None => return err_wrong_args("MSETNX"), + }; + let value = match extract_bytes(&pair[1]) { + Some(v) => v.clone(), + None => return err_wrong_args("MSETNX"), + }; + db.set_string(key, value); + } + Frame::Integer(1) +} + /// INCR command handler. pub fn incr(db: &mut Database, args: &[Frame]) -> Frame { if args.len() != 1 { @@ -258,7 +337,15 @@ pub fn decrby(db: &mut Database, args: &[Frame]) -> Frame { )); } }; - incrby_internal(db, key, -delta) + // Guard i64::MIN: -(i64::MIN) is unrepresentable. Redis returns an overflow + // error rather than negating with a debug panic / release wrap. + let neg = match delta.checked_neg() { + Some(n) => n, + None => { + return Frame::Error(Bytes::from_static(b"ERR decrement would overflow")); + } + }; + incrby_internal(db, key, neg) } /// Internal helper for INCR/DECR/INCRBY/DECRBY. @@ -580,7 +667,19 @@ pub fn setex(db: &mut Database, args: &[Frame]) -> Frame { Some(v) => v.clone(), None => return err_wrong_args("SETEX"), }; - db.set_string_with_expiry(key, value, current_time_ms() + (seconds as u64) * 1000); + let expires_at_ms = match (seconds as u64) + .checked_mul(1000) + .and_then(|d| current_time_ms().checked_add(d)) + .filter(|ms| expiry_ms_in_range(*ms)) + { + Some(ms) => ms, + None => { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'SETEX' command", + )); + } + }; + db.set_string_with_expiry(key, value, expires_at_ms); ok() } @@ -611,7 +710,18 @@ pub fn psetex(db: &mut Database, args: &[Frame]) -> Frame { Some(v) => v.clone(), None => return err_wrong_args("PSETEX"), }; - db.set_string_with_expiry(key, value, current_time_ms() + millis as u64); + let expires_at_ms = match current_time_ms() + .checked_add(millis as u64) + .filter(|ms| expiry_ms_in_range(*ms)) + { + Some(ms) => ms, + None => { + return Frame::Error(Bytes::from_static( + b"ERR invalid expire time in 'PSETEX' command", + )); + } + }; + db.set_string_with_expiry(key, value, expires_at_ms); ok() } @@ -636,8 +746,14 @@ pub fn getset(db: &mut Database, args: &[Frame]) -> Frame { )), }); - // GETSET removes TTL (sets new entry without expiry) - db.set_string(key, value); - - old.unwrap_or(Frame::Null) + // Check-before-mutate (Redis parity): a wrong-type key must return WRONGTYPE + // and stay untouched. Previously db.set_string ran unconditionally, destroying it. + match old { + Some(Frame::Error(e)) => Frame::Error(e), + other => { + // GETSET removes TTL (sets new entry without expiry) + db.set_string(key, value); + other.unwrap_or(Frame::Null) + } + } } diff --git a/src/persistence/replay.rs b/src/persistence/replay.rs index 0dd4d4f3..bd6fc9e7 100644 --- a/src/persistence/replay.rs +++ b/src/persistence/replay.rs @@ -203,6 +203,56 @@ mod tests { ); } + // ── EXPIRE<=0 replay / propagation guard ────────────────────────────────── + + /// P0 propagation guard: WAL replay of `EXPIRE k -1` (non-positive TTL) must + /// DELETE the key, exactly as the live handler now does. Replay re-dispatches + /// the raw command, so a WAL-recovered replica stays consistent with the + /// master. Before the EXPIRE<=0 fix this failed (replay kept the key because + /// the handler returned an error and mutated nothing). + #[test] + fn replay_expire_nonpositive_deletes_key() { + let engine = DispatchReplayEngine::new(); + let mut databases = vec![make_db_with_key(b"k", b"v")]; + let mut selected = 0usize; + assert_eq!( + get_key(&mut databases[0], b"k").as_deref(), + Some(b"v".as_ref()), + "precondition: key present before replay" + ); + + let args = framevec![ + Frame::BulkString(bytes::Bytes::from_static(b"k")), + Frame::BulkString(bytes::Bytes::from_static(b"-1")), + ]; + engine.replay_command(&mut databases, b"EXPIRE", &args, &mut selected); + + assert_eq!( + get_key(&mut databases[0], b"k"), + None, + "replay of EXPIRE k -1 must delete the key (no master/replica divergence)" + ); + } + + /// Sibling anchor: EXPIREAT with a past timestamp already deletes on replay. + /// Confirms the EXPIRE fix matches the established EXPIREAT propagation. + #[test] + fn replay_expireat_past_deletes_key() { + let engine = DispatchReplayEngine::new(); + let mut databases = vec![make_db_with_key(b"k", b"v")]; + let mut selected = 0usize; + let args = framevec![ + Frame::BulkString(bytes::Bytes::from_static(b"k")), + Frame::BulkString(bytes::Bytes::from_static(b"-1")), + ]; + engine.replay_command(&mut databases, b"EXPIREAT", &args, &mut selected); + assert_eq!( + get_key(&mut databases[0], b"k"), + None, + "replay of EXPIREAT k -1 must delete the key" + ); + } + /// Same-index SWAPDB is a no-op during replay. #[test] fn replay_swapdb_same_index_noop() { diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 44708e05..1169cba1 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -1286,6 +1286,8 @@ pub(super) async fn try_handle_cross_shard_commands( &ctx.dispatch_tx, &ctx.spsc_notifiers, &ctx.cached_clock, + ctx.aof_pool.as_ref(), + &ctx.repl_state, &(), // monoio: coordinator uses oneshot, not response_pool ) .await; diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 03d16e23..4758babe 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -1035,7 +1035,7 @@ pub(crate) async fn handle_connection_sharded_inner< // --- Multi-key commands --- if is_multi_key_command(cmd, cmd_args) { - let response = crate::shard::coordinator::coordinate_multi_key(cmd, cmd_args, ctx.shard_id, ctx.num_shards, conn.selected_db, &ctx.shard_databases, &ctx.dispatch_tx, &ctx.spsc_notifiers, &ctx.cached_clock, &()).await; + let response = crate::shard::coordinator::coordinate_multi_key(cmd, cmd_args, ctx.shard_id, ctx.num_shards, conn.selected_db, &ctx.shard_databases, &ctx.dispatch_tx, &ctx.spsc_notifiers, &ctx.cached_clock, ctx.aof_pool.as_ref(), &ctx.repl_state, &()).await; responses.push(response); continue; } diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index 5b221580..e8ef4fa5 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -376,6 +376,9 @@ pub(crate) fn is_multi_key_command(cmd: &[u8], args: &[Frame]) -> bool { let b0 = cmd[0] | 0x20; match (len, b0) { (4, b'm') => cmd.eq_ignore_ascii_case(b"MGET") || cmd.eq_ignore_ascii_case(b"MSET"), + // MSETNX: atomic multi-key write; the coordinator rejects it (CROSSSLOT) when + // keys span shards, and runs it atomically when they are co-located. + (6, b'm') => cmd.eq_ignore_ascii_case(b"MSETNX"), // DEL, UNLINK, EXISTS with multiple keys (3, b'd') => args.len() > 1 && cmd.eq_ignore_ascii_case(b"DEL"), (6, b'u') => args.len() > 1 && cmd.eq_ignore_ascii_case(b"UNLINK"), diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index a3222687..ea16bc54 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -40,6 +40,12 @@ pub async fn coordinate_multi_key( dispatch_tx: &Rc>>>, spsc_notifiers: &[Arc], cached_clock: &CachedClock, + // Local-leg persistence context (review Finding 1): the coordinator's + // in-process local legs for MSET/MSETNX append to the owning shard's AOF + // via these, matching the local single-key write contract. `None` disables + // persistence (tests / no-AOF deployments). + aof_pool: Option<&Arc>, + repl_state: ReplStateRef<'_>, _response_pool: &(), // placeholder — coordinator uses oneshot internally ) -> Frame { if cmd.eq_ignore_ascii_case(b"MGET") { @@ -65,6 +71,23 @@ pub async fn coordinate_multi_key( dispatch_tx, spsc_notifiers, cached_clock, + aof_pool, + repl_state, + _response_pool, + ) + .await + } else if cmd.eq_ignore_ascii_case(b"MSETNX") { + coordinate_msetnx( + args, + my_shard, + num_shards, + db_index, + shard_databases, + dispatch_tx, + spsc_notifiers, + cached_clock, + aof_pool, + repl_state, _response_pool, ) .await @@ -196,6 +219,62 @@ async fn run_on_owner( } } +/// Type of the replication-state handle threaded into the coordinator's local +/// persistence path (same shape `AofWriterPool::issue_append_lsn` expects). +type ReplStateRef<'a> = + &'a Option>>; + +/// Persist a coordinator LOCAL-leg write to the owning shard's AOF, matching the +/// local single-key write contract (the `is_write` block in +/// `handler_monoio`/`handler_sharded`): issue an LSN off `repl_state`, then +/// durable-append. Under `appendfsync=always` this awaits the writer's fsync ack; +/// under everysec/no it is fire-and-forget. WAL append is external to +/// `cmd_dispatch`, so the coordinator's in-process local legs (`run_local`, +/// `coordinate_mset` fast path / local slice) MUST call this or their writes are +/// lost on restart while the remote legs (via `wal_append_and_fanout`) survive. +/// +/// `serialized` MUST cover only keys OWNED by `my_shard`: +/// - co-located command (MSETNX; MSET fast path) → the whole command, +/// - scattered MSET local slice → a synthesized MSET over +/// just the local keys (never the full scattered command — `my_shard` does +/// not own the remote keys and replay would misapply them on this shard). +/// +/// Returns `Err(())` on AOF failure so the caller surfaces `AOF_FSYNC_ERR` +/// instead of a false `+OK` (design-for-failure; matches the handler). +async fn persist_local_leg( + aof_pool: Option<&Arc>, + repl_state: ReplStateRef<'_>, + my_shard: usize, + serialized: Bytes, +) -> Result<(), ()> { + let Some(pool) = aof_pool else { return Ok(()) }; + let lsn = crate::persistence::aof::AofWriterPool::issue_append_lsn( + repl_state, + my_shard, + serialized.len(), + ); + match pool + .try_send_append_durable(my_shard, lsn, serialized) + .await + { + Ok(()) => Ok(()), + Err(_) => Err(()), + } +} + +/// Serialize an `MSET k v ...` command over `pairs` for AOF logging of a local +/// MSET leg. Used for both the fast path (all keys local) and a scattered MSET's +/// local slice (only the local keys) — never the full scattered command. +fn serialize_local_mset(pairs: &[(Bytes, Bytes)]) -> Bytes { + let mut parts: Vec = Vec::with_capacity(pairs.len() * 2 + 1); + parts.push(Frame::BulkString(Bytes::from_static(b"MSET"))); + for (k, v) in pairs { + parts.push(Frame::BulkString(k.clone())); + parts.push(Frame::BulkString(v.clone())); + } + crate::persistence::aof::serialize_command(&Frame::Array(parts.into())) +} + fn bulk(b: &Bytes) -> Frame { Frame::BulkString(b.clone()) } @@ -728,6 +807,8 @@ async fn coordinate_mset( dispatch_tx: &Rc>>>, spsc_notifiers: &[Arc], cached_clock: &CachedClock, + aof_pool: Option<&Arc>, + repl_state: ReplStateRef<'_>, _response_pool: &(), // placeholder — coordinator uses oneshot internally ) -> Frame { if args.is_empty() || !args.len().is_multiple_of(2) { @@ -761,10 +842,22 @@ async fn coordinate_mset( // Fast path: all keys on local shard if groups.len() == 1 && groups.contains_key(&my_shard) { - return crate::shard::slice::with_shard_db(db_index, |db| { + let resp = crate::shard::slice::with_shard_db(db_index, |db| { db.refresh_now_from_cache(cached_clock); crate::command::string::mset(db, args) }); + // Local leg (review Finding 1): persist the whole MSET — every key is + // owned by my_shard — matching the local single-key write contract. + if let Some(pairs) = groups.get(&my_shard) { + let serialized = serialize_local_mset(pairs); + if persist_local_leg(aof_pool, repl_state, my_shard, serialized) + .await + .is_err() + { + return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + } + } + return resp; } let mut pending_shards: Vec>> = Vec::new(); @@ -804,9 +897,114 @@ async fn coordinate_mset( let _ = reply_rx.recv().await; } + // Local leg (review Finding 1): persist a synthesized MSET over ONLY the + // local keys. The remote slices persisted themselves on their owner shards + // via MultiExecute -> wal_append_and_fanout; my_shard must not log their keys + // (replay re-dispatches raw commands, so a full-command log here would try to + // write keys this shard doesn't own). + if let Some(pairs) = groups.get(&my_shard) { + let serialized = serialize_local_mset(pairs); + if persist_local_leg(aof_pool, repl_state, my_shard, serialized) + .await + .is_err() + { + return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + } + } + Frame::SimpleString(Bytes::from_static(b"OK")) } +/// Coordinate MSETNX across shards. +/// +/// MSETNX is atomic by contract: set every pair iff *none* of the keys already +/// exist. Moon cannot honor that atomically across shards (like MSET, cross-shard +/// writes scatter with no two-phase commit or rollback), so — by design — MSETNX +/// is rejected with a CROSSSLOT error when its keys hash to more than one shard. +/// When all keys are co-located on a single shard (including via `{hash-tag}`), +/// the whole command runs atomically on that shard's owner. +#[allow(clippy::too_many_arguments)] +async fn coordinate_msetnx( + args: &[Frame], + my_shard: usize, + num_shards: usize, + db_index: usize, + shard_databases: &Arc, + dispatch_tx: &Rc>>>, + spsc_notifiers: &[Arc], + cached_clock: &CachedClock, + aof_pool: Option<&Arc>, + repl_state: ReplStateRef<'_>, + _response_pool: &(), // placeholder — coordinator uses oneshot internally +) -> Frame { + if args.is_empty() || !args.len().is_multiple_of(2) { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'MSETNX' command", + )); + } + + // Every key must hash to the same shard; otherwise MSETNX cannot be atomic. + let first_key = match extract_key(&args[0]) { + Some(k) => k, + None => { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'MSETNX' command", + )); + } + }; + let owner = key_to_shard(&first_key, num_shards); + for pair in args.chunks(2) { + let key = match extract_key(&pair[0]) { + Some(k) => k, + None => { + return Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'MSETNX' command", + )); + } + }; + if key_to_shard(&key, num_shards) != owner { + return Frame::Error(Bytes::from_static( + b"CROSSSLOT Keys in MSETNX request don't hash to the same shard", + )); + } + } + + // All keys co-located -> run the whole MSETNX atomically on the owning shard. + // Branch on ownership explicitly (rather than via run_on_owner) so the LOCAL + // leg can persist to my_shard's AOF on a successful write (review Finding 1); + // the REMOTE leg persists on the owner via MultiExecute -> wal_append_and_fanout. + let mut command_parts: Vec = Vec::with_capacity(args.len() + 1); + command_parts.push(Frame::BulkString(Bytes::from_static(b"MSETNX"))); + command_parts.extend_from_slice(args); + if owner == my_shard { + let resp = run_local(shard_databases, db_index, cached_clock, b"MSETNX", args); + // Persist only on an actual write (:1). A :0 means some key already + // existed and MSETNX wrote nothing — there is nothing to log. + if matches!(resp, Frame::Integer(1)) { + let serialized = + crate::persistence::aof::serialize_command(&Frame::Array(command_parts.into())); + if persist_local_leg(aof_pool, repl_state, my_shard, serialized) + .await + .is_err() + { + return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + } + } + resp + } else { + run_remote( + owner, + &first_key, + Frame::Array(command_parts.into()), + my_shard, + db_index, + dispatch_tx, + spsc_notifiers, + ) + .await + } +} + /// Coordinate DEL/UNLINK/EXISTS with multiple keys across shards using VLL pattern. /// /// Groups keys by shard in ascending order (BTreeMap), dispatches sub-commands @@ -2185,6 +2383,8 @@ mod tests { &dispatch_tx, ¬ifiers, &cached_clock, + None, + &None, &response_pool, ) .await; diff --git a/tests/coordinator_local_leg_durability.rs b/tests/coordinator_local_leg_durability.rs new file mode 100644 index 00000000..378a9681 --- /dev/null +++ b/tests/coordinator_local_leg_durability.rs @@ -0,0 +1,482 @@ +//! ADD milestone `v3-4-kv-correctness` — cross-shard coordinator LOCAL-leg WAL +//! durability (review Finding 1). +//! +//! The cross-shard coordinator splits a multi-key write (MSET / MSETNX) into a +//! LOCAL leg (keys owned by the connection's own shard, executed in-process) and +//! REMOTE legs (keys owned by other shards, executed via `ShardMessage:: +//! MultiExecute`). The remote legs persist through `wal_append_and_fanout`; the +//! local leg historically executed the write in memory but NEVER appended it to +//! the owning shard's AOF. So a co-located MSET/MSETNX was durable when its owner +//! was a *remote* shard but silently **non-durable** when its owner was the +//! connection's *own* shard — the write survived until the next SIGKILL, then +//! vanished on restart. +//! +//! ## Why this is deterministic despite SO_REUSEPORT +//! +//! The connection lands on some shard `C` (SO_REUSEPORT picks it, we cannot +//! predict which). Each test writes ONE co-located group per shard (via a +//! `{hash-tag}` that provably routes to that shard). Whichever shard `C` is, +//! exactly ONE group is the LOCAL leg and the rest are remote. We assert ALL +//! groups survive a crash+restart: +//! - pre-fix: the group whose owner == `C` is gone (local leg never persisted) +//! → the assertion fails (RED) no matter which shard `C` turned out to be. +//! - post-fix: the local leg persists via the same `aof_pool` path every local +//! write uses → every group recovers (GREEN). +//! +//! All writes go through a SINGLE reused connection so `my_shard` (== `C`) is +//! stable across the whole write phase. +//! +//! Harness: `CARGO_BIN_EXE_moon` + a hand-rolled RESP2 client (no redis-cli), so +//! this runs as a live gate under both runtimes. Fsync policy + 1.5s quiescing +//! sleep before SIGKILL mirror the proven `crash_matrix_per_shard_aof.rs` +//! everysec pattern (the sleep drains the fire-and-forget remote-leg appends so +//! the ONLY thing a crash can drop pre-fix is the un-persisted local leg). +//! +//! Run alone with: cargo test --test coordinator_local_leg_durability + +use std::io::{Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +use moon::shard::dispatch::key_to_shard; + +// --------------------------------------------------------------------------- +// Harness (CARGO_BIN_EXE pattern, mirrors msetnx_cross_shard_reject.rs + +// crash_matrix_per_shard_aof.rs) +// --------------------------------------------------------------------------- + +const SHARDS: u32 = 4; +/// Keys per co-located group (a multi-key MSET/MSETNX, not a single SET). +const GROUP_SIZE: usize = 3; + +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 +} + +/// Spawn moon with per-shard AOF enabled (`--appendonly yes`). `everysec` + +/// a 1.5s quiescing sleep before the kill gives 100% durability for everything +/// that was actually appended — so the only pre-fix loss is the local leg that +/// was never appended at all. +fn spawn_moon_aof(port: u16, dir: &std::path::Path, shards: u32) -> Child { + Command::new(moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "yes", + "--appendfsync", + "everysec", + ]) + .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 (CARGO_BIN_EXE_moon)") +} + +struct ServerGuard(Child); +impl Drop for ServerGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// SIGKILL round-1 and reap it so the port + AOF file handles are released +/// before round-2 spawns. `Child::kill()` sends SIGKILL on Unix. +fn sigkill(mut child: Child) { + let _ = child.kill(); + let _ = child.wait(); +} + +fn connect(port: u16, deadline: Duration) -> TcpStream { + let addr = format!("127.0.0.1:{port}") + .to_socket_addrs() + .expect("addr") + .next() + .expect("one addr"); + let start = Instant::now(); + loop { + match TcpStream::connect_timeout(&addr, Duration::from_millis(200)) { + Ok(s) => { + s.set_read_timeout(Some(Duration::from_secs(5))).ok(); + s.set_write_timeout(Some(Duration::from_secs(5))).ok(); + return s; + } + Err(_) if start.elapsed() < deadline => { + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => panic!("server never accepted on {port}: {e}"), + } + } +} + +fn wait_ready(port: u16) { + let mut s = connect(port, Duration::from_secs(30)); + let start = Instant::now(); + loop { + s.write_all(b"PING\r\n").expect("write PING"); + let mut buf = [0u8; 64]; + if let Ok(n) = s.read(&mut buf) + && n > 0 + && buf[..n].windows(4).any(|w| w == b"PONG") + { + return; + } + assert!( + start.elapsed() < Duration::from_secs(10), + "server accepted TCP but never answered PING" + ); + std::thread::sleep(Duration::from_millis(100)); + s = connect(port, Duration::from_secs(5)); + } +} + +// --------------------------------------------------------------------------- +// Minimal RESP2 client +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, PartialEq)] +enum Resp { + Simple(String), + Error(String), + Int(i64), + Bulk(Option>), + Array(Option>), +} + +struct Conn { + s: TcpStream, + buf: Vec, + pos: usize, +} + +impl Conn { + fn open(port: u16) -> Self { + Conn { + s: connect(port, Duration::from_secs(10)), + buf: Vec::with_capacity(16 * 1024), + pos: 0, + } + } + + fn cmd(&mut self, parts: &[&str]) -> Resp { + 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).expect("write cmd"); + self.frame() + } + + fn fill(&mut self) { + let mut chunk = [0u8; 16 * 1024]; + let n = self.s.read(&mut chunk).expect("read"); + assert!(n > 0, "connection closed mid-frame"); + self.buf.extend_from_slice(&chunk[..n]); + } + + fn line(&mut self) -> String { + loop { + if let Some(rel) = self.buf[self.pos..].windows(2).position(|w| w == b"\r\n") { + let line = + String::from_utf8_lossy(&self.buf[self.pos..self.pos + rel]).into_owned(); + self.pos += rel + 2; + return line; + } + self.fill(); + } + } + + fn exact(&mut self, n: usize) -> Vec { + while self.buf.len() - self.pos < n + 2 { + self.fill(); + } + let out = self.buf[self.pos..self.pos + n].to_vec(); + self.pos += n + 2; + out + } + + fn frame(&mut self) -> Resp { + if self.pos > 0 && self.pos == self.buf.len() { + self.buf.clear(); + self.pos = 0; + } + let line = self.line(); + let (tag, rest) = line.split_at(1); + match tag { + "+" => Resp::Simple(rest.to_string()), + "-" => Resp::Error(rest.to_string()), + ":" => Resp::Int(rest.parse().unwrap_or(0)), + "$" => { + let n: i64 = rest.parse().unwrap_or(-1); + if n < 0 { + Resp::Bulk(None) + } else { + Resp::Bulk(Some(self.exact(n as usize))) + } + } + "*" => { + let n: i64 = rest.parse().unwrap_or(-1); + if n < 0 { + Resp::Array(None) + } else { + let mut items = Vec::with_capacity(n as usize); + for _ in 0..n { + items.push(self.frame()); + } + Resp::Array(Some(items)) + } + } + other => panic!("unexpected RESP tag {other:?} (line {line:?})"), + } + } +} + +/// Find, for each shard, a `{hash-tag}` value whose tagged keys provably route +/// to that shard (via moon's own `key_to_shard`, so co-location is proven, not +/// assumed). `out[s]` is a tag whose `{tag}...` keys land on shard `s`. +fn tags_per_shard(num_shards: usize) -> Vec { + let mut out: Vec> = vec![None; num_shards]; + let mut found = 0; + for i in 0..1_000_000 { + let tag = format!("s{i}"); + // Only the tag inside `{}` is hashed, so any suffix routes identically. + let probe = format!("{{{}}}k", tag); + let s = key_to_shard(probe.as_bytes(), num_shards); + if out[s].is_none() { + out[s] = Some(tag); + found += 1; + if found == num_shards { + break; + } + } + } + out.into_iter() + .map(|o| o.expect("found a tag for every shard")) + .collect() +} + +/// The `(key, value)` pairs of the co-located group on shard `s`. +fn group_pairs(tag: &str) -> Vec<(String, String)> { + (0..GROUP_SIZE) + .map(|j| (format!("{{{}}}:k{}", tag, j), format!("{}-v{}", tag, j))) + .collect() +} + +/// A moon `MOONERR diskfull` write-pause reply. When the data volume dips below +/// the 5%-free guard, writes are refused and this durability test cannot run — +/// callers SKIP (not fail), so a full dev/CI disk doesn't masquerade as a +/// regression. On a healthy host (ample `/tmp`, as on CI) this never fires. +fn is_diskfull(r: &Resp) -> bool { + matches!(r, Resp::Error(m) if m.contains("diskfull")) +} + +/// Kill round-1, restart, and assert EVERY written pair recovered. The failure +/// message identifies the missing group so a RED run points straight at the +/// un-persisted local leg. +fn assert_all_survive_restart( + port: u16, + dir: &std::path::Path, + child1: Child, + expected: &[(String, String)], + what: &str, +) { + // > 1s so the everysec fsync window flushed every append that WAS made. + std::thread::sleep(Duration::from_millis(1500)); + sigkill(child1); + + let _guard = ServerGuard(spawn_moon_aof(port, dir, SHARDS)); + wait_ready(port); + + let mut c = Conn::open(port); + let mut missing: Vec = Vec::new(); + let mut mismatched: Vec = Vec::new(); + for (k, v) in expected { + match c.cmd(&["GET", k]) { + Resp::Bulk(Some(got)) if got == v.as_bytes() => {} + Resp::Bulk(Some(got)) => mismatched.push(format!( + "{k}: want={v} got={}", + String::from_utf8_lossy(&got) + )), + Resp::Bulk(None) => missing.push(k.clone()), + other => panic!("unexpected GET reply for {k}: {other:?}"), + } + } + + assert!( + missing.is_empty() && mismatched.is_empty(), + "{what}: coordinator LOCAL leg lost writes across restart — {} missing, {} mismatched. \ + Missing (the co-located group whose owner == the connection's own shard, whose local \ + leg never hit the AOF): {:?}; mismatched: {:?}", + missing.len(), + mismatched.len(), + missing.iter().take(GROUP_SIZE + 1).collect::>(), + mismatched.iter().take(5).collect::>(), + ); +} + +// --------------------------------------------------------------------------- +// MSET — fast path (all keys of a group co-located on one shard). +// The group whose owner == the connection's shard takes the `groups.len() == 1 +// && contains_key(&my_shard)` fast path → `string::mset(db, args)` with no AOF. +// --------------------------------------------------------------------------- + +#[test] +fn mset_colocated_local_leg_persists_across_restart() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + wait_ready(port); + + let tags = tags_per_shard(SHARDS as usize); + let mut expected: Vec<(String, String)> = Vec::new(); + + // ONE connection for the whole write phase → my_shard is stable. + let mut c = Conn::open(port); + for tag in &tags { + let pairs = group_pairs(tag); + let mut argv: Vec<&str> = vec!["MSET"]; + for (k, v) in &pairs { + argv.push(k); + argv.push(v); + } + let resp = c.cmd(&argv); + if is_diskfull(&resp) { + eprintln!( + "SKIP mset_colocated: MOONERR diskfull — durability untestable on a \ + <5%-free filesystem; needs ample /tmp (CI has it)" + ); + return; + } + assert_eq!( + resp, + Resp::Simple("OK".to_string()), + "co-located MSET on tag {tag} should return OK" + ); + expected.extend(pairs); + } + drop(c); + + assert_all_survive_restart( + port, + dir.path(), + child1, + &expected, + "MSET co-located fast-path", + ); +} + +// --------------------------------------------------------------------------- +// MSET — scatter path (ONE MSET spanning every shard). The slice owned by the +// connection's shard is the local leg (`db.set_string` per key, no AOF); the +// other slices scatter remotely (persisted). +// --------------------------------------------------------------------------- + +#[test] +fn mset_scatter_local_slice_persists_across_restart() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + wait_ready(port); + + let tags = tags_per_shard(SHARDS as usize); + // One key per shard, all in a single MSET → guaranteed multi-shard scatter + // with a local slice on whichever shard the connection landed on. + let expected: Vec<(String, String)> = tags + .iter() + .map(|tag| (format!("{{{}}}:scatter", tag), format!("{}-scatter", tag))) + .collect(); + + let mut argv: Vec<&str> = vec!["MSET"]; + for (k, v) in &expected { + argv.push(k); + argv.push(v); + } + let mut c = Conn::open(port); + let resp = c.cmd(&argv); + if is_diskfull(&resp) { + eprintln!( + "SKIP mset_scatter: MOONERR diskfull — durability untestable on a \ + <5%-free filesystem; needs ample /tmp (CI has it)" + ); + return; + } + assert_eq!( + resp, + Resp::Simple("OK".to_string()), + "scatter MSET should return OK" + ); + drop(c); + + assert_all_survive_restart( + port, + dir.path(), + child1, + &expected, + "MSET scatter local-slice", + ); +} + +// --------------------------------------------------------------------------- +// MSETNX — co-located (all-new → :1). The group whose owner == the connection's +// shard runs via `run_on_owner` → `run_local` with no AOF append. +// --------------------------------------------------------------------------- + +#[test] +fn msetnx_colocated_local_leg_persists_across_restart() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let child1 = spawn_moon_aof(port, dir.path(), SHARDS); + wait_ready(port); + + let tags = tags_per_shard(SHARDS as usize); + let mut expected: Vec<(String, String)> = Vec::new(); + + let mut c = Conn::open(port); + for tag in &tags { + let pairs = group_pairs(tag); + let mut argv: Vec<&str> = vec!["MSETNX"]; + for (k, v) in &pairs { + argv.push(k); + argv.push(v); + } + let resp = c.cmd(&argv); + if is_diskfull(&resp) { + eprintln!( + "SKIP msetnx_colocated: MOONERR diskfull — durability untestable on a \ + <5%-free filesystem; needs ample /tmp (CI has it)" + ); + return; + } + assert_eq!( + resp, + Resp::Int(1), + "all-new co-located MSETNX on tag {tag} should return 1" + ); + expected.extend(pairs); + } + drop(c); + + assert_all_survive_restart( + port, + dir.path(), + child1, + &expected, + "MSETNX co-located local leg", + ); +} diff --git a/tests/msetnx_cross_shard_reject.rs b/tests/msetnx_cross_shard_reject.rs new file mode 100644 index 00000000..262d4860 --- /dev/null +++ b/tests/msetnx_cross_shard_reject.rs @@ -0,0 +1,311 @@ +//! ADD milestone `v3-4-kv-correctness` — MSETNX cross-shard contract. +//! +//! MSETNX is atomic by contract (set every pair iff none of the keys exist). +//! Moon cannot honor that atomically across shards, so — by deliberate design +//! decision — a MSETNX whose keys span more than one shard is REJECTED with a +//! CROSSSLOT error and writes nothing. When the keys are co-located on a single +//! shard (naturally or via a `{hash-tag}`) the whole command runs atomically on +//! that shard's owner. +//! +//! These are green-after-fix regression tests. They assert BOTH sides so the +//! suite cannot pass vacuously: an "always reject" bug fails the co-located +//! case; a "never reject / route-by-connection" bug fails the cross-shard case. +//! +//! Run alone with: cargo test --test msetnx_cross_shard_reject + +use std::io::{Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +use moon::shard::dispatch::key_to_shard; + +// --------------------------------------------------------------------------- +// Harness (CARGO_BIN_EXE pattern, mirrors cross_shard_consistency_red.rs) +// --------------------------------------------------------------------------- + +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 +} + +fn spawn_moon(port: u16, dir: &std::path::Path, shards: u32) -> Child { + Command::new(moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + ]) + .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 (CARGO_BIN_EXE_moon)") +} + +struct ServerGuard(Child); +impl Drop for ServerGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn connect(port: u16, deadline: Duration) -> TcpStream { + let addr = format!("127.0.0.1:{port}") + .to_socket_addrs() + .expect("addr") + .next() + .expect("one addr"); + let start = Instant::now(); + loop { + match TcpStream::connect_timeout(&addr, Duration::from_millis(200)) { + Ok(s) => { + s.set_read_timeout(Some(Duration::from_secs(5))).ok(); + s.set_write_timeout(Some(Duration::from_secs(5))).ok(); + return s; + } + Err(_) if start.elapsed() < deadline => { + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => panic!("server never accepted on {port}: {e}"), + } + } +} + +fn wait_ready(port: u16) -> TcpStream { + let mut s = connect(port, Duration::from_secs(30)); + let start = Instant::now(); + loop { + s.write_all(b"PING\r\n").expect("write PING"); + let mut buf = [0u8; 64]; + if let Ok(n) = s.read(&mut buf) + && n > 0 + && buf[..n].windows(4).any(|w| w == b"PONG") + { + return s; + } + assert!( + start.elapsed() < Duration::from_secs(10), + "server accepted TCP but never answered PING" + ); + std::thread::sleep(Duration::from_millis(100)); + s = connect(port, Duration::from_secs(5)); + } +} + +// --------------------------------------------------------------------------- +// Minimal RESP2 reader +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, PartialEq)] +enum Resp { + Simple(String), + Error(String), + Int(i64), + Bulk(Option>), + Array(Option>), +} + +struct Conn { + s: TcpStream, + buf: Vec, + pos: usize, +} + +impl Conn { + fn open(port: u16) -> Self { + Conn { + s: connect(port, Duration::from_secs(10)), + buf: Vec::with_capacity(16 * 1024), + pos: 0, + } + } + + fn cmd_s(&mut self, parts: &[&str]) -> Resp { + 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).expect("write cmd"); + self.frame() + } + + fn fill(&mut self) { + let mut chunk = [0u8; 16 * 1024]; + let n = self.s.read(&mut chunk).expect("read"); + assert!(n > 0, "connection closed mid-frame"); + self.buf.extend_from_slice(&chunk[..n]); + } + + fn line(&mut self) -> String { + loop { + if let Some(rel) = self.buf[self.pos..].windows(2).position(|w| w == b"\r\n") { + let line = + String::from_utf8_lossy(&self.buf[self.pos..self.pos + rel]).into_owned(); + self.pos += rel + 2; + return line; + } + self.fill(); + } + } + + fn exact(&mut self, n: usize) -> Vec { + while self.buf.len() - self.pos < n + 2 { + self.fill(); + } + let out = self.buf[self.pos..self.pos + n].to_vec(); + self.pos += n + 2; + out + } + + fn frame(&mut self) -> Resp { + if self.pos > 0 && self.pos == self.buf.len() { + self.buf.clear(); + self.pos = 0; + } + let line = self.line(); + let (tag, rest) = line.split_at(1); + match tag { + "+" => Resp::Simple(rest.to_string()), + "-" => Resp::Error(rest.to_string()), + ":" => Resp::Int(rest.parse().unwrap_or(0)), + "$" => { + let n: i64 = rest.parse().unwrap_or(-1); + if n < 0 { + Resp::Bulk(None) + } else { + Resp::Bulk(Some(self.exact(n as usize))) + } + } + "*" => { + let n: i64 = rest.parse().unwrap_or(-1); + if n < 0 { + Resp::Array(None) + } else { + let mut items = Vec::with_capacity(n as usize); + for _ in 0..n { + items.push(self.frame()); + } + Resp::Array(Some(items)) + } + } + other => panic!("unexpected RESP tag {other:?} (line {line:?})"), + } + } +} + +/// Return one key per shard (index = shard id), generated deterministically via +/// moon's own hash so cross-shard-ness is provable, not assumed. +fn keys_per_shard(prefix: &str, num_shards: usize) -> Vec { + let mut out: Vec> = vec![None; num_shards]; + let mut found = 0; + for i in 0..10_000 { + let k = format!("{prefix}{i}"); + let s = key_to_shard(k.as_bytes(), num_shards); + if out[s].is_none() { + out[s] = Some(k); + found += 1; + if found == num_shards { + break; + } + } + } + out.into_iter() + .map(|o| o.expect("found a key for every shard")) + .collect() +} + +const SHARDS: u32 = 4; + +// --------------------------------------------------------------------------- +// Cross-shard MSETNX is rejected (CROSSSLOT) and writes NOTHING. +// --------------------------------------------------------------------------- + +#[test] +fn msetnx_cross_shard_rejected_no_partial_write() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + drop(wait_ready(port)); + + // Two keys that PROVABLY land on different shards. + let ks = keys_per_shard("msnx:", SHARDS as usize); + let (k0, k1) = (&ks[0], &ks[1]); + assert_ne!( + key_to_shard(k0.as_bytes(), SHARDS as usize), + key_to_shard(k1.as_bytes(), SHARDS as usize), + "test precondition: keys must span shards" + ); + + let mut c = Conn::open(port); + let r = c.cmd_s(&["MSETNX", k0, "v0", k1, "v1"]); + match &r { + Resp::Error(m) => assert!( + m.starts_with("CROSSSLOT"), + "cross-shard MSETNX must be a CROSSSLOT error (got {r:?})" + ), + other => panic!("cross-shard MSETNX must be rejected, got {other:?}"), + } + + // Reject is total: NEITHER key was written (no partial side effects). + assert_eq!( + c.cmd_s(&["GET", k0]), + Resp::Bulk(None), + "rejected MSETNX must not write k0" + ); + assert_eq!( + c.cmd_s(&["GET", k1]), + Resp::Bulk(None), + "rejected MSETNX must not write k1" + ); +} + +// --------------------------------------------------------------------------- +// Co-located MSETNX runs atomically on the owner shard (all-or-nothing). +// --------------------------------------------------------------------------- + +#[test] +fn msetnx_colocated_is_atomic() { + let port = free_port(); + let dir = tempfile::tempdir().expect("tempdir"); + let _guard = ServerGuard(spawn_moon(port, dir.path(), SHARDS)); + drop(wait_ready(port)); + + let mut c = Conn::open(port); + + // All keys share the {t} hash-tag -> one shard, even at --shards 4. + assert_eq!( + c.cmd_s(&["MSETNX", "{t}a", "v1", "{t}b", "v2"]), + Resp::Int(1), + "all-new co-located MSETNX returns 1" + ); + assert_eq!(c.cmd_s(&["GET", "{t}a"]), Resp::Bulk(Some(b"v1".to_vec()))); + assert_eq!(c.cmd_s(&["GET", "{t}b"]), Resp::Bulk(Some(b"v2".to_vec()))); + + // One key already exists -> whole command is a no-op, returns 0. + assert_eq!( + c.cmd_s(&["MSETNX", "{t}b", "vX", "{t}c", "v3"]), + Resp::Int(0), + "MSETNX with any existing key returns 0" + ); + // Atomic: the new key {t}c must NOT have been written. + assert_eq!( + c.cmd_s(&["GET", "{t}c"]), + Resp::Bulk(None), + "MSETNX no-op must not write {{t}}c" + ); + // And the pre-existing value is unchanged. + assert_eq!(c.cmd_s(&["GET", "{t}b"]), Resp::Bulk(Some(b"v2".to_vec()))); +}