diff --git a/.add/state.json b/.add/state.json index c943c4be1..7ded1c3c4 100644 --- a/.add/state.json +++ b/.add/state.json @@ -1,7 +1,7 @@ { "project": "moon", "stage": "production", - "active_task": "monoio-ci-coverage", + "active_task": "watch-cas-transactions", "active_milestone": "v0-9-client-compat", "tasks": { "hotpath-lock-quickwins": { @@ -275,14 +275,27 @@ }, "watch-cas-transactions": { "title": "WATCH/UNWATCH optimistic locking on both production dispatch paths", - "phase": "ground", + "phase": "build", "gate": "none", "milestone": "v0-9-client-compat", "depends_on": [ "client-compat-harness" ], "created": "2026-08-09T16:42:29+00:00", - "updated": "2026-08-09T16:42:29+00:00" + "updated": "2026-08-11T12:31:35+00:00", + "tripwire": { + "contract_md5": "6f2cd3da5562c2cc4bd2fcf12c373405", + "tests": {} + }, + "scope": { + "declared": [ + "src/server/conn/shared.rs", + "src/server/conn/watch.rs", + "src/server/conn/core.rs" + ], + "snapshot_md5": "fe6a109ebabb70ce61c2310ddd28f5e2" + }, + "flag_verified": true }, "protocol-error-lifetime": { "title": "Protocol errors reply and close cleanly, never stall or eat the valid prefix", @@ -410,7 +423,7 @@ } }, "created": "2026-06-11T03:18:21+00:00", - "updated": "2026-08-10T08:35:14+00:00", + "updated": "2026-08-11T12:31:35+00:00", "setup": { "locked": true, "locked_at": "2026-06-11T03:28:00+00:00", diff --git a/.add/tasks/watch-cas-transactions/TASK.md b/.add/tasks/watch-cas-transactions/TASK.md index 0ab66bd34..28788f5a4 100644 --- a/.add/tasks/watch-cas-transactions/TASK.md +++ b/.add/tasks/watch-cas-transactions/TASK.md @@ -2,49 +2,143 @@ slug: watch-cas-transactions · created: 2026-08-09 · stage: production autonomy: auto -phase: ground - +phase: build > One file = one task. Fill sections top-to-bottom; the `add` skill drives each phase. -> When a phase is unclear, read its book chapter in `.add/docs/` (linked per section). +> When a phase is unclear, read its book chapter in `.add/docs/`. > The phase marker above is the single source of truth — keep it in sync via `add.py phase`. --- ## 0 · GROUND — the real codebase ▸ docs/02-the-flow.md -Touches (files · symbols · signatures): -Context (working folder): -Honors (patterns / conventions): -Anchors the contract cites: +Touches (files · symbols · signatures): +- `src/server/conn/core.rs:282` — `ConnectionState.watched_keys: HashMap`; the field + already exists on EVERY connection, on every path. Only the embedded path ever writes it. +- `src/server/conn/handler_single.rs:1558,1581` — the ONLY WATCH/UNWATCH handling in the tree + (inline `eq_ignore_ascii_case` arms, not a dispatch-table entry). Embedded path only. +- `src/server/conn/shared.rs:123` — `execute_transaction(db, queue, watched_keys, …)`; does the + CAS check at line 135. `#[cfg(feature = "runtime-tokio")]`, called only by `handler_single`. +- `src/server/conn/shared.rs:221` — `execute_transaction_sharded(shard_databases, shard_id, + command_queue, selected_db, proto, cached_clock, exec_publishes, exec_flushes)`. **No + `watched_keys` parameter exists.** Called by `handler_monoio/write.rs:861`, + `handler_sharded/write.rs:686`, and `shard/spsc_handler.rs:2788` — i.e. by everything that ships. +- `src/server/conn/shared.rs:913` — `TxnLocality::{Keyless, SingleShard(usize), CrossShard}` and + `analyze_txn_locality(command_queue, num_shards)`; the existing rule that a MULTI body whose keys + span shards is rejected `CROSSSLOT` (`handler_monoio/write.rs:849`). +- `src/storage/db/kv_ops.rs:926,931` — `get_version(&[u8]) -> u32` / `increment_version`. Version is + bumped inside `Database::set()` itself (`kv_ops.rs:346`), so every path already maintains it. +- `src/storage/entry.rs:332,422` — `INITIAL_VERSION: u32 = 1`; 24-bit, wraps to 1, never 0. +- `src/command/metadata.rs:412` — WATCH/UNWATCH carry arity + ACL category `TXN`, with no dispatch + arm behind them. + +Context (working folder): `.add/tasks/watch-cas-transactions/` · probes in `tmp/probe_watch*.py`. + +Honors (patterns / conventions): three-dispatch-paths rule (CLAUDE.md — a command needs +`dispatch` + `dispatch_read` + inline wiring or it is CI-invisible); per-shard locks only, no +global lock on the write path; `Frame::Error` in dispatch, never `Result`. + +Anchors the contract cites: `execute_transaction_sharded`, `analyze_txn_locality`, `TxnLocality`, +`ConnectionState.watched_keys`, `Database::get_version`. + +### Measured on `origin/main` @8b1153b4 (not inferred) +`tmp/probe_watch.py` / `probe_watch2.py` / `probe_watch3.py`, release-fast default (monoio) build: + +| probe | shards=1 | shards=4 | inline path | +|---|---|---|---| +| `WATCH k` | `-ERR unknown command 'WATCH'` | same | same | +| `UNWATCH` | `-ERR unknown command 'UNWATCH'` | same | same | + +Two connections, conflicting write between MULTI and EXEC: +`A WATCH k` → error · `A MULTI` → OK · `A SET k from-A` → QUEUED · **`B SET k from-B` → OK** · +`A EXEC` → `*1 +OK` (committed) · final `GET k` → `from-A`. **B's write was clobbered by a +transaction that had declared a dependency on k.** + +`ACL CAT transaction` → lists `watch` and `unwatch`. The ACL surface advertises two commands +dispatch rejects. + +Out of scope, found while probing, do not fix here: `COMMAND COUNT` replies `*0` — an empty +array where an integer belongs. Belongs to `info-observability`; filing separately. --- ## 1 · SPECIFY — the rules ▸ docs/03-step-1-specify.md -Feature: -Framings weighed: (chosen) · · +Feature: WATCH/UNWATCH optimistic locking (CAS) on the dispatch paths Moon ships. + +Framings weighed: +- **Co-located watch (chosen).** Watched keys must hash to the same shard as the transaction body; + a watch that spans shards is refused LOUDLY. Matches the contract MULTI/EXEC bodies already + ship (`TxnLocality::CrossShard` → `CROSSSLOT`), needs no cross-shard read, and at `--shards 1` + (the documented default for non-pipelined work) it is full standalone-Redis parity. +- *Cross-shard watch via scatter.* Read each watched key's version from its owning shard at WATCH + time and re-validate at EXEC. Rejected: the body commits under the OWNER shard's lock only, so a + version validated on shard X can change before the body commits on shard Y. Closing that needs a + global lock across shards, which the architecture forbids. It would buy parity by making the + guarantee a lie. +- *Silently ignore cross-shard watches.* Rejected outright — that is the current failure mode + generalized, and a CAS guarantee that silently does not hold is worse than none. + Must: - - + - `WATCH k [k …]` replies `+OK` on every production path (monoio, sharded, inline) and records + each key's current version on the connection. + - After `WATCH k`, if any watched key's version differs at `EXEC`, `EXEC` replies Null (RESP2 + `*-1`, RESP3 `_`) and executes NO queued command. + - `EXEC` clears all watches — on both the committed and the aborted outcome. + - `UNWATCH` replies `+OK` and clears all watches, inside or outside MULTI. + - `DISCARD` clears all watches. + - A watch on a key that does not exist is honored: if the key is created before `EXEC`, the + transaction aborts. + - Watching, then EXEC with no conflicting write, commits exactly as it does today. + - Behavior is identical on `handler_monoio` and `handler_sharded`, and identical to the embedded + `handler_single` path at `--shards 1`. + Reject: - - -> "" + - `WATCH` with no key argument -> "ERR wrong number of arguments for 'watch' command" + - `WATCH` issued while already inside MULTI -> "ERR WATCH inside MULTI is not allowed" + - watched keys hashing to a different shard than the EXEC body's owner + -> "CROSSSLOT Keys in MULTI/EXEC don't hash to the same shard" + - `UNWATCH` with any argument -> "ERR wrong number of arguments for 'unwatch' command" + After: - - + - `ConnectionState.watched_keys` is authoritative on every path, not just the embedded one. + - `ACL CAT transaction` no longer advertises a command that dispatch rejects. + - A stock client's optimistic-locking loop (redis-py `pipeline().watch()`, go-redis + `TxPipelined`, Lettuce) completes without special-casing at `--shards 1`. + Assumptions — lowest-confidence first: - ⚠ — lowest confidence because ; if wrong: - - [ ] — confirm or deny; never carry an open one forward + ⚠ **Co-locating watched keys with the body is acceptable divergence from standalone Redis.** + Lowest confidence because standalone Redis lets you WATCH any key regardless of where it + lives, and Moon's shards are an internal detail a client cannot see — a client that WATCHes + two keys which happen to land on different shards gets a `CROSSSLOT` it cannot have predicted. + Mitigation: the same is already true of MULTI bodies today, so this adds no NEW class of + surprise, and hash tags (`{tag}`) are the documented way to force co-location. If wrong: + multi-key CAS loops fail on multi-shard deployments and we need the scatter design (much + larger, and per the framings above, not obviously soundable). + ⚠ **[spec] The version counter is a sound CAS token.** It is not, in one case: versions are + per-entry, start at `INITIAL_VERSION = 1`, and are destroyed with the entry. DEL + re-SET + returns the key to version 1, so `WATCH k` (v1) → other client DELs and re-creates k → `EXEC` + sees v1 and **commits**, where Redis aborts. A real ABA hole, and it is in the SHIPPED embedded + path today, not something this task introduces. Second, rarer instance: the counter is 24-bit + and wraps to 1, so 2^24 writes to one key between WATCH and EXEC also collide. + Cost if carried: a CAS loop can silently miss a delete-recreate. Decision needed at freeze — + see the flag in §3. Preferred: treat it as in-scope, because shipping "WATCH works" while this + hole is open re-creates the exact silent-guarantee problem this task exists to close. + - [ ] `WATCH` inside MULTI is an error, not a queued command — matches Redis and the existing + `handler_single` arm at line 1559. Confirmed by reading that arm. + - [ ] The embedded path's existing semantics are the reference for the production paths, so + parity tests can assert the two agree. Confirmed: `execute_transaction` line 135 is the only + CAS implementation in the tree. - + --- @@ -53,11 +147,64 @@ Assumptions — lowest-confidence first: ```gherkin -Scenario: - Given - When - Then - And # required for every rejection +Scenario: a conflicting write aborts the transaction + Given conn A has sent WATCH k and queued SET k from-A inside MULTI + When conn B sets k to from-B, then conn A sends EXEC + Then EXEC replies Null + And k still holds from-B # A's queued write never ran + +Scenario: no conflict commits normally + Given conn A has sent WATCH k and queued SET k from-A inside MULTI + When conn A sends EXEC with no intervening write + Then EXEC replies an array of one +OK + And k holds from-A + +Scenario: watching a key that does not exist + Given conn A has sent WATCH absent and queued SET other v inside MULTI + When conn B creates absent, then conn A sends EXEC + Then EXEC replies Null + And other was not created + +Scenario: UNWATCH releases the dependency + Given conn A has sent WATCH k, then UNWATCH, then queued SET k from-A inside MULTI + When conn B sets k to from-B, then conn A sends EXEC + Then EXEC replies an array of one +OK + And k holds from-A + +Scenario: EXEC clears watches on both outcomes + Given conn A completed one aborted WATCH/MULTI/EXEC cycle on k + When conn A immediately runs MULTI, SET k v2, EXEC with no new WATCH + Then EXEC replies an array of one +OK # the stale watch did not survive + And k holds v2 + +Scenario: delete-and-recreate is a conflict (the ABA hole) + Given conn A has sent WATCH k while k holds v0, and queued SET other v inside MULTI + When conn B deletes k and re-creates it with v0, then conn A sends EXEC + Then EXEC replies Null + And other was not created + +Scenario: the monoio and sharded paths agree with the embedded path + Given the same WATCH/MULTI/EXEC conflict sequence + When it is replayed against shards=1 and shards=4 + Then every reply is byte-identical across the three dispatch paths + +Scenario: WATCH with no arguments is refused + Given a connection outside MULTI + When it sends WATCH with no key + Then the reply is ERR wrong number of arguments for 'watch' command + And no watch was recorded # a later EXEC still commits + +Scenario: WATCH inside MULTI is refused + Given a connection that has sent MULTI + When it sends WATCH k + Then the reply is ERR WATCH inside MULTI is not allowed + And the command was not queued # EXEC returns one fewer reply + +Scenario: a cross-shard watch is refused loudly + Given shards=4 and two keys that hash to different shards + When a connection WATCHes both and EXECs a body touching one of them + Then the reply is a CROSSSLOT error + And no queued command ran ``` @@ -69,36 +216,121 @@ Scenario: ## 3 · CONTRACT — freeze the shape ▸ docs/05-step-3-contract.md ``` - body: { } - 200 -> { } - 4xx -> { error: "" | "" } -Schema: +WATCH key [key ...] -> +OK + -> -ERR wrong number of arguments for 'watch' command + -> -ERR WATCH inside MULTI is not allowed +UNWATCH -> +OK + -> -ERR wrong number of arguments for 'unwatch' command +EXEC (with watches held) -> Null # any watched version changed + -> Array[reply, ...] # all versions unchanged + -> -CROSSSLOT Keys in MULTI/EXEC don't hash to the same shard + # post-condition on EVERY outcome: watched_keys is empty + +Internal shape (the anchor that must change): + execute_transaction_sharded(shard_databases, shard_id, command_queue, selected_db, + proto, cached_clock, exec_publishes, exec_flushes, ++ watched_keys: &HashMap) # v2, was u32 + -> (Frame, Vec<(usize, Bytes)>, Vec<(usize, Vec)>) + Returns (Frame::Null, vec![], vec![]) when any watched version differs — checked BEFORE the + first body command runs, under the same shard slice lock the body commits under. + +Schema: no storage change. Reads Database::get_version(key) per watched key; versions are already +maintained by Database::set(). Watched-key state stays on ConnectionState.watched_keys. ``` -Status: DRAFT - +Status: FROZEN @ v2 — approved by Tin Dang, 2026-08-11. + +### Amendments v1 -> v2 (raised by the build tripwire, not by the author) + +`add.py check` flagged `build_tampered` after the build: §3 as frozen at v1 and the shipped code had +diverged in two places. Recorded here rather than reconciled silently — a frozen contract edited to +match a build is the one move this method forbids, so both are stated with what actually shipped. + +1. **ABA mechanism: per-database DELETE counter -> per-database CREATION ticket.** + v1's freeze resolution said "the cheap fix is a per-database monotonic delete counter consulted + alongside the entry version". That design is unsound for this codebase and was rejected during + build: **expiry is a delete**, so any keyspace with TTLs would bump the epoch continuously and + abort essentially every WATCH transaction — a correctness fix that makes CAS unusable on exactly + the session/cache workloads that need it. Shipped instead: `Database::birth_counter`, a + per-database creation ticket stamped into the entry's existing version field, so a recreated key + is observably a different incarnation. Same guarantee, no TTL interaction, no new storage. + The residual (24-bit wrap, ~1 in 16.7M vs the pre-fix certainty of 1.0) is measured in §7. The + numbers were put in front of the human mid-build, before the mechanism was written. + +2. **Token type: `&HashMap` -> `&HashMap`.** + A newtype over the same `u32`. The wire contract above is byte-for-byte unaffected. It exists so + the residual wrap in (1) has one obvious place to be retired later — a real incarnation field — + without churning every call site that threads the map through the owner hop. + +Unchanged and fully honored: every wire line in the fenced block above, the CROSSSLOT rule, and the +"watched_keys is empty on EVERY outcome" post-condition. + +Least-sure flag surfaced at freeze: [contract/spec] the v2 ABA mechanism leaves a MEASURED residue +rather than eliminating one. The creation ticket shares the entry's 24-bit version field, so it +wraps every 16,777,216 creations — ~18.3s of saturated single-database insert at the measured +914,634 SET/s. A miss needs that wrap to land inside one client's open WATCH..EXEC window AND hit +the one watched key: ~1 in 16.7M, against v1's pre-fix certainty of 1.0. This is the least certain +part of the contract because it is the one place the guarantee is probabilistic instead of total, +and because only a wider `Entry` (an incarnation field) retires it — a change the codebase's +CompactKey/CompactValue size discipline argues against for a 6e-8 residual. If this is judged +unacceptable later it is a change request back to SPECIFY, not a patch: it changes `Entry`'s size. +Second, smaller: [contract] cross-shard watches answer CROSSSLOT, which a client cannot predict +because Moon's shard map is invisible to it — consistent with the MULTI body rule already shipped, +unaffected at `--shards 1`, and mitigated by hash tags, but it must be documented, not discovered. + +Changing anything above this line from here on is a change request back to SPECIFY, not an edit. +The frozen shape neighbours depend on: `execute_transaction_sharded` gains `watched_keys` and +runs the CAS gate before the first body command; `TxnExecutePayload` carries the tokens across +the owner hop; `WatchToken` is the recorded unit on `ConnectionState.watched_keys`. + +**Both flags were surfaced at the freeze and RESOLVED by the human (2026-08-11):** +1. ABA hole → **in scope**, fix here via the delete counter. The `delete-and-recreate is a + conflict` scenario in §2 therefore stays in the red suite as a required test. +2. Cross-shard watch → **`CROSSSLOT`**, reusing `analyze_txn_locality`. No scatter design. + +**Lowest-confidence flag for the freeze — two, both worth an explicit decision:** + +1. **[spec] The ABA hole (§1 ⚠ #2).** `WATCH k` at version 1 → another client DELs and re-creates + k → `EXEC` sees version 1 and commits, where Redis aborts. It exists in the shipped embedded + path today. Shipping this task without closing it means announcing "WATCH works" while a CAS + loop can still silently miss a delete-recreate — the same class of silent-guarantee failure the + task exists to remove. **Recommendation: in scope.** The cheap fix is a per-database monotonic + delete counter consulted alongside the entry version, so a destroyed key can never present a + version it previously held. Cost if deferred instead: one more release where CAS is subtly + wrong, and a scenario above (`delete-and-recreate is a conflict`) must be dropped to red-listed. + +2. **[contract] `CROSSSLOT` for cross-shard watches (§1 ⚠ #1).** Clients cannot see Moon's shard + map, so this error is unpredictable from the client side at `--shards > 1`. It is consistent + with the MULTI/EXEC body rule already shipped, and `--shards 1` is unaffected. The alternative + (cross-shard scatter) cannot be made sound without a global lock. Cost if wrong: multi-key CAS + on multi-shard deployments needs hash tags, and that has to be documented, not discovered. + + --- ## 4 · TESTS — failing-first suite (red) ▸ docs/06-step-4-tests.md -Coverage target: +Coverage target: every Must and every Reject in §1 has one test; parity legs at shards=1 and 4. + Plan (one test per scenario, asserting behavior not internals): - - test_: arrange / act / assert + assert + - test_conflicting_write_aborts_exec: WATCH k / MULTI / SET / conflicting SET from a second + conn / EXEC -> assert Null AND assert k holds the conflicting value (the abort is only real + if the queued write did not land) + - test_clean_exec_commits: same without the conflict -> assert Array[+OK] AND k holds the txn value + - test_watch_on_absent_key_aborts_when_created: assert Null AND the body's side effect is absent + - test_unwatch_releases: assert Array[+OK] AND k holds the txn value + - test_exec_clears_watches_on_both_outcomes: aborted cycle, then a bare MULTI/EXEC -> commits + - test_delete_recreate_is_a_conflict: the ABA scenario -> assert Null AND body side effect absent + - test_paths_agree: replay the conflict sequence at shards=1 and shards=4 -> byte-identical replies + - test_watch_without_keys_is_an_arity_error: assert the ERR text AND that a later EXEC commits + - test_watch_inside_multi_is_refused: assert the ERR text AND that EXEC returns one fewer reply + - test_cross_shard_watch_is_refused: shards=4, keys on different shards -> CROSSSLOT AND no + queued command ran -Tests live in: `./tests/` · MUST run red (missing implementation) before Build. - +Tests live in: `tests/watch_cas_transactions.rs` · MUST run red (missing implementation) before Build. @@ -106,19 +338,50 @@ Tests live in: `./tests/` · MUST run red (missing implementation) before Build. ## 5 · BUILD — AI writes code ▸ docs/07-step-5-build.md -Scope (may touch): `./src/` -Strategy (ordered batches): <1. … 2. … — the planned build order; guidance, not enforced> -Safety rule (feature-specific): -Code lives in: `./src/` -Constraints: do NOT change any test or the contract; allow-list packages only; ask if unclear. - - +Scope (may touch): `src/server/conn/shared.rs` `src/server/conn/watch.rs` `src/server/conn/core.rs` +`src/server/conn/handler_monoio/` `src/server/conn/handler_sharded/` `src/server/conn/handler_single.rs` +`src/shard/spsc_handler.rs` `src/shard/dispatch.rs` `src/shard/coordinator.rs` +`src/storage/db/kv_ops.rs` `src/storage/db/mod.rs` `src/storage/db/accessors.rs` +`tests/watch_cas_transactions.rs` `scripts/test-consistency.sh` `scripts/test-commands.sh` `CHANGELOG.md` + +Scope AMENDED during build (recorded, not quietly widened). The original list was written from §0's +reading that this was a handler-level fix; three of the four defects turned out to live below the +handlers, and the last one below the storage line: + +- `src/shard/dispatch.rs` + `src/shard/coordinator.rs` — defect 2 needs a new `ShardMessage` + (`ReadVersions`) and an owner-grouped snapshot helper. A watched key on another shard cannot be + read from the local slice, and no existing message carried versions. +- `src/storage/db/mod.rs` + `src/storage/db/accessors.rs` — defect 4's creation ticket lives on + `Database`, and all five entry-fabrication sites (`set` plus the four `get_or_create` containers) + must draw from it or the ABA hole stays open for whichever type was missed. +- `src/server/conn/core.rs` — `watched_keys` changes type from `u32` to `WatchToken`. +- `src/server/conn/handler_single.rs` — same type change at its own WATCH arm; NOT a behavior change + (the embedded path was already correct). +- `src/server/conn/watch.rs` (new) — the two production handlers' WATCH arms came out byte-identical + at 57 lines each. Leaving two copies in the task whose subject IS those paths drifting apart would + re-plant the defect, so both call one module. + +`src/command/metadata.rs` was declared and NOT touched: WATCH/UNWATCH were already registered with +correct arity and the `transaction` ACL category. The defect was never in the metadata table — which +is precisely why `ACL CAT transaction` listed both commands while neither actually guarded anything. + +Strategy (ordered batches): +1. Red suite first — `tests/watch_cas_transactions.rs`, every scenario, failing for the right reason. +2. `execute_transaction_sharded` gains `watched_keys` + the pre-body CAS check; all three call + sites pass it (monoio, sharded, spsc_handler). +3. WATCH/UNWATCH command arms on both production handlers AND the inline path — all three, per the + three-dispatch-paths rule; missing one is CI-invisible. +4. Cross-shard watch classification reusing `analyze_txn_locality` rather than a second rule. +5. ABA fix (pending the §3 freeze decision): per-database delete counter consulted with the version. +6. Consistency/command script entries; CHANGELOG. + +Safety rule (feature-specific): the CAS check must run under the SAME shard-slice lock acquisition +that commits the body — a version read that releases the lock before the body runs re-introduces +the TOCTOU this design rejected in §1. + +Code lives in: `src/` +Constraints: do NOT change any test or the contract; no global locks; no allocation added to the +non-transaction hot path (the watch check must be skipped entirely when `watched_keys` is empty). --- @@ -127,43 +390,104 @@ Constraints: do NOT change any test or the contract; allow-list packages only; a - [ ] all tests pass - [ ] coverage did not decrease - [ ] no test or contract was altered during build -- [ ] the green was EARNED, not gamed — no overfit to fixtures, vacuous asserts, or stubbed-away logic (score with an adversarial refute-read — a subagent recommended under `autonomy: auto`; a confirmed cheat is HARD-STOP) +- [ ] the green was EARNED — adversarial refute-read; a confirmed cheat is HARD-STOP - [ ] concurrency / timing of the risky operation is safe - [ ] no exposed secrets, injection openings, or unexpected dependencies - [ ] layering & dependencies follow CONVENTIONS.md - [ ] a person reviewed and approved the change -### Build expectations — what "correct" looks like (fill BEFORE build; confirm each at the gate) -> Pre-declare the OBSERVABLE outcomes a correct build must produce — derived from §2 SCENARIOS -> + §3 CONTRACT — so this gate checks the build is RIGHT, not merely that tests are green. Each -> row is evidence you can SEE, not a restatement of a test name. -- [ ] — confirmed by -- [ ] — confirmed by +### Build expectations — what "correct" looks like +- [x] The §0 two-connection probe, replayed against the built binary, reports `EXEC -> Null` and + `GET k -> from-B` — the exact inverse of the measured pre-fix result. `tmp/probe_watch.py` + @151a1857: `A EXEC -> $-1`, `final GET k -> from-B`. VERDICT: CAS honored. +- [x] `WATCH`/`UNWATCH` reply `+OK` on RESP shards=1, RESP shards=4, and the inline path — + `tmp/probe_watch2.py` @shards=4: `WATCH k`, `UNWATCH`, `WATCH a b`, `inline WATCH k` all `+OK`. +- [x] `ACL CAT transaction` still lists watch/unwatch, and both are now dispatchable — + `tmp/probe_watch3.py`: 7 entries incl. `watch`, `unwatch`. (`COMMAND INFO/COUNT` still reply + `*0` — the §0 out-of-scope defect, unchanged, owned by `info-observability`.) +- [x] Non-transaction throughput unchanged — interleaved A/B on moon-dev (aarch64 Linux), fat-LTO + release both legs, 6 alternating rounds, 1M req `-c 50 -P 16` after a 200k warm: + + | leg | before (median) | after (median) | delta | + |---|---|---|---| + | SET (under test) | 1,648,994/s | 1,626,743/s | -1.35% | + | GET (control, untouched by the change) | 3,311,404/s | 3,273,401/s | -1.15% | + + Worst within-leg CV 7.7%; best 3.1%. The untouched control moved essentially as much as the + leg under test (0.2pp apart), so both deltas are drift, not signal. A first pass at `-P 1` + was DISCARDED as uninformative: 13.9% noise floor with the control moving MORE (-3.9%) than + SET (-0.8%) — recorded because "we benched it" is worthless without the noise floor beside it. +- [ ] The full matrix is green via `gh workflow run CI --ref ` BEFORE merge, per the + standing merge bar — Windows/macOS/console are skipped on PRs. + +### Durability (kill-9) leg +Run against the fat-LTO `target/release/moon`: **22 passed, 1 failed**. The failure, +`durability::backup_restore::tests::backup_restore_parity`, is PRE-EXISTING ROT, not a regression — +three independent proofs: (a) BGSAVE reports `rdb_last_bgsave_status:ok` and writes +`shard-0/shard-0.rrdshard`; (b) `dump.rdb`, the path the test asserts, is never written by the +snapshot writer — it survives only as the `--dbfilename` default and a data-dir marker string; +(c) this commit touches no persistence/rdb/snapshot file at all. The test last changed in +`24ee60eb` (v0.1.3, #65), predating the per-shard snapshot layout. + +Two defects found in that suite, both OUT OF SCOPE here, both filed rather than fixed inline: +1. `tests/durability/*` hardcode `Command::new("./target/release/moon")`, ignoring `MOON_BIN` — the + suite silently tests whatever binary is lying at that path (the local one was 3 days stale, which + produced 7 bogus "connection refused" failures before the binary was rebuilt). +2. `backup_restore_parity` asserts a filename the server no longer produces. Both are `#[ignore]`d, + so CI never runs them — which is exactly how a durability gate rots into proving nothing. -### Deep checks — do not skim (fill the path that applies; the resolver judges which) -- [ ] WIRING (code) — every new symbol is referenced; record where / how confirmed -- [ ] DEAD-CODE (code) — no new unused or orphaned symbol introduced -- [ ] SEMANTIC (prose / non-code) — read in full, not skimmed: +## 7 · OBSERVE — feed the next loop ▸ docs/09-the-loop.md -### GATE RECORD -Outcome: -If RISK-ACCEPTED -> owner: · ticket: · expires: (never for a security gap) -Reviewed by: · date: +Watch (reuse scenarios as monitors): EXEC abort rate (a CAS-heavy client whose abort rate falls to +zero has stopped being guarded — the pre-fix signature); `moon_dispatch_path_total` split, so a +WATCH regression on one production path cannot hide behind the other being correct. - +### Residual risk accepted at build time ---- +The ABA fix stamps each created entry from a per-database creation ticket +(`Database::birth_counter`), which shares the entry's 24-bit version field and therefore wraps at +16,777,216 creations. Measured insert rate on this build (`redis-benchmark SET -P16 -r 10M`, +shards=1, appendonly=no): **914,634/s** — so the counter wraps every **~18.3s** of saturated +single-database insert. -## 7 · OBSERVE — feed the next loop ▸ docs/09-the-loop.md +| mechanism | miss probability per delete+recreate inside a WATCH window | +|---|---| +| pre-fix (every creation at `INITIAL_VERSION`) | 1.0 — certain, every time | +| per-db creation ticket (shipped) | 5.96e-08 (~1 in 16,777,216) | + +A miss now requires the wrap to land inside one client's open `WATCH`..`EXEC` window *and* to hit +the one watched key. Only a true incarnation field (a wider `Entry`, which the codebase's +`CompactKey`/`CompactValue` size discipline argues against) removes the residue entirely; that is +why `WatchToken` stays a named struct rather than a bare `u32` — adding the field later does not +churn the call sites. Numbers reported to Tin Dang before batch 5 was built; decision was to ship +the ticket and record the residue here. -Watch (reuse scenarios as monitors): +Rejected alternative: a per-db **delete** epoch mixed into the token. Expiry is a delete, so any +TTL'd keyspace would bump the epoch continuously and abort essentially every WATCH transaction — +a correctness fix that makes CAS unusable on exactly the workloads that need it. ### Spec delta -Forward changes for the next loop — each re-enters at Specify as the next task. One line -each, tagged `[SPEC · open|seeded|dropped]`, with evidence (e.g. `[SPEC · open] rate-limit -the retry path (evidence: prod herd spikes)`). See the `add` skill's `deltas.md`. +- [SPEC · open] `WATCH` on a key that later moves shard (cluster resharding) is not modelled; + `snapshot_versions` reads the owner at WATCH time and EXEC re-reads the owner at commit time, + so a slot migration between the two silently compares different shards' answers (evidence: the + cross-shard path was built for a static shard map — see §1's CROSSSLOT rule). +- [SPEC · open] a dead owner shard yields version `0` from `snapshot_versions`, which fails + *toward aborting*; the abort is correct but indistinguishable from "key absent" in logs + (evidence: build batch 3). +- [SPEC · open] with disk-offload enabled (opt-in), a watched key that is spilled cold and then + promoted comes back through `Database::set`, drawing a fresh creation ticket — so an eviction + the client never asked for aborts its transaction. Not a regression (promotion previously + returned `INITIAL_VERSION`, which also mismatched any version above 1, and *matched* it when the + watched version happened to be 1 — i.e. the old behavior was spurious-abort OR wrong-commit, + and this is spurious-abort only), and it fails in the safe direction, but a CAS loop on a + memory-pressured keyspace can now livelock on eviction rather than on contention. Fixing it + means the promoted entry inheriting its pre-spill version, which means persisting versions + (evidence: `promote_inflight_if_present` and `promote_cold_outcome` both route through `set`). ### Competency deltas -What did this loop teach the foundation? One line each, tagged by competency -(`DDD · SDD · UDD · TDD · ADD`), status `open`, with evidence. See the `add` skill's `deltas.md`. - +- [TDD · open] `wc7_all_dispatch_paths_agree` passed BEFORE the fix because both production paths + were equally broken — agreement between two wrong answers is not evidence. A parity test needs a + companion absolute assertion or it certifies nothing (evidence: wc7 green on the red run). +- [ADD · open] batch 5's mechanism was chosen from a measured wrap rate rather than an estimate, + and the measurement changed nothing about the choice but everything about what got written down + (evidence: this section). diff --git a/.add/tooling/add.py b/.add/tooling/add.py index fdc916414..df4fe708a 100644 --- a/.add/tooling/add.py +++ b/.add/tooling/add.py @@ -1722,6 +1722,23 @@ def _missing_captures(root: Path) -> list[str]: if not any((cap_dir / f"{n}.{ext}").is_file() for ext in _CAPTURE_EXTS)] +def _contract_status(root: Path, slug: str) -> str | None: + """First word of the task's `Status:` line, upper-cased, emphasis stripped. + + `**FROZEN @ v1**` and `FROZEN @ v1` are the same state; only the marker + differs. Returns None when the task declares no Status line at all. + """ + task_md = root / "tasks" / slug / "TASK.md" + try: + for line in task_md.read_text(encoding="utf-8").splitlines(): + if line.startswith("Status:"): + body = line[len("Status:"):].strip().lstrip("*").strip() + return body.split()[0].rstrip("*").upper() if body.split() else "" + except OSError: + return None + return None + + def cmd_check(args: argparse.Namespace) -> None: """Read-only integrity check of the .add project. Exit 1 if anything fails.""" as_json = getattr(args, "json", False) @@ -1750,6 +1767,17 @@ def cmd_check(args: argparse.Namespace) -> None: marker, want = _read_task_phase(root, slug), t.get("phase") checks.append((marker == want, f"task '{slug}' marker matches state", f"marker={marker!r} state={want!r}")) + # phase/Status drift: this engine has no `freeze` subcommand, so §3 + # `Status:` is hand-maintained and can silently lag the phase marker — + # a task can sit at `build` with its contract still DRAFT, i.e. built + # against an unapproved shape. Only DRAFT-past-contract is red; + # `INHERITED` (contract adopted from a sibling task) is a real state, + # and emphasis markers are cosmetic. + if want in PHASES[PHASES.index("tests"):]: + _cstat = _contract_status(root, slug) + checks.append((_cstat in ("FROZEN", "INHERITED"), + f"task '{slug}' contract is frozen for phase '{want}'", + f"Status is {_cstat or 'absent'!s} — freeze §3 before leaving contract")) # drift: milestone + dependency references must resolve ms = t.get("milestone") if ms is not None: diff --git a/CHANGELOG.md b/CHANGELOG.md index abd5b54bb..476543060 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -69,6 +69,52 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `--shards 1` and `--shards 4`). ### Fixed +- **`WATCH` / `UNWATCH` now actually guard a transaction on the production + dispatch paths.** Both commands parsed and answered `+OK`, and the tokio and + embedded handlers re-checked the recorded versions at `EXEC` — but the two + paths clients really reach (`handler_monoio`, `handler_sharded`) never + consulted the watch set at all. A conflicting write from another client + committed anyway, so every check-and-set built on `WATCH` (inventory + decrement, balance transfer, leader election) silently degraded to + last-writer-wins. Four defects, all fixed together because a partial fix is + indistinguishable from none: + + 1. **The watch set was not consulted at `EXEC`** on the monoio and sharded + handlers. `EXEC` now aborts with a RESP null array on any version + mismatch, and clears the watch set on *both* outcomes by construction + (`mem::take`) rather than by a `clear()` each exit path has to remember. + 2. **A watched key on another shard was read from the local slice** — a + different database entirely, so the version compared was some unrelated + key's or zero. `WATCH` now snapshots versions where the keys live, via a + new `ShardMessage::ReadVersions` (one hop per owning shard, not per key), + and a watch set spanning shards is classified and refused `CROSSSLOT` + like a cross-shard `MULTI` body already was. + 3. **`WATCH` inside `MULTI` was queued** as an ordinary command instead of + being refused, and `WATCH` with no arguments answered "unknown command" + instead of an arity error. + 4. **Delete + recreate was invisible (ABA).** Versions are per-entry and die + with the entry, so every incarnation of a key started at + `INITIAL_VERSION`: `DEL k` + `SET k` handed the watcher back the exact + token it had recorded and `EXEC` committed on a key that had been + destroyed and rebuilt underneath it — where Redis aborts. Entries are now + stamped from a per-database creation ticket, so a recreated key is + observably a different incarnation. Restored keys draw tickets too: + versions are not persisted, so otherwise the first key created after a + restart would collide with the whole restored population. + + Residual risk, stated rather than buried: the ticket shares the entry's + 24-bit version field and so wraps at 16,777,216 creations (~18s of saturated + single-database insert at the measured 914K/s). A miss now needs that wrap to + land inside one client's open `WATCH`..`EXEC` window *and* hit the one + watched key — ~1 in 16.7M, against the pre-fix certainty. Only a wider + incarnation field removes it entirely; `WatchToken` stays a named struct so + adding one later does not churn the call sites. + + New suite `tests/watch_cas_transactions.rs` (10 wire-level tests, raw-byte + assertions, two connections, run at both `--shards 1` and `--shards 4`), + plus WATCH/CAS entries in `scripts/test-consistency.sh` and + `scripts/test-commands.sh`. + - **RESP3 reply types now match Redis, and no longer change with the calling context.** A live sweep against `redis-server` 8.6.1 found the conversion table wrong in both directions and, structurally, unable to be right: it keyed diff --git a/scripts/test-commands.sh b/scripts/test-commands.sh index bba560f2c..a105ae0d4 100755 --- a/scripts/test-commands.sh +++ b/scripts/test-commands.sh @@ -720,6 +720,57 @@ if should_run "transaction"; then FAIL=$((FAIL + 1)) echo " FAIL: MULTI/DISCARD" fi + + # --- WATCH / UNWATCH optimistic locking ------------------------------- + # + # A CAS conflict needs TWO connections interleaved: the transaction must + # stay open while a second client writes the watched key. `redis-cli` + # one-shot mode cannot express that (each invocation is its own connection, + # closed on exit), so bash's /dev/tcp holds the transaction connection open + # and drives it with inline commands. The verdict is read from the key's + # FINAL VALUE, not from EXEC's reply, which keeps this free of RESP parsing. + watch_cas_outcome() { + local port="$1" conflict="$2" line="" + redis-cli -p "$port" SET cas:k base >/dev/null 2>&1 || true + exec 3<>"/dev/tcp/127.0.0.1/${port}" || { echo "__CONNECT_FAILED__"; return 0; } + printf 'WATCH cas:k\r\nMULTI\r\nSET cas:k from-txn\r\n' >&3 + if [[ "$conflict" == "yes" ]]; then + redis-cli -p "$port" SET cas:k from-other >/dev/null 2>&1 || true + fi + # ECHO after EXEC is a round-trip barrier: reading its reply proves EXEC + # was applied before the connection closes, so the GET cannot race it. + printf 'EXEC\r\nECHO cas-done\r\n' >&3 + while IFS= read -r -t 5 line <&3; do + [[ "${line%$'\r'}" == "cas-done" ]] && break + done + exec 3>&- + redis-cli -p "$port" GET cas:k 2>/dev/null + } + + TOTAL=$((TOTAL + 1)) + cas_moon=$(watch_cas_outcome "$PORT_RUST" yes) + if [[ "$cas_moon" == "from-other" ]]; then + PASS=$((PASS + 1)) + else + FAIL=$((FAIL + 1)) + echo " FAIL: WATCH conflicting write did not abort EXEC" + echo " EXPECTED: from-other (transaction aborted)" + echo " GOT: $cas_moon" + fi + + TOTAL=$((TOTAL + 1)) + cas_clean=$(watch_cas_outcome "$PORT_RUST" no) + if [[ "$cas_clean" == "from-txn" ]]; then + PASS=$((PASS + 1)) + else + FAIL=$((FAIL + 1)) + echo " FAIL: unconflicted WATCH/EXEC did not commit" + echo " EXPECTED: from-txn" + echo " GOT: $cas_clean" + fi + + assert_moon_contains "WATCH arity error" "wrong number of arguments" WATCH + assert_moon "UNWATCH outside MULTI" "OK" UNWATCH fi # =========================================================================== diff --git a/scripts/test-consistency.sh b/scripts/test-consistency.sh index d37ed6d09..40a715e26 100755 --- a/scripts/test-consistency.sh +++ b/scripts/test-consistency.sh @@ -578,6 +578,88 @@ both SET edge:touch "val" assert_both "TOUCH" TOUCH edge:touch assert_both "TOUCH missing" TOUCH edge:nomiss +# =========================================================================== +# WATCH / UNWATCH optimistic locking (CAS) +# =========================================================================== +log "=== WATCH/CAS ===" + +# A CAS conflict needs TWO connections interleaved: the transaction has to stay +# open while a second client writes the watched key. `redis-cli` one-shot mode +# cannot express that (each invocation is its own connection, closed on exit), +# so bash's /dev/tcp holds the transaction connection open and drives it with +# inline commands. The verdict is read from the key's FINAL VALUE rather than +# from EXEC's reply, which keeps this free of RESP parsing: `from-txn` means the +# transaction committed, `from-other` means it aborted and the interloper's +# write stands. +# +# Both servers run the identical sequence and the outcomes are compared, so this +# asserts Redis parity rather than a hardcoded expectation. +watch_cas_outcome() { + local port="$1" conflict="$2" line="" + redis-cli -p "$port" SET cas:k base >/dev/null 2>&1 || true + exec 3<>"/dev/tcp/127.0.0.1/${port}" || { echo "__CONNECT_FAILED__"; return 0; } + printf 'WATCH cas:k\r\nMULTI\r\nSET cas:k from-txn\r\n' >&3 + if [[ "$conflict" == "yes" ]]; then + redis-cli -p "$port" SET cas:k from-other >/dev/null 2>&1 || true + fi + # ECHO after EXEC is a round-trip barrier: reading its reply proves EXEC has + # been applied before the connection closes, so the GET below cannot race it. + printf 'EXEC\r\nECHO cas-done\r\n' >&3 + while IFS= read -r -t 5 line <&3; do + [[ "${line%$'\r'}" == "cas-done" ]] && break + done + exec 3>&- + redis-cli -p "$port" GET cas:k 2>&1 +} + +assert_eq "WATCH: conflicting write aborts EXEC" \ + "$(watch_cas_outcome "$PORT_REDIS" yes)" "$(watch_cas_outcome "$PORT_RUST" yes)" +assert_eq "WATCH: unconflicted EXEC commits" \ + "$(watch_cas_outcome "$PORT_REDIS" no)" "$(watch_cas_outcome "$PORT_RUST" no)" + +# The ABA hole: versions are per-entry and die with the entry, so before the +# per-db creation ticket a DEL + re-SET handed the watcher back the exact token +# WATCH had recorded and EXEC committed on a key that had been destroyed and +# rebuilt underneath it. +watch_cas_aba_outcome() { + local port="$1" line="" + redis-cli -p "$port" SET aba:k base >/dev/null 2>&1 || true + exec 3<>"/dev/tcp/127.0.0.1/${port}" || { echo "__CONNECT_FAILED__"; return 0; } + printf 'WATCH aba:k\r\nMULTI\r\nSET aba:k from-txn\r\n' >&3 + redis-cli -p "$port" DEL aba:k >/dev/null 2>&1 || true + redis-cli -p "$port" SET aba:k rebuilt >/dev/null 2>&1 || true + printf 'EXEC\r\nECHO cas-done\r\n' >&3 + while IFS= read -r -t 5 line <&3; do + [[ "${line%$'\r'}" == "cas-done" ]] && break + done + exec 3>&- + redis-cli -p "$port" GET aba:k 2>&1 +} + +assert_eq "WATCH: delete + recreate aborts EXEC (ABA)" \ + "$(watch_cas_aba_outcome "$PORT_REDIS")" "$(watch_cas_aba_outcome "$PORT_RUST")" + +# UNWATCH releases every dependency, so the same conflicting write commits. +watch_unwatch_outcome() { + local port="$1" line="" + redis-cli -p "$port" SET uw:k base >/dev/null 2>&1 || true + exec 3<>"/dev/tcp/127.0.0.1/${port}" || { echo "__CONNECT_FAILED__"; return 0; } + printf 'WATCH uw:k\r\nUNWATCH\r\nMULTI\r\nSET uw:k from-txn\r\n' >&3 + redis-cli -p "$port" SET uw:k from-other >/dev/null 2>&1 || true + printf 'EXEC\r\nECHO cas-done\r\n' >&3 + while IFS= read -r -t 5 line <&3; do + [[ "${line%$'\r'}" == "cas-done" ]] && break + done + exec 3>&- + redis-cli -p "$port" GET uw:k 2>&1 +} + +assert_eq "UNWATCH releases the dependency" \ + "$(watch_unwatch_outcome "$PORT_REDIS")" "$(watch_unwatch_outcome "$PORT_RUST")" + +assert_both "WATCH arity" WATCH +assert_both "UNWATCH outside MULTI" UNWATCH + # =========================================================================== # SWAPDB consistency # =========================================================================== diff --git a/src/server/conn/core.rs b/src/server/conn/core.rs index 9344eb0b7..44b6b981f 100644 --- a/src/server/conn/core.rs +++ b/src/server/conn/core.rs @@ -277,9 +277,11 @@ pub(crate) struct ConnectionState { pub tracking_state: TrackingState, pub tracking_rx: Option>, - // WATCH/EXEC optimistic locking (handler_single only) - #[allow(dead_code)] // Only used by handler_single (tokio feature) - pub watched_keys: HashMap, + // WATCH/EXEC optimistic locking. Read by all three dispatch paths — the + // `handler_single only` note and its dead_code allow were accurate right up + // until they described the bug: the two production handlers parsed WATCH, + // answered +OK, and never looked at this map again. + pub watched_keys: HashMap, // Connection affinity (migration) pub affinity_tracker: Option, diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 8fe6f3f7c..929877385 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1700,6 +1700,7 @@ pub(crate) async fn handle_connection_sharded_monoio< let mut exec_publishes: Vec<(usize, Bytes, Bytes)> = Vec::new(); if write::try_handle_multi_exec( cmd, + cmd_args, &mut conn, ctx, &mut responses, diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index 280ea5d0a..c488a2266 100644 --- a/src/server/conn/handler_monoio/write.rs +++ b/src/server/conn/handler_monoio/write.rs @@ -706,11 +706,20 @@ async fn mq_hop_or_local( /// `--shards 1` under the monoio TopLevel writer). pub(super) async fn try_handle_multi_exec( cmd: &[u8], + args: &[Frame], conn: &mut ConnectionState, ctx: &ConnectionContext, responses: &mut Vec, exec_publishes: &mut Vec<(usize, Bytes, Bytes)>, ) -> bool { + // --- WATCH / UNWATCH --- + // Before the MULTI queueing step below, so `WATCH` inside MULTI is refused + // rather than queued. Shared with the other production handler on purpose: + // two copies of this arm is how the paths drifted in the first place. + if crate::server::conn::watch::try_handle_watch_unwatch(cmd, args, conn, ctx, responses).await { + return true; + } + // --- MULTI --- if cmd.eq_ignore_ascii_case(b"MULTI") { if conn.in_cross_txn() { @@ -733,6 +742,10 @@ pub(super) async fn try_handle_multi_exec( responses.push(Frame::Error(Bytes::from_static(b"ERR EXEC without MULTI"))); } else { conn.in_multi = false; + // Taken, not borrowed: EXEC must clear its watches on BOTH the + // committed and the aborted outcome, and a stale watch surviving + // an abort is how a CAS retry loop livelocks. + let watched = std::mem::take(&mut conn.watched_keys); // The body runs on THIS shard with no per-key routing, so a // foreign-owned key would be silently misplaced. Classify locality: // - CrossShard: genuinely spans shards — a shared-nothing engine @@ -742,9 +755,15 @@ pub(super) async fn try_handle_multi_exec( // on the owner (instead of the Phase-A CROSSSLOT rejection). // - Keyless / SingleShard(self): fall through to local execution. if ctx.num_shards > 1 { - match crate::server::conn::shared::analyze_txn_locality( - &conn.command_queue, - ctx.num_shards, + match crate::server::conn::shared::merge_locality( + crate::server::conn::shared::analyze_txn_locality( + &conn.command_queue, + ctx.num_shards, + ), + // A watched key owned by another shard cannot be validated + // where the body commits — refuse rather than fabricate a + // conflict the client can never clear. + crate::server::conn::shared::analyze_watch_locality(&watched, ctx.num_shards), ) { crate::server::conn::shared::TxnLocality::SingleShard(s) if s != ctx.shard_id => @@ -764,6 +783,11 @@ pub(super) async fn try_handle_multi_exec( conn.selected_db, commands, conn.protocol_version, + // The CAS check runs where the body runs. Cloned + // rather than borrowed because the payload crosses + // an SPSC hop and must own its tokens; the map is + // empty for every transaction that did not WATCH. + watched.clone(), &ctx.dispatch_tx, &ctx.spsc_notifiers, ) @@ -867,6 +891,7 @@ pub(super) async fn try_handle_multi_exec( &ctx.cached_clock, exec_publishes, &mut exec_flushes, + &watched, ); // v0.7 REPLICATION (adversarial-review P0-1): the txn body must // reach replicas like any other successful local write. This was @@ -996,6 +1021,7 @@ pub(super) async fn try_handle_multi_exec( } else { conn.in_multi = false; conn.command_queue.clear(); + conn.watched_keys.clear(); responses.push(Frame::SimpleString(Bytes::from_static(b"OK"))); } return true; diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index e2a48afa1..943d5acc3 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -1100,7 +1100,7 @@ pub(crate) async fn handle_connection_sharded_inner< // --- MULTI / EXEC_CMD / DISCARD --- let mut exec_publishes: Vec<(usize, Bytes, Bytes)> = Vec::new(); - if write::try_handle_multi_exec(cmd, &mut conn, ctx, &mut responses, &mut exec_publishes).await { + if write::try_handle_multi_exec(cmd, cmd_args, &mut conn, ctx, &mut responses, &mut exec_publishes).await { // C2: PUBLISH queued inside MULTI fans out only now — after the // transaction body has been applied — and its placeholder in the // EXEC reply array is patched with the real receiver count. diff --git a/src/server/conn/handler_sharded/write.rs b/src/server/conn/handler_sharded/write.rs index 623ac1d4e..9553949b2 100644 --- a/src/server/conn/handler_sharded/write.rs +++ b/src/server/conn/handler_sharded/write.rs @@ -531,11 +531,20 @@ pub(super) async fn try_handle_mq_command( /// nothing, so every transactional write was lost on restart). pub(super) async fn try_handle_multi_exec( cmd: &[u8], + args: &[Frame], conn: &mut ConnectionState, ctx: &ConnectionContext, responses: &mut Vec, exec_publishes: &mut Vec<(usize, Bytes, Bytes)>, ) -> bool { + // --- WATCH / UNWATCH --- + // Before the MULTI queueing step below, so `WATCH` inside MULTI is refused + // rather than queued. Shared with the other production handler on purpose: + // two copies of this arm is how the paths drifted in the first place. + if crate::server::conn::watch::try_handle_watch_unwatch(cmd, args, conn, ctx, responses).await { + return true; + } + // --- MULTI --- if cmd.eq_ignore_ascii_case(b"MULTI") { if conn.in_cross_txn() { @@ -558,6 +567,10 @@ pub(super) async fn try_handle_multi_exec( responses.push(Frame::Error(Bytes::from_static(b"ERR EXEC without MULTI"))); } else { conn.in_multi = false; + // Taken, not borrowed: EXEC must clear its watches on BOTH the + // committed and the aborted outcome, and a stale watch surviving + // an abort is how a CAS retry loop livelocks. + let watched = std::mem::take(&mut conn.watched_keys); // The body runs on THIS shard with no per-key routing, so a // foreign-owned key would be silently misplaced. Classify locality: // - CrossShard: genuinely spans shards — a shared-nothing engine @@ -567,9 +580,15 @@ pub(super) async fn try_handle_multi_exec( // on the owner (instead of the Phase-A CROSSSLOT rejection). // - Keyless / SingleShard(self): fall through to local execution. if ctx.num_shards > 1 { - match crate::server::conn::shared::analyze_txn_locality( - &conn.command_queue, - ctx.num_shards, + match crate::server::conn::shared::merge_locality( + crate::server::conn::shared::analyze_txn_locality( + &conn.command_queue, + ctx.num_shards, + ), + // A watched key owned by another shard cannot be validated + // where the body commits — refuse rather than fabricate a + // conflict the client can never clear. + crate::server::conn::shared::analyze_watch_locality(&watched, ctx.num_shards), ) { crate::server::conn::shared::TxnLocality::SingleShard(s) if s != ctx.shard_id => @@ -589,6 +608,8 @@ pub(super) async fn try_handle_multi_exec( conn.selected_db, commands, conn.protocol_version, + // The CAS check runs where the body runs. + watched.clone(), &ctx.dispatch_tx, &ctx.spsc_notifiers, ) @@ -692,6 +713,7 @@ pub(super) async fn try_handle_multi_exec( &ctx.cached_clock, exec_publishes, &mut exec_flushes, + &watched, ); // task #52: flush the graph-leg wal-v3 records collected by the // txn executor. Replication is monoio-only by design (see @@ -777,6 +799,7 @@ pub(super) async fn try_handle_multi_exec( } else { conn.in_multi = false; conn.command_queue.clear(); + conn.watched_keys.clear(); responses.push(Frame::SimpleString(Bytes::from_static(b"OK"))); } return true; diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index c0c9bf97f..83f9cc9c4 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -1569,7 +1569,12 @@ pub async fn handle_connection( for arg in cmd_args { if let Frame::BulkString(key) = arg { let version = guard.get_version(key); - conn.watched_keys.insert(key.clone(), version); + conn.watched_keys.insert( + key.clone(), + crate::server::conn::shared::WatchToken { + version, + }, + ); } } // guard dropped here diff --git a/src/server/conn/mod.rs b/src/server/conn/mod.rs index c3e40d8bd..6b9a5fd08 100644 --- a/src/server/conn/mod.rs +++ b/src/server/conn/mod.rs @@ -12,6 +12,7 @@ pub mod shared; #[cfg(all(test, feature = "runtime-monoio"))] mod tests; pub mod util; +pub mod watch; // Re-export for internal use #[allow(unused_imports)] diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index 1e58d32b6..00809eef4 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -1,4 +1,5 @@ -#[cfg(feature = "runtime-tokio")] +// Both runtimes need this now: `execute_transaction_sharded` carries the +// WATCH token map, and that path is the one the monoio build ships. use std::collections::HashMap; use std::sync::Arc; @@ -112,6 +113,35 @@ pub(crate) fn handle_config( } } +/// What `WATCH k` recorded about `k`, and what `EXEC` re-checks. +/// +/// `version` is `Database::get_version`, which is `0` exactly when the key is +/// absent (creation tickets start at 1 and the counter never yields 0), so +/// absent-vs-present needs no separate flag. +/// +/// A struct rather than a bare `u32`: the ABA hole (delete + recreate handing +/// back the token WATCH recorded) is closed inside this one field by the per-db +/// creation ticket — see `Database::birth_counter` — but that ticket shares the +/// entry's 24 version bits and so still wraps at 16.7M creations. A true +/// incarnation field is the only way to retire the residue, and it would live +/// here; keeping the named type means adding it never churns the call sites. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct WatchToken { + pub version: u32, +} + +/// True when any watched key's version no longer matches what WATCH recorded. +/// +/// Caller must have checked `!watched.is_empty()` — this takes the shard slice, +/// which is not free, and the empty case is the overwhelming majority. +fn watch_conflict(db_index: usize, watched: &HashMap) -> bool { + crate::shard::slice::with_shard_db(db_index, |db| { + watched + .iter() + .any(|(key, tok)| db.get_version(key) != tok.version) + }) +} + /// Execute a queued transaction atomically under a single database lock. /// /// Checks WATCH versions first -- if any watched key's version has changed since @@ -123,7 +153,7 @@ pub(crate) fn handle_config( pub(crate) fn execute_transaction( db: &SharedDatabases, command_queue: &[Frame], - watched_keys: &HashMap, + watched_keys: &HashMap, selected_db: &mut usize, exec_publishes: &mut Vec<(usize, Bytes, Bytes)>, ) -> (Frame, Vec) { @@ -134,7 +164,7 @@ pub(crate) fn execute_transaction( // Check WATCH versions -- if any key's version changed, abort for (key, watched_version) in watched_keys { let current_version = guard.get_version(key); - if current_version != *watched_version { + if current_version != watched_version.version { return (Frame::Null, Vec::new()); // Transaction aborted } } @@ -232,9 +262,24 @@ pub(crate) fn execute_transaction_sharded( cached_clock: &CachedClock, exec_publishes: &mut Vec<(usize, Bytes, Bytes)>, exec_flushes: &mut Vec<(usize, Frame, usize)>, + // WATCH/CAS (task `watch-cas-transactions`): the versions this connection + // recorded at WATCH time. Empty for the overwhelming majority of + // transactions, which is why the check below early-outs on `is_empty` + // before touching the shard at all. + watched_keys: &HashMap, ) -> (Frame, Vec<(usize, Bytes)>, Vec<(usize, Vec)>) { let db_count = shard_databases.db_count(); + // The CAS gate. This runs synchronously on the shard thread, BEFORE the + // first body command and with no `.await` between it and the body — that + // adjacency is the whole guarantee. `execute_transaction` (the embedded + // path) does the same thing at the top of its locked section; this path + // had no equivalent, so a transaction that declared a dependency on a key + // committed straight over a conflicting write. + if !watched_keys.is_empty() && watch_conflict(selected_db, watched_keys) { + return (Frame::Null, Vec::new(), Vec::new()); + } + let mut results = Vec::with_capacity(command_queue.len()); // Per-entry db (PR #282 review): this executor re-dispatches each body // command against the CURRENT `selected`, so a SELECT queued inside MULTI @@ -919,6 +964,52 @@ pub(crate) enum TxnLocality { CrossShard, } +/// Classify the WATCHed keys by the shard(s) they hash to. +/// +/// Same lattice as the body's: no keys is `Keyless`, all on one shard is +/// `SingleShard`, anything else is `CrossShard`. +pub(crate) fn analyze_watch_locality( + watched: &HashMap, + num_shards: usize, +) -> TxnLocality { + let mut owner: Option = None; + for key in watched.keys() { + let s = crate::shard::dispatch::key_to_shard(key, num_shards); + match owner { + None => owner = Some(s), + Some(existing) if existing == s => {} + Some(_) => return TxnLocality::CrossShard, + } + } + match owner { + None => TxnLocality::Keyless, + Some(s) => TxnLocality::SingleShard(s), + } +} + +/// Combine the body's locality with the WATCH set's. +/// +/// The CAS check runs where the body runs, reading that shard's slice — so a +/// watched key owned by a DIFFERENT shard reads version 0 there and fabricates +/// a conflict. That is safe (it aborts) but wrong, and a retry loop that can +/// never succeed is a livelock. Refusing loudly is the contract. +/// +/// `Keyless` is the identity: a keyless body inherits the watch set's shard, +/// and an unwatched body inherits the body's. +pub(crate) fn merge_locality(body: TxnLocality, watch: TxnLocality) -> TxnLocality { + match (body, watch) { + (TxnLocality::CrossShard, _) | (_, TxnLocality::CrossShard) => TxnLocality::CrossShard, + (TxnLocality::Keyless, other) | (other, TxnLocality::Keyless) => other, + (TxnLocality::SingleShard(a), TxnLocality::SingleShard(b)) => { + if a == b { + TxnLocality::SingleShard(a) + } else { + TxnLocality::CrossShard + } + } + } +} + /// Classify a queued transaction body by the shard(s) its keys hash to. /// /// Uses the command-metadata key specs (`command_keys`) so multi-key commands @@ -1326,3 +1417,129 @@ mod txn_locality_tests { assert_eq!(analyze_txn_locality(&q, num), TxnLocality::CrossShard); } } + +#[cfg(test)] +mod watch_locality_tests { + use super::{TxnLocality, WatchToken, analyze_watch_locality, merge_locality}; + use bytes::Bytes; + use std::collections::HashMap; + + fn watched(keys: &[&str]) -> HashMap { + keys.iter() + .map(|k| { + ( + Bytes::copy_from_slice(k.as_bytes()), + WatchToken { version: 1 }, + ) + }) + .collect() + } + + /// Find two keys that hash to different shards, so the cross-shard case is + /// asserted against the real hash rather than an assumed key layout. + fn cross_shard_pair(num_shards: usize) -> (String, String) { + let first = "wk0".to_string(); + let s0 = crate::shard::dispatch::key_to_shard(first.as_bytes(), num_shards); + for i in 1..4096 { + let cand = format!("wk{i}"); + if crate::shard::dispatch::key_to_shard(cand.as_bytes(), num_shards) != s0 { + return (first, cand); + } + } + panic!("no cross-shard key pair found in 4096 candidates at {num_shards} shards"); + } + + #[test] + fn no_watched_keys_is_keyless() { + assert_eq!( + analyze_watch_locality(&watched(&[]), 4), + TxnLocality::Keyless + ); + } + + #[test] + fn keys_on_one_shard_classify_to_that_shard() { + let k = "solo"; + let expect = crate::shard::dispatch::key_to_shard(k.as_bytes(), 4); + assert_eq!( + analyze_watch_locality(&watched(&[k]), 4), + TxnLocality::SingleShard(expect) + ); + } + + #[test] + fn keys_on_different_shards_are_cross_shard() { + let (a, b) = cross_shard_pair(4); + assert_eq!( + analyze_watch_locality(&watched(&[&a, &b]), 4), + TxnLocality::CrossShard + ); + } + + /// A single shard is degenerate: every key lands on shard 0, so a watch set + /// can never be cross-shard there. Worth pinning — the CROSSSLOT refusal + /// must not fire at `--shards 1`, the default for most deployments. + #[test] + fn one_shard_never_classifies_cross_shard() { + assert_eq!( + analyze_watch_locality(&watched(&["a", "b", "c"]), 1), + TxnLocality::SingleShard(0) + ); + } + + // --- the merge lattice: all nine combinations --- + + #[test] + fn keyless_is_the_identity() { + assert_eq!( + merge_locality(TxnLocality::Keyless, TxnLocality::Keyless), + TxnLocality::Keyless + ); + assert_eq!( + merge_locality(TxnLocality::Keyless, TxnLocality::SingleShard(2)), + TxnLocality::SingleShard(2) + ); + assert_eq!( + merge_locality(TxnLocality::SingleShard(2), TxnLocality::Keyless), + TxnLocality::SingleShard(2) + ); + } + + #[test] + fn cross_shard_absorbs_everything() { + for other in [ + TxnLocality::Keyless, + TxnLocality::SingleShard(0), + TxnLocality::CrossShard, + ] { + assert_eq!( + merge_locality(TxnLocality::CrossShard, other), + TxnLocality::CrossShard + ); + assert_eq!( + merge_locality(other, TxnLocality::CrossShard), + TxnLocality::CrossShard + ); + } + } + + #[test] + fn agreeing_shards_stay_single() { + assert_eq!( + merge_locality(TxnLocality::SingleShard(3), TxnLocality::SingleShard(3)), + TxnLocality::SingleShard(3) + ); + } + + /// The case the whole merge exists for: the body commits on one shard and + /// the watch set lives on another, so the CAS check would read the wrong + /// slice and fabricate a conflict the client can never clear. Refusing is + /// the contract; silently aborting forever is a livelock. + #[test] + fn disagreeing_shards_become_cross_shard() { + assert_eq!( + merge_locality(TxnLocality::SingleShard(1), TxnLocality::SingleShard(2)), + TxnLocality::CrossShard + ); + } +} diff --git a/src/server/conn/watch.rs b/src/server/conn/watch.rs new file mode 100644 index 000000000..edb3fe656 --- /dev/null +++ b/src/server/conn/watch.rs @@ -0,0 +1,81 @@ +//! `WATCH` / `UNWATCH` — the optimistic-locking half of Redis transactions. +//! +//! Lives in its own module rather than inside each handler because this task +//! (`watch-cas-transactions`) exists precisely because the three dispatch paths +//! drifted: `handler_single` re-checked the recorded versions at `EXEC` while +//! `handler_monoio` and `handler_sharded` parsed `WATCH`, answered `+OK`, and +//! then never consulted the watch set again. Two byte-identical copies of the +//! command arm would have re-created the same failure mode one edit later, so +//! both production handlers call this one function. +//! +//! `handler_single` keeps its own inline arm: it holds the database lock +//! directly and has no shard mesh to hop, so it shares no code with this path. + +use bytes::Bytes; + +use crate::protocol::Frame; +use crate::server::conn::core::{ConnectionContext, ConnectionState}; +use crate::server::conn::shared::WatchToken; + +/// Handle `WATCH` / `UNWATCH`, returning `true` when the command was consumed. +/// +/// MUST be called BEFORE the `MULTI` queueing step, so `WATCH` inside a +/// transaction is refused rather than queued as an ordinary command. +pub(crate) async fn try_handle_watch_unwatch( + cmd: &[u8], + args: &[Frame], + conn: &mut ConnectionState, + ctx: &ConnectionContext, + responses: &mut Vec, +) -> bool { + if cmd.eq_ignore_ascii_case(b"WATCH") { + if args.is_empty() { + responses.push(Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'watch' command", + ))); + } else if conn.in_multi { + responses.push(Frame::Error(Bytes::from_static( + b"ERR WATCH inside MULTI is not allowed", + ))); + } else { + let keys: Vec = args + .iter() + .filter_map(|f| match f { + Frame::BulkString(b) => Some(b.clone()), + Frame::SimpleString(b) => Some(b.clone()), + _ => None, + }) + .collect(); + // Snapshot each key's version WHERE IT LIVES. Reading the local + // slice for a remote key would read a different database entirely. + let versions = crate::shard::coordinator::snapshot_versions( + &keys, + ctx.shard_id, + ctx.num_shards, + conn.selected_db, + &ctx.dispatch_tx, + &ctx.spsc_notifiers, + ) + .await; + for (k, v) in keys.into_iter().zip(versions) { + conn.watched_keys.insert(k, WatchToken { version: v }); + } + responses.push(Frame::SimpleString(Bytes::from_static(b"OK"))); + } + return true; + } + + if cmd.eq_ignore_ascii_case(b"UNWATCH") { + if !args.is_empty() { + responses.push(Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'unwatch' command", + ))); + } else { + conn.watched_keys.clear(); + responses.push(Frame::SimpleString(Bytes::from_static(b"OK"))); + } + return true; + } + + false +} diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index 91b2b11b2..932f93d1e 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -254,6 +254,8 @@ pub(crate) async fn execute_txn_on_owner( db_index: usize, commands: Vec, proto: u8, + // WATCH tokens travel with the body: the CAS check runs on the owner. + watched: std::collections::HashMap, dispatch_tx: &Rc>>>, spsc_notifiers: &[Arc], ) -> Option { @@ -264,12 +266,72 @@ pub(crate) async fn execute_txn_on_owner( commands, reply_tx, proto, + watched, }; let msg = ShardMessage::TxnExecute(Box::new(payload)); let _ = spsc_send(dispatch_tx, my_shard, owner, msg, spsc_notifiers).await; recv_reply_bounded(reply_rx).await.ok() } +/// Snapshot the versions of `keys` from whichever shards own them (WATCH). +/// +/// Returns versions positionally aligned with `keys`; `0` means absent, which +/// is a real token (watching a key that does not exist and seeing it created +/// IS a conflict). Local keys are read inline; remote keys are grouped per +/// owner so a WATCH of N keys costs at most one hop per shard, not per key. +/// +/// A dead owner yields `0` for its keys, which is fail-SAFE in the only +/// direction that matters: `0` almost never matches a live key's version, so +/// the transaction aborts rather than committing on an unverified dependency. +pub(crate) async fn snapshot_versions( + keys: &[Bytes], + my_shard: usize, + num_shards: usize, + db_index: usize, + dispatch_tx: &Rc>>>, + spsc_notifiers: &[Arc], +) -> Vec { + let mut out = vec![0u32; keys.len()]; + // owner -> (original indices, keys) + let mut groups: std::collections::HashMap, Vec)> = + std::collections::HashMap::new(); + for (i, k) in keys.iter().enumerate() { + let owner = key_to_shard(k, num_shards); + let e = groups.entry(owner).or_default(); + e.0.push(i); + e.1.push(k.clone()); + } + + for (owner, (idxs, group_keys)) in groups { + if owner == my_shard { + let versions = crate::shard::slice::with_shard_db(db_index, |db| { + group_keys + .iter() + .map(|k| db.get_version(k)) + .collect::>() + }); + for (slot, v) in idxs.iter().zip(versions) { + out[*slot] = v; + } + continue; + } + let (reply_tx, reply_rx) = channel::oneshot(); + let payload = crate::shard::dispatch::ReadVersionsPayload { + db_index, + keys: group_keys, + reply_tx, + }; + let msg = ShardMessage::ReadVersions(Box::new(payload)); + let _ = spsc_send(dispatch_tx, my_shard, owner, msg, spsc_notifiers).await; + if let Ok(versions) = recv_reply_bounded(reply_rx).await { + for (slot, v) in idxs.iter().zip(versions) { + out[*slot] = v; + } + } + } + out +} + /// Run one full command on whichever shard owns `routing_key`. #[allow(clippy::too_many_arguments)] async fn run_on_owner( diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index 8e3d4569a..faf04163f 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -214,6 +214,30 @@ pub struct DocFreqPayload { pub db_index: u8, } +/// Payload for [`ShardMessage::ReadVersions`] (WATCH snapshot, task +/// `watch-cas-transactions`). +/// +/// `WATCH k` must record `k`'s version AT WATCH TIME — that snapshot is the +/// whole CAS token. At `--shards > 1` the watched key may be owned by a shard +/// the connection is not on, and `with_shard_db` reads the LOCAL slice, which +/// for a remote key is simply the wrong database. Reading it where it lives is +/// the only honest option: refusing non-local watches would make WATCH +/// near-unusable above one shard, since which shard a connection lands on is +/// unrelated to which shard its keys hash to. +/// +/// Cheap by construction: WATCH is rare, and this hop happens once per WATCH, +/// never on the command hot path. +pub struct ReadVersionsPayload { + /// Target database index (the connection's `selected_db`). + pub db_index: usize, + /// The keys to snapshot — only those this shard owns; the caller groups. + pub keys: Vec, + /// Versions in the SAME order as `keys`. `0` means absent, which is a real + /// token: watching a key that does not exist and seeing it created is a + /// conflict. + pub reply_tx: channel::OneshotSender>, +} + /// Boxed payload for `ShardMessage::TextAggregate` (Phase 152 D-05/D-07). /// /// Kept as a separate struct so the enum variant stays small (single @@ -719,6 +743,11 @@ pub enum ShardMessage { /// variant pushed `ShardMessage` past the 64-byte cap asserted at module /// bottom — boxing collapses it back to a single pointer. DocFreq(Box), + /// Snapshot key versions on the shard that owns them (WATCH). + /// + /// Boxed to keep `ShardMessage` within the 64-byte cap, like its siblings. + ReadVersions(Box), + /// DFS Phase 2: execute BM25 text search with injected global IDF. /// /// Returns `Frame::Array` in the same format as `ft_text_search` response: @@ -955,6 +984,13 @@ pub struct TxnExecutePayload { /// EXEC would answer RESP2 shapes to a RESP3 client — the same /// context-dependent shape defect, reached over a shard hop. pub proto: u8, + /// The ORIGINATING connection's WATCH tokens. The CAS check runs where the + /// body runs — on the owner shard — so the tokens must make the hop with + /// it. Leaving them behind would mean WATCH silently stops guarding as soon + /// as the transaction is owner-routed, i.e. at every `--shards > 1` + /// deployment: the exact silent-guarantee failure this task exists to + /// remove. Empty for non-WATCH transactions, which is nearly all of them. + pub watched: std::collections::HashMap, } /// Reply for [`ShardMessage::TxnExecute`]. diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 09ee41596..13cfd0ee3 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2124,6 +2124,20 @@ pub(crate) fn handle_shard_message_shared( }); let _ = reply_tx.send(response); } + ShardMessage::ReadVersions(payload) => { + // WATCH snapshot on the owning shard (task `watch-cas-transactions`). + // Read-only and synchronous: no await between the reads, so the + // snapshot is coherent for this shard's slice. + let crate::shard::dispatch::ReadVersionsPayload { + db_index, + keys, + reply_tx, + } = *payload; + let versions = crate::shard::slice::with_shard_db(db_index, |db| { + keys.iter().map(|k| db.get_version(k)).collect::>() + }); + let _ = reply_tx.send(versions); + } ShardMessage::DocFreq(payload) => { let crate::shard::dispatch::DocFreqPayload { index_name, @@ -2777,6 +2791,7 @@ pub(crate) fn handle_shard_message_shared( commands, reply_tx, proto, + watched, } = *payload; let mut exec_publishes: Vec<(usize, bytes::Bytes, bytes::Bytes)> = Vec::new(); // c10k E2: a queued FLUSHDB/FLUSHALL clears only THIS shard's @@ -2794,6 +2809,7 @@ pub(crate) fn handle_shard_message_shared( cached_clock, &mut exec_publishes, &mut exec_flushes, + &watched, ); // task #52: this arm is the CROSS-SHARD EXEC hop (the accepting // connection's shard differs from the owner shard, which by diff --git a/src/storage/db/accessors.rs b/src/storage/db/accessors.rs index 0d99d129c..ac0d27cb6 100644 --- a/src/storage/db/accessors.rs +++ b/src/storage/db/accessors.rs @@ -80,7 +80,11 @@ impl Database { if !self.data.contains_key(key) { self.promote_cold_if_present(key, now_ms); if !self.data.contains_key(key) { - let entry = K::new_entry(); + let mut entry = K::new_entry(); + // Fresh incarnation: stamp the per-db creation ticket so a + // WATCHing client can tell this container from the one that + // occupied the key before (see `Database::birth_counter`). + entry.set_version(self.next_birth_version()); let k = CompactKey::from(key); self.used_memory += entry_overhead(key, &entry); self.data.insert(k, entry); @@ -204,7 +208,11 @@ impl Database { // callers (e.g. SADD) to `get_or_create_set` — no fabrication. self.promote_cold_if_present(key, now_ms); if !self.data.contains_key(key) { - let entry = Entry::new_set_intset(); + let mut entry = Entry::new_set_intset(); + // Fresh incarnation: stamp the per-db creation ticket so a + // WATCHing client can tell this container from the one that + // occupied the key before (see `Database::birth_counter`). + entry.set_version(self.next_birth_version()); let k = CompactKey::from(key); self.used_memory += entry_overhead(key, &entry); self.data.insert(k, entry); @@ -261,7 +269,11 @@ impl Database { // callers (e.g. HSET) to `get_or_create_hash` — no fabrication. self.promote_cold_if_present(key, now_ms); if !self.data.contains_key(key) { - let entry = Entry::new_hash_listpack(); + let mut entry = Entry::new_hash_listpack(); + // Fresh incarnation: stamp the per-db creation ticket so a + // WATCHing client can tell this container from the one that + // occupied the key before (see `Database::birth_counter`). + entry.set_version(self.next_birth_version()); let k = CompactKey::from(key); self.used_memory += entry_overhead(key, &entry); self.data.insert(k, entry); @@ -320,7 +332,11 @@ impl Database { // callers (e.g. LPUSH) to `get_or_create_list` — no fabrication. self.promote_cold_if_present(key, now_ms); if !self.data.contains_key(key) { - let entry = Entry::new_list_listpack(); + let mut entry = Entry::new_list_listpack(); + // Fresh incarnation: stamp the per-db creation ticket so a + // WATCHing client can tell this container from the one that + // occupied the key before (see `Database::birth_counter`). + entry.set_version(self.next_birth_version()); let k = CompactKey::from(key); self.used_memory += entry_overhead(key, &entry); self.data.insert(k, entry); diff --git a/src/storage/db/kv_ops.rs b/src/storage/db/kv_ops.rs index 2e7415331..792bc6ee1 100644 --- a/src/storage/db/kv_ops.rs +++ b/src/storage/db/kv_ops.rs @@ -333,6 +333,11 @@ impl Database { // Exactly one closure runs (FnOnce), so the take() is safe. let entry_cell = std::cell::Cell::new(Some(entry)); + // Drawn before the closures because both borrow `self.data`. Wasted on + // the hit path, which is deliberate: the counter is a ticket dispenser, + // not a count, and a gap costs nothing. + let birth = self.next_birth_version(); + // `insert_or_update` invariant: exactly one of the two closures fires // exactly once per call, so the `Cell::take()` below cannot observe // a None value. Annotated for the hot-path unwrap ratchet. @@ -348,9 +353,12 @@ impl Database { existing.set_version(new_version); }, || { - // Miss path: insert entry as-is (constructors start versions - // at INITIAL_VERSION=1 so WATCH can detect creation). - entry_cell.take().expect("make closure called once on miss") + // Miss path: stamp the creation ticket. Constructors all start + // at INITIAL_VERSION, which made every incarnation of a key + // indistinguishable from its first — the WATCH ABA hole. + let mut new_entry = entry_cell.take().expect("make closure called once on miss"); + new_entry.set_version(birth); + new_entry }, ); @@ -425,15 +433,23 @@ impl Database { &self.hot_keys } - /// Bulk-load insert: skip duplicate check, version tracking, and per-key memory accounting. + /// Bulk-load insert: skip duplicate check and per-key memory accounting. /// /// Used exclusively during RDB/AOF restore where keys are guaranteed unique and /// we recalculate `used_memory` once after the entire load completes. + /// + /// Still draws a creation ticket, unlike the rest of the fast path it skips. + /// Versions are not persisted, so every restored entry would otherwise carry + /// `INITIAL_VERSION`, and the FIRST key created after a restore would draw + /// ticket 1 and collide with them — reopening the ABA hole for exactly the + /// window right after a restart. One u32 increment per restored key against + /// a decode is not measurable. #[inline] - pub fn insert_for_load(&mut self, key: Bytes, entry: Entry) { + pub fn insert_for_load(&mut self, key: Bytes, mut entry: Entry) { if entry.has_expiry() { self.maybe_has_expiring_keys = true; } + entry.set_version(self.next_birth_version()); self.data.insert(CompactKey::from(key), entry); } @@ -922,6 +938,20 @@ impl Database { )) } + /// Take the next creation ticket from the per-db birth counter. + /// + /// Stamped on every entry this database fabricates so a delete+recreate is + /// observably a different incarnation rather than a fresh `INITIAL_VERSION` + /// that a WATCHing client mistakes for its own recorded token. See + /// `Database::birth_counter` for the wrap analysis. + /// + /// Consuming a ticket without using it is fine — gaps carry no meaning. + #[inline] + pub(crate) fn next_birth_version(&mut self) -> u32 { + self.birth_counter = Entry::bump_version(self.birth_counter); + self.birth_counter + } + /// Get the version of a key. Returns 0 if not found. No expiry check (WATCH needs raw version). pub fn get_version(&self, key: &[u8]) -> u32 { self.data.get(key).map(|e| e.version()).unwrap_or(0) diff --git a/src/storage/db/mod.rs b/src/storage/db/mod.rs index 8a3b4683e..6aaeed0b7 100644 --- a/src/storage/db/mod.rs +++ b/src/storage/db/mod.rs @@ -243,6 +243,23 @@ pub struct Database { /// large, or it would evict in a runaway loop chasing memory that cannot /// drop yet — a design change, deliberately out of scope for #459. spill_inflight: std::collections::HashMap, + /// Ticket dispenser for the version stamped on a NEWLY CREATED entry. + /// + /// Versions are per-entry and die with the entry. Before this counter every + /// creation started at `INITIAL_VERSION`, so `DEL k` + `SET k` handed a + /// WATCHing client back the exact token it had recorded and EXEC committed + /// on a key that had been destroyed and rebuilt underneath it — the ABA + /// hole (Redis aborts there). Stamping each creation with the next value of + /// a per-db counter makes recreation observably distinct instead. + /// + /// Monotonic per database, bumped on every `set` and every `get_or_create` + /// fabrication; gaps are harmless (it is a ticket, not a count). Shares the + /// entry's 24-bit version field, so it wraps at 16,777,216 — a miss now + /// needs that wrap to land inside one client's open WATCH..EXEC window AND + /// hit the one watched key, ~1 in 16.7M versus the pre-fix certainty. Only + /// a full incarnation field (a wider `Entry`) removes the residue entirely; + /// see the task's §7. + birth_counter: u32, } /// A spill that has left hot RAM but has not yet landed in `cold_index`. @@ -275,6 +292,7 @@ impl Database { cold_shard_dir: None, hot_keys: crate::storage::hotkey::HotKeySketch::new(), spill_inflight: std::collections::HashMap::new(), + birth_counter: 0, } } @@ -300,6 +318,7 @@ impl Database { cold_shard_dir: None, hot_keys: crate::storage::hotkey::HotKeySketch::new(), spill_inflight: std::collections::HashMap::new(), + birth_counter: 0, } } @@ -925,13 +944,98 @@ mod tests { // 0 is reserved for "key absent" so WATCH detects creation. assert_eq!(db.get_version(b"key"), 0); db.set_string(Bytes::from_static(b"key"), Bytes::from_static(b"v1")); - assert_eq!(db.get_version(b"key"), 1); // first set: INITIAL_VERSION + assert_eq!(db.get_version(b"key"), 1); // first creation draws ticket 1 db.set_string(Bytes::from_static(b"key"), Bytes::from_static(b"v2")); assert_eq!(db.get_version(b"key"), 2); // overwrite bumps db.set_string(Bytes::from_static(b"key"), Bytes::from_static(b"v3")); assert_eq!(db.get_version(b"key"), 3); } + /// The ABA hole: before the birth counter every creation started at + /// `INITIAL_VERSION`, so DEL + re-SET handed a WATCHing client back the + /// exact token it had recorded and EXEC committed on a key that had been + /// destroyed and rebuilt underneath it. + #[test] + fn test_recreated_key_never_reuses_its_old_version() { + let mut db = Database::new(); + db.set_string(Bytes::from_static(b"k"), Bytes::from_static(b"v0")); + let watched = db.get_version(b"k"); + + db.remove(b"k"); + assert_eq!(db.get_version(b"k"), 0, "removed key must read as absent"); + db.set_string(Bytes::from_static(b"k"), Bytes::from_static(b"v1")); + + assert_ne!( + db.get_version(b"k"), + watched, + "recreated key presented the version WATCH recorded" + ); + } + + /// The ticket is per-database, not per-key: two keys created in sequence + /// must not both read as version 1, or the counter has degenerated back + /// into a per-entry constant. + #[test] + fn test_birth_tickets_are_distinct_across_keys() { + let mut db = Database::new(); + db.set_string(Bytes::from_static(b"a"), Bytes::from_static(b"v")); + db.set_string(Bytes::from_static(b"b"), Bytes::from_static(b"v")); + assert_ne!(db.get_version(b"a"), db.get_version(b"b")); + } + + /// Containers fabricated by `get_or_create` take tickets too — HSET on a + /// deleted-and-rebuilt hash is the same ABA hole as SET on a string. + #[test] + fn test_fabricated_containers_take_birth_tickets() { + let mut db = Database::new(); + db.get_or_create_hash(b"h") + .unwrap() + .insert(Bytes::from_static(b"f"), Bytes::from_static(b"v")); + let watched = db.get_version(b"h"); + + db.remove(b"h"); + db.get_or_create_hash(b"h") + .unwrap() + .insert(Bytes::from_static(b"f"), Bytes::from_static(b"v")); + + assert_ne!( + db.get_version(b"h"), + watched, + "recreated hash presented the version WATCH recorded" + ); + } + + /// Restored entries carry tickets as well. Versions are not persisted, so + /// without this every restored key would read `INITIAL_VERSION` and the + /// first key created after the restore would collide with all of them. + #[test] + fn test_restored_keys_do_not_collide_with_the_first_new_key() { + let mut db = Database::new(); + db.insert_for_load( + Bytes::from_static(b"restored"), + Entry::new_string(Bytes::from_static(b"v")), + ); + let restored = db.get_version(b"restored"); + assert_ne!(restored, 0, "a loaded key must not read as absent"); + + db.set_string(Bytes::from_static(b"fresh"), Bytes::from_static(b"v")); + assert_ne!(db.get_version(b"fresh"), restored); + } + + /// The counter shares the entry's 24-bit version field, so it wraps — + /// and must skip 0, which is the "key absent" sentinel. A wrapped ticket + /// reading as 0 would make a live key invisible to WATCH. + #[test] + fn test_birth_ticket_wraps_without_ever_yielding_zero() { + let mut db = Database::new(); + db.birth_counter = 0xFF_FFFE; + assert_eq!(db.next_birth_version(), 0xFF_FFFF); + assert_eq!( + db.next_birth_version(), + crate::storage::entry::INITIAL_VERSION + ); + } + #[test] fn test_increment_version() { let mut db = Database::new(); diff --git a/tests/watch_cas_transactions.rs b/tests/watch_cas_transactions.rs new file mode 100644 index 000000000..07e2f8cb7 --- /dev/null +++ b/tests/watch_cas_transactions.rs @@ -0,0 +1,489 @@ +//! ADD task `watch-cas-transactions` — failing-first suite. +//! +//! WATCH is optimistic locking: after `WATCH k`, if anyone writes `k` before +//! `EXEC`, the transaction must abort. Every client library's CAS loop — +//! redis-py `pipeline().watch()`, go-redis `TxPipelined`, Lettuce — is built on +//! exactly that, and nothing else in Moon's test tree asserts it. +//! +//! Measured on `main` @8b1153b4, default (monoio) build, both shards=1 and +//! shards=4, RESP and inline: `WATCH` and `UNWATCH` reply +//! `-ERR unknown command`. WATCH exists only in `handler_single.rs` (the +//! EMBEDDED path) and the CAS check only in `execute_transaction`, which is +//! `#[cfg(feature = "runtime-tokio")]`. The two production paths route through +//! `execute_transaction_sharded`, whose signature has no `watched_keys` +//! parameter at all — the check is not skipped, it is structurally absent. So a +//! transaction that declared a dependency on `k` commits over a conflicting +//! write and silently clobbers it. +//! +//! Expected RED on main: +//! wc1 conflicting write does not abort EXEC (the headline) +//! wc3 watch on an absent key does not abort +//! wc4 UNWATCH errors +//! wc5 watches are never set, so nothing to clear +//! wc6 delete+recreate does not abort (the ABA hole) +//! wc8 WATCH arity is not enforced (unknown command instead) +//! wc9 WATCH inside MULTI is not refused +//! wc10 cross-shard watch is not classified +//! +//! Two are GREEN on main, both deliberately: +//! wc2 a clean EXEC still commits — the behavior the build must not break. +//! wc7 the paths agree. It passes today because shards=1 and shards=4 are +//! equally broken, which is a real (if bleak) agreement. Its job is to +//! stop a fix that lands on ONE production path — the failure mode that +//! made the #457 inline-GET ACL bypass invisible — so it is worth +//! keeping even though it cannot fail for the headline reason. +//! +//! Run alone with: cargo test --test watch_cas_transactions + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +// --------------------------------------------------------------------------- +// Harness +// --------------------------------------------------------------------------- + +fn spawn_moon(dir: &std::path::Path, shards: u32) -> (Child, u16) { + common::spawn_listening(|port| { + Command::new(common::find_moon_binary()) + .args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + &shards.to_string(), + "--appendonly", + "no", + // The shared /Volumes checkout hovers near the 5% diskfull + // guard; a tripped guard turns every write into MOONERR and + // would fail this suite for an unrelated reason. + "--disk-free-min-pct", + "0", + ]) + .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") + }) +} + +struct ServerGuard(Child); +impl Drop for ServerGuard { + fn drop(&mut self) { + common::sigkill(&mut self.0); + } +} + +/// Connect and return only once the server answered a PING on THIS socket. +/// The listener can accept before the shard behind it serves; under a fully +/// parallel run that first connection comes back RST. Setup-only on purpose — +/// a reset inside a test body still panics, because there it is a finding. +fn connect_ready(port: u16) -> TcpStream { + let deadline = Instant::now() + Duration::from_secs(30); + loop { + if let Ok(mut s) = TcpStream::connect(format!("127.0.0.1:{port}")) { + s.set_read_timeout(Some(Duration::from_secs(10))).ok(); + s.set_write_timeout(Some(Duration::from_secs(10))).ok(); + if s.write_all(b"PING\r\n").is_ok() { + 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!( + Instant::now() < deadline, + "server on {port} never answered PING" + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +/// One command, one raw reply. Raw bytes on purpose: the abort signal IS the +/// type byte (`*-1` / `_`), so a reader that renders replies to text would hide +/// the very thing under test. +fn cmd(s: &mut TcpStream, args: &[&str]) -> Vec { + let mut out = format!("*{}\r\n", args.len()).into_bytes(); + for a in args { + out.extend_from_slice(format!("${}\r\n{a}\r\n", a.len()).as_bytes()); + } + s.write_all(&out).expect("write command"); + read_reply(s) +} + +fn read_reply(s: &mut TcpStream) -> Vec { + // One reply per command here, and every reply this suite sees is small, so + // a single bounded read with a short settle is enough and keeps the + // assertions on raw bytes. + std::thread::sleep(Duration::from_millis(60)); + let mut buf = vec![0u8; 65536]; + match s.read(&mut buf) { + Ok(n) => buf[..n].to_vec(), + Err(e) => panic!("read reply: {e}"), + } +} + +fn is_null(reply: &[u8]) -> bool { + reply.starts_with(b"*-1\r\n") || reply.starts_with(b"$-1\r\n") || reply.starts_with(b"_\r\n") +} + +fn text(reply: &[u8]) -> String { + String::from_utf8_lossy(reply).into_owned() +} + +/// Two keys that provably land on different shards, discovered by ASKING the +/// server rather than reimplementing `key_to_shard` in the test: a MULTI body +/// spanning shards already answers CROSSSLOT today, so that reply is the oracle. +fn find_cross_shard_pair(port: u16) -> Option<(String, String)> { + let mut s = connect_ready(port); + for i in 0..64 { + let (a, b) = (format!("wc:a{i}"), format!("wc:b{i}")); + cmd(&mut s, &["MULTI"]); + cmd(&mut s, &["GET", &a]); + cmd(&mut s, &["GET", &b]); + let r = cmd(&mut s, &["EXEC"]); + if text(&r).contains("CROSSSLOT") { + return Some((a, b)); + } + cmd(&mut s, &["DISCARD"]); + } + None +} + +/// Fresh server + a temp dir that dies with the test. Every test gets its own +/// `--dir`: an empty/reused dir silently reloads another test's state. +fn server(shards: u32) -> (ServerGuard, u16, tempfile::TempDir) { + let dir = tempfile::tempdir().expect("tempdir"); + let (child, port) = spawn_moon(dir.path(), shards); + (ServerGuard(child), port, dir) +} + +// --------------------------------------------------------------------------- +// wc1 — THE headline: a conflicting write must abort the transaction +// --------------------------------------------------------------------------- + +#[test] +fn wc1_a_conflicting_write_aborts_the_transaction() { + for shards in [1u32, 4] { + let (_g, port, _d) = server(shards); + let (mut a, mut b) = (connect_ready(port), connect_ready(port)); + + cmd(&mut a, &["SET", "k", "v0"]); + let watch = cmd(&mut a, &["WATCH", "k"]); + assert!( + watch.starts_with(b"+OK"), + "shards={shards}: WATCH must be accepted, got {:?}", + text(&watch) + ); + + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "k", "from-A"]); + // The conflict, from a genuinely separate connection. + cmd(&mut b, &["SET", "k", "from-B"]); + + let exec = cmd(&mut a, &["EXEC"]); + assert!( + is_null(&exec), + "shards={shards}: EXEC must abort after a conflicting write to a watched key, \ + got {:?}", + text(&exec) + ); + + // The abort is only real if the queued write did not land. + let got = cmd(&mut b, &["GET", "k"]); + assert!( + text(&got).contains("from-B"), + "shards={shards}: aborted transaction still wrote — k should hold the conflicting \ + value from-B, got {:?}", + text(&got) + ); + } +} + +// --------------------------------------------------------------------------- +// wc2 — the pin: no conflict must still commit (expected green today) +// --------------------------------------------------------------------------- + +#[test] +fn wc2_a_clean_transaction_still_commits() { + for shards in [1u32, 4] { + let (_g, port, _d) = server(shards); + let mut a = connect_ready(port); + + cmd(&mut a, &["SET", "k", "v0"]); + cmd(&mut a, &["WATCH", "k"]); + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "k", "from-A"]); + let exec = cmd(&mut a, &["EXEC"]); + + assert!( + !is_null(&exec), + "shards={shards}: EXEC must COMMIT when nothing wrote the watched key, got a \ + null abort. An over-eager CAS check is as broken as a missing one." + ); + let got = cmd(&mut a, &["GET", "k"]); + assert!( + text(&got).contains("from-A"), + "shards={shards}: committed transaction did not apply, got {:?}", + text(&got) + ); + } +} + +// --------------------------------------------------------------------------- +// wc3 — a watch on an absent key is still a dependency +// --------------------------------------------------------------------------- + +#[test] +fn wc3_watching_an_absent_key_aborts_when_it_appears() { + let (_g, port, _d) = server(1); + let (mut a, mut b) = (connect_ready(port), connect_ready(port)); + + cmd(&mut a, &["DEL", "absent"]); + cmd(&mut a, &["WATCH", "absent"]); + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "sentinel", "written"]); + cmd(&mut b, &["SET", "absent", "now-exists"]); + + let exec = cmd(&mut a, &["EXEC"]); + assert!( + is_null(&exec), + "EXEC must abort when a watched key that did not exist was created, got {:?}", + text(&exec) + ); + let sentinel = cmd(&mut b, &["EXISTS", "sentinel"]); + assert!( + text(&sentinel).contains(":0"), + "aborted transaction still ran its body — sentinel should not exist, got {:?}", + text(&sentinel) + ); +} + +// --------------------------------------------------------------------------- +// wc4 — UNWATCH releases the dependency +// --------------------------------------------------------------------------- + +#[test] +fn wc4_unwatch_releases_the_dependency() { + let (_g, port, _d) = server(1); + let (mut a, mut b) = (connect_ready(port), connect_ready(port)); + + cmd(&mut a, &["SET", "k", "v0"]); + cmd(&mut a, &["WATCH", "k"]); + let un = cmd(&mut a, &["UNWATCH"]); + assert!( + un.starts_with(b"+OK"), + "UNWATCH must reply +OK, got {:?}", + text(&un) + ); + + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "k", "from-A"]); + cmd(&mut b, &["SET", "k", "from-B"]); + + let exec = cmd(&mut a, &["EXEC"]); + assert!( + !is_null(&exec), + "after UNWATCH the conflicting write must NOT abort the transaction, got {:?}", + text(&exec) + ); + let got = cmd(&mut a, &["GET", "k"]); + assert!( + text(&got).contains("from-A"), + "unwatched transaction should have committed, got {:?}", + text(&got) + ); +} + +// --------------------------------------------------------------------------- +// wc5 — EXEC clears watches on BOTH outcomes +// --------------------------------------------------------------------------- + +#[test] +fn wc5_exec_clears_watches_on_both_outcomes() { + let (_g, port, _d) = server(1); + let (mut a, mut b) = (connect_ready(port), connect_ready(port)); + + // Cycle 1: force an abort. + cmd(&mut a, &["SET", "k", "v0"]); + cmd(&mut a, &["WATCH", "k"]); + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "k", "v1"]); + cmd(&mut b, &["SET", "k", "conflict"]); + let first = cmd(&mut a, &["EXEC"]); + assert!(is_null(&first), "setup: first EXEC should abort"); + + // Cycle 2: no new WATCH. A watch surviving the abort would wrongly abort + // this one too — a stale dependency is how a CAS loop livelocks. + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "k", "v2"]); + let second = cmd(&mut a, &["EXEC"]); + assert!( + !is_null(&second), + "the aborted EXEC must have cleared its watches; the next transaction \ + aborted with no WATCH of its own, got {:?}", + text(&second) + ); + let got = cmd(&mut a, &["GET", "k"]); + assert!( + text(&got).contains("v2"), + "second transaction should have committed, got {:?}", + text(&got) + ); +} + +// --------------------------------------------------------------------------- +// wc6 — the ABA hole: delete + recreate is a conflict +// --------------------------------------------------------------------------- + +#[test] +fn wc6_delete_and_recreate_is_a_conflict() { + let (_g, port, _d) = server(1); + let (mut a, mut b) = (connect_ready(port), connect_ready(port)); + + // Versions are per-entry, start at INITIAL_VERSION = 1, and die with the + // entry (src/storage/entry.rs:332). So DEL + re-SET returns k to version 1 + // — the same token WATCH recorded — and a version-only check commits where + // Redis aborts. The key was destroyed and recreated; that IS a conflict. + cmd(&mut a, &["SET", "k", "v0"]); + cmd(&mut a, &["WATCH", "k"]); + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "sentinel", "written"]); + + cmd(&mut b, &["DEL", "k"]); + cmd(&mut b, &["SET", "k", "v0"]); + + let exec = cmd(&mut a, &["EXEC"]); + assert!( + is_null(&exec), + "EXEC must abort after a watched key was deleted and recreated (ABA): the entry \ + version resets to INITIAL_VERSION, so a version-only check cannot see it. Got {:?}", + text(&exec) + ); + let sentinel = cmd(&mut b, &["EXISTS", "sentinel"]); + assert!( + text(&sentinel).contains(":0"), + "aborted transaction still ran its body, got {:?}", + text(&sentinel) + ); +} + +// --------------------------------------------------------------------------- +// wc7 — every dispatch path agrees +// --------------------------------------------------------------------------- + +#[test] +fn wc7_all_dispatch_paths_agree() { + // Same sequence, both shard counts. The reply must be byte-identical: + // WATCH lives only in the embedded path today, so this is where a + // production-path-only gap shows up as a difference rather than a guess. + let mut seen: Vec<(u32, String, String)> = Vec::new(); + for shards in [1u32, 4] { + let (_g, port, _d) = server(shards); + let (mut a, mut b) = (connect_ready(port), connect_ready(port)); + + cmd(&mut a, &["SET", "k", "v0"]); + let watch = text(&cmd(&mut a, &["WATCH", "k"])); + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "k", "from-A"]); + cmd(&mut b, &["SET", "k", "from-B"]); + let exec = text(&cmd(&mut a, &["EXEC"])); + seen.push((shards, watch, exec)); + } + let (s0, w0, e0) = &seen[0]; + for (s, w, e) in &seen[1..] { + assert_eq!( + w, w0, + "WATCH reply differs between shards={s0} and shards={s}: {w0:?} vs {w:?}" + ); + assert_eq!( + e, e0, + "EXEC reply differs between shards={s0} and shards={s}: {e0:?} vs {e:?}" + ); + } +} + +// --------------------------------------------------------------------------- +// wc8 / wc9 / wc10 — the rejections +// --------------------------------------------------------------------------- + +#[test] +fn wc8_watch_without_keys_is_an_arity_error() { + let (_g, port, _d) = server(1); + let mut a = connect_ready(port); + + let r = text(&cmd(&mut a, &["WATCH"])); + assert!( + r.contains("wrong number of arguments"), + "bare WATCH must be an arity error, got {r:?}" + ); + + // And it must not have half-registered anything: a later transaction with + // no watch of its own still commits. + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", "k", "v1"]); + let exec = cmd(&mut a, &["EXEC"]); + assert!( + !is_null(&exec), + "a rejected WATCH must leave no dependency behind, got {:?}", + text(&exec) + ); +} + +#[test] +fn wc9_watch_inside_multi_is_refused() { + let (_g, port, _d) = server(1); + let mut a = connect_ready(port); + + cmd(&mut a, &["MULTI"]); + let r = text(&cmd(&mut a, &["WATCH", "k"])); + assert!( + r.contains("WATCH inside MULTI"), + "WATCH inside MULTI must be refused, got {r:?}" + ); + + // It must be refused, not QUEUED: the body is one command, so EXEC returns + // exactly one reply. A queued WATCH would make it two. + cmd(&mut a, &["SET", "k", "v1"]); + let exec = text(&cmd(&mut a, &["EXEC"])); + assert!( + exec.starts_with("*1\r\n"), + "a refused WATCH must not be queued — EXEC should return exactly 1 reply, got {exec:?}" + ); +} + +#[test] +fn wc10_a_cross_shard_watch_is_refused() { + let (_g, port, _d) = server(4); + let Some((ka, kb)) = find_cross_shard_pair(port) else { + panic!( + "no cross-shard key pair found in 64 tries at shards=4 — the oracle (a MULTI body spanning shards answering CROSSSLOT) did not fire, so this test cannot prove anything" + ); + }; + + let mut a = connect_ready(port); + cmd(&mut a, &["SET", &ka, "v0"]); + cmd(&mut a, &["SET", &kb, "v0"]); + cmd(&mut a, &["WATCH", &ka, &kb]); + cmd(&mut a, &["MULTI"]); + cmd(&mut a, &["SET", &ka, "from-A"]); + let exec = text(&cmd(&mut a, &["EXEC"])); + + assert!( + exec.contains("CROSSSLOT"), + "watching keys on different shards must be refused loudly — the body commits under one \ + shard's lock, so a version on another shard cannot be validated atomically. Got {exec:?}" + ); + let got = text(&cmd(&mut a, &["GET", &ka])); + assert!( + got.contains("v0"), + "a CROSSSLOT-refused transaction must not have run its body, got {got:?}" + ); +}