fix(transactions): WATCH/UNWATCH now actually guard EXEC on the production dispatch paths - #470
Merged
Merged
Conversation
…ction dispatch paths WATCH and UNWATCH parsed, answered +OK, and recorded versions -- and then the two dispatch paths clients really reach (handler_monoio, handler_sharded) never consulted the watch set at EXEC. Only handler_single (embedded) re-checked. 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 against a server that reported success. Four defects, 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 at the top of the arm) rather than via a clear() each exit path has to remember. 2. A watched key owned by another shard was read from the LOCAL slice -- a different database entirely, so the compared version belonged to some unrelated key or to nothing. WATCH now snapshots versions where the keys live via a new ShardMessage::ReadVersions (grouped per owner: one hop per shard, not per key), the tokens travel with the body over the SPSC hop so the CAS check runs where the body commits, and a watch set spanning shards is classified and refused CROSSSLOT like a cross-shard MULTI body already was. Refusing beats fabricating a conflict the client can never clear. 3. WATCH inside MULTI was queued as an ordinary command instead of 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 (Database::birth_counter), 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 entire restored population, reopening the hole at boot. The WATCH/UNWATCH command arm lives in one new module (src/server/conn/watch.rs) that both production handlers call, rather than as two byte-identical copies. Two copies is how the paths drifted in the first place; a task whose entire subject is that drift should not re-plant it. 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. Residual risk, recorded rather than buried: the ticket shares the entry's 24-bit version field and wraps at 16,777,216 creations (~18.3s of saturated single-database insert at the measured 914,634/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 of 1.0. Only a wider incarnation field retires it; WatchToken stays a named struct so adding one later does not churn the call sites. Full analysis in the task's section 7. 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. Tests: tests/watch_cas_transactions.rs -- 10 wire-level tests, raw-byte assertions (the abort signal IS the type byte, so a reader that renders replies to text would hide the thing under test), two real connections for conflicts, every case run at --shards 1 and --shards 4. Five storage unit tests for the birth counter, each verified RED against a stubbed-constant counter before being accepted green. Nine-combination unit coverage of the locality merge lattice. WATCH/CAS entries added to scripts/test-consistency.sh and scripts/test-commands.sh, which drive the open transaction over /dev/tcp because redis-cli one-shot mode cannot hold a connection across an interleaved second client. Known gap left open (filed in section 7, not fixed here): with disk-offload enabled (opt-in), a watched key that spills cold and is promoted returns through Database::set and draws a fresh ticket, so an eviction the client never asked for aborts its transaction. Fails safe, but a CAS loop on a memory-pressured keyspace can livelock on eviction rather than contention; fixing it requires persisting versions. author: Tin Dang
…ord the ABA mechanism change `add.py check` raised `build_tampered` after the build: the §3 frozen at v1 and the code that shipped had diverged in two places. Recorded here rather than reconciled silently, because editing a frozen contract to match a build is the one move the method forbids. 1. ABA mechanism: per-database DELETE counter -> per-database CREATION ticket. v1's freeze resolution named "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 carrying TTLs bumps the epoch continuously and aborts essentially every WATCH transaction — a correctness fix that makes CAS unusable on exactly the session and cache workloads that need it. Shipped instead: Database::birth_counter, a per-database creation ticket stamped into the entry's existing version field. Same guarantee against delete+recreate, no TTL interaction, no storage change. The numbers were put in front of the human mid-build, before the mechanism was written. 2. Token type: &HashMap<Bytes, u32> -> &HashMap<Bytes, WatchToken>. A newtype over the same u32. The wire contract 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 across the owner hop. The v2 freeze also carries the "Least-sure flag surfaced at freeze" unit the engine requires and v1 was missing (`unflagged_freeze` refused the first crossing attempt). The flag is not boilerplate: v2's guarantee is probabilistic where v1's was total. The creation ticket shares the entry's 24-bit version field and wraps every 16,777,216 creations (~18.3s of saturated single-database insert at the measured 914,634 SET/s), leaving a ~1-in-16.7M miss against v1's pre-fix certainty of 1.0. Retiring that residue means widening Entry, which is a change request back to SPECIFY rather than a patch. Unchanged and fully honored: every wire line in the fenced contract block, the CROSSSLOT rule, and the "watched_keys is empty on EVERY outcome" post-condition. No code changes. The tests->build crossing was re-run so the tripwire witnesses the approved v2 contract instead of a stale v1 hash. author: Tin Dang
|
ⓘ Qodo reviews are paused because the subscription is no longer active. Ask your workspace admin to reactivate the subscription to resume reviews. Manage billing |
📝 WalkthroughWalkthroughThe PR adds Redis-style ChangesWATCH/CAS transactions
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant WATCH_Handler
participant Shard_Coordinator
participant Owner_Shard
participant Transaction_Executor
Client->>WATCH_Handler: WATCH keys
WATCH_Handler->>Shard_Coordinator: snapshot_versions(keys)
Shard_Coordinator->>Owner_Shard: ReadVersions
Owner_Shard-->>Shard_Coordinator: key versions
Shard_Coordinator-->>WATCH_Handler: WatchToken values
Client->>Transaction_Executor: EXEC with watched keys
Transaction_Executor->>Transaction_Executor: validate versions before body
Transaction_Executor-->>Client: commit reply or null abort
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/server/conn/handler_sharded/mod.rs (1)
1103-1141: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftRewrite
WATCHkeys beforetry_handle_watch_unwatch
workspace_rewrite_argsincludesWATCH, but this handler processesWATCHbefore the workspace rewrite. Workspace-bound connections therefore snapshot raw keys while transaction commands use{ws_hex}:<key>.EXECcan validate the wrong version and misclassify same-workspace watched keys by shard. RewriteWATCHarguments before capturing versions.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_sharded/mod.rs` around lines 1103 - 1141, Move workspace argument rewriting before the `write::try_handle_multi_exec` path so `WATCH` keys are transformed before `try_handle_watch_unwatch` captures versions. Ensure transaction handling receives the rewritten `cmd_args`, while preserving the existing workspace prefixing and shard-routing behavior for subsequent dispatch.
🧹 Nitpick comments (5)
src/server/conn/handler_single.rs (1)
1557-1590: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAlign
handler_singlewithtry_handle_watch_unwatch.The inline arm differs from
src/server/conn/watch.rs:
WATCHchecksconn.in_multibefore argument count.- It accepts only
Frame::BulkString, while the helper also acceptsFrame::SimpleString.UNWATCH keyreturns+OKinstead of an arity error.Extract the shared validation and key extraction. Keep version snapshotting path-specific. Add consistency cases for these inputs; the current script does not cover them.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_single.rs` around lines 1557 - 1590, Align the inline WATCH/UNWATCH handling in handler_single with try_handle_watch_unwatch: reuse shared validation and key extraction so WATCH checks MULTI before arity and accepts both BulkString and SimpleString keys, while UNWATCH rejects arguments with the same arity error as the helper. Keep version snapshotting in the inline WATCH path, and add coverage for these differing inputs.src/shard/spsc_handler.rs (1)
2131-2138: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winClamp
db_indexlike the sibling arms.Every other arm in this function narrows the incoming index with
db_index.min(db_count.saturating_sub(1))before touchings.databases. This arm passes the rawdb_indextowith_shard_db. Ifwith_shard_dbindexes the database vector directly, an out-of-range value panics the shard thread and aborts the process.♻️ Proposed clamp
let crate::shard::dispatch::ReadVersionsPayload { db_index, keys, reply_tx, } = *payload; - let versions = crate::shard::slice::with_shard_db(db_index, |db| { + let db_idx = db_index.min(shard_databases.db_count().saturating_sub(1)); + let versions = crate::shard::slice::with_shard_db(db_idx, |db| { keys.iter().map(|k| db.get_version(k)).collect::<Vec<u32>>() });Run the following script to confirm the bounds behaviour of
with_shard_db:#!/bin/bash # Description: Inspect with_shard_db for index bounds handling. ast-grep run --pattern 'pub fn with_shard_db($$$) { $$$ }' --lang rust src/shard/slice.rs rg -n -C 8 'fn with_shard_db' src/shard/slice.rs🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/shard/spsc_handler.rs` around lines 2131 - 2138, Clamp db_index to db_index.min(db_count.saturating_sub(1)) in the ReadVersionsPayload arm before passing it to with_shard_db, matching the sibling arms and preventing out-of-range database access.src/storage/db/mod.rs (1)
1025-1038: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a test that
clear()does not reset the ticket counter.
Database::clear(FLUSHDB / FLUSHALL / recovery wipe) resetsdata,used_memory, andhot_keys, but deliberately leavesbirth_counteralone. That is the correct behaviour: resetting it would make the first key created after a flush draw ticket 1 again and re-open the ABA hole for any client that was watching a key at flush time. No test pins that invariant today, so a future cleanup ofclear()could silently undo it.💚 Proposed test
/// `clear()` must NOT rewind the ticket dispenser: a key created after a /// FLUSHDB must not present a token an earlier watcher recorded. #[test] fn test_clear_does_not_rewind_the_birth_counter() { 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.clear(); db.set_string(Bytes::from_static(b"k"), Bytes::from_static(b"v1")); assert_ne!( db.get_version(b"k"), watched, "a key recreated after FLUSHDB presented the version WATCH recorded" ); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/db/mod.rs` around lines 1025 - 1038, Add a regression test alongside test_birth_ticket_wraps_without_ever_yielding_zero that creates and watches a key, calls Database::clear, recreates the key, and asserts the new get_version result differs from the previously recorded version, pinning that clear does not reset birth_counter.src/storage/db/accessors.rs (1)
83-91: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueOptional: extract the repeated fabricate-and-stamp block.
The same four lines appear at Lines 83-91, 211-219, 272-280, and 335-343: stamp the birth version, build the
CompactKey, chargeentry_overhead, insert. A small private helper keeps the stamping rule in one place, so a future fabrication site cannot forget it.♻️ Proposed helper
/// Insert a freshly fabricated entry: stamp the creation ticket and charge /// its overhead. The single place that knows a new incarnation must be /// observably distinct from the one that occupied the key before. fn insert_fresh(&mut self, key: &[u8], mut entry: Entry) { entry.set_version(self.next_birth_version()); self.used_memory += entry_overhead(key, &entry); self.data.insert(CompactKey::from(key), entry); }if !self.data.contains_key(key) { - 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); + self.insert_fresh(key, K::new_entry()); }Also applies to: 211-219, 272-280, 335-343
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/storage/db/accessors.rs` around lines 83 - 91, Extract the repeated fresh-entry fabrication logic into a private helper on the containing accessor type, such as insert_fresh, that stamps next_birth_version, updates used_memory via entry_overhead, and inserts using CompactKey::from. Replace the duplicated four-line blocks at all four fabrication sites with calls to this helper, preserving their existing behavior..add/tasks/watch-cas-transactions/TASK.md (1)
186-190: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAdd an absolute correctness assertion to the path-parity scenario.
Byte-identical replies do not prove correctness when both paths return the same wrong result. The task records that
wc7_all_dispatch_paths_agreepassed before the fix for this reason. AssertEXEC -> Nulland preservation of the conflicting value on each path, then compare the replies.Also applies to: 318-327
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In @.add/tasks/watch-cas-transactions/TASK.md around lines 186 - 190, Update the path-parity scenario around the WATCH/MULTI/EXEC sequence to assert the expected absolute behavior on each dispatch path: EXEC must return Null and the conflicting value must remain preserved. Keep the existing byte-identical reply comparison across embedded, shards=1, and shards=4 after these per-path correctness assertions.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In @.add/state.json:
- Around line 290-296: Move every approved path from the Scope (may touch)
declaration in .add/tasks/watch-cas-transactions/TASK.md (lines 341-345) onto
its first declaration line, preserving the complete scope for parsing. Then
rerun the appropriate add.py workflow to regenerate the tests-to-build snapshot;
do not edit .add/state.json manually.
In @.add/tasks/watch-cas-transactions/TASK.md:
- Around line 477-485: Update disk-offload promotion flows, especially
promote_inflight_if_present and promote_cold_outcome, so the promoted entry
inherits its pre-spill watch/version ticket instead of receiving a fresh one
through Database::set. Preserve WATCH/EXEC behavior without client writes, and
add a regression test covering read-only promotion and CAS retries.
- Around line 249-257: Update the frozen contract to use the per-database
creation-ticket mechanism consistently: replace the remaining per-database
delete-counter instruction near the related implementation guidance with
Database::birth_counter stamped into entry versions. Remove any wording that
prescribes or implies delete-epoch updates on deletion or TTL expiry, while
preserving the documented ABA guarantee and wraparound behavior.
In @.add/tooling/add.py:
- Around line 1770-1780: Update _contract_status to catch UnicodeDecodeError
when reading malformed TASK.md content, alongside the existing OSError handling.
Return the same failed-check representation used for unreadable contract files
so add.py check reports the issue instead of crashing.
- Around line 1725-1738: Update _contract_status to catch UnicodeError alongside
OSError while reading and decoding TASK.md, returning None so malformed UTF-8 is
handled as a failed contract check without propagating a traceback.
In `@CHANGELOG.md`:
- Around line 105-111: Update the changelog entry to mention the disk-offload
WATCH limitation: promotion assigns a fresh creation ticket and may abort EXEC
without any client write. Add this as a concise deployment note or qualify the
stated WATCH guarantee, preserving the existing residual-risk details.
In `@scripts/test-consistency.sh`:
- Around line 597-613: scripts/test-consistency.sh:597-613 (anchor) update
watch_cas_outcome to append an ECHO cas-armed barrier after WATCH and drain fd 3
until it returns before the conflicting SET. scripts/test-consistency.sh:624-637
add the same barrier and drain to watch_cas_aba_outcome before the DEL/SET pair.
scripts/test-consistency.sh:643-655 add it to watch_unwatch_outcome before the
interleaved SET, also ensuring UNWATCH has been processed.
scripts/test-commands.sh:732-748 apply the barrier to watch_cas_outcome and emit
__WATCH_BARRIER_TIMEOUT__ if draining times out.
In `@src/server/conn/watch.rs`:
- Around line 41-62: Update the watch-token insertion in the WATCH handling flow
to use the map entry API, inserting a new WatchToken only when the key is not
already present. Preserve the earliest snapshot version for duplicate keys so a
later WATCH does not overwrite the existing dependency.
In `@src/shard/coordinator.rs`:
- Around line 318-331: In the error branch of recv_reply_bounded within the
WATCH version-fetch flow, add a warning log and invoke
record_xshard_reply_timeout with the "watch_versions" operation label. Keep the
existing successful version assignment unchanged and preserve the current
fallback where affected slots remain zero.
In `@src/shard/spsc_handler.rs`:
- Around line 2127-2140: The ReadVersions handling currently bypasses FIFO
ordering by falling into other_messages, which are processed after writes.
Update drain_spsc_shared to explicitly route ShardMessage::ReadVersions into the
ordered execute_batch alongside PipelineBatchSlotted and ExecuteSlotted, while
preserving the existing ReadVersions snapshot and reply behavior.
In `@tests/watch_cas_transactions.rs`:
- Around line 1-36: Update the module header in watch_cas_transactions.rs to
describe the implemented WATCH behavior and the invariants protected by the test
suite, removing stale claims about unknown commands, missing production paths,
and tests expected to fail. Revise the wc6 comment to state that
delete-and-recreate changes the key’s identity/version and must abort a
transaction, reflecting the per-database creation ticket in kv_ops.rs.
- Around line 121-135: Update read_reply to remove the fixed sleep and read
repeatedly until a complete RESP reply is framed by its terminating CRLF,
preserving all bytes needed for that single reply. Treat Ok(0) as a
panic/failure instead of returning an empty buffer, while continuing to panic on
read errors; keep is_null unchanged and return only the complete framed
response.
---
Outside diff comments:
In `@src/server/conn/handler_sharded/mod.rs`:
- Around line 1103-1141: Move workspace argument rewriting before the
`write::try_handle_multi_exec` path so `WATCH` keys are transformed before
`try_handle_watch_unwatch` captures versions. Ensure transaction handling
receives the rewritten `cmd_args`, while preserving the existing workspace
prefixing and shard-routing behavior for subsequent dispatch.
---
Nitpick comments:
In @.add/tasks/watch-cas-transactions/TASK.md:
- Around line 186-190: Update the path-parity scenario around the
WATCH/MULTI/EXEC sequence to assert the expected absolute behavior on each
dispatch path: EXEC must return Null and the conflicting value must remain
preserved. Keep the existing byte-identical reply comparison across embedded,
shards=1, and shards=4 after these per-path correctness assertions.
In `@src/server/conn/handler_single.rs`:
- Around line 1557-1590: Align the inline WATCH/UNWATCH handling in
handler_single with try_handle_watch_unwatch: reuse shared validation and key
extraction so WATCH checks MULTI before arity and accepts both BulkString and
SimpleString keys, while UNWATCH rejects arguments with the same arity error as
the helper. Keep version snapshotting in the inline WATCH path, and add coverage
for these differing inputs.
In `@src/shard/spsc_handler.rs`:
- Around line 2131-2138: Clamp db_index to
db_index.min(db_count.saturating_sub(1)) in the ReadVersionsPayload arm before
passing it to with_shard_db, matching the sibling arms and preventing
out-of-range database access.
In `@src/storage/db/accessors.rs`:
- Around line 83-91: Extract the repeated fresh-entry fabrication logic into a
private helper on the containing accessor type, such as insert_fresh, that
stamps next_birth_version, updates used_memory via entry_overhead, and inserts
using CompactKey::from. Replace the duplicated four-line blocks at all four
fabrication sites with calls to this helper, preserving their existing behavior.
In `@src/storage/db/mod.rs`:
- Around line 1025-1038: Add a regression test alongside
test_birth_ticket_wraps_without_ever_yielding_zero that creates and watches a
key, calls Database::clear, recreates the key, and asserts the new get_version
result differs from the previously recorded version, pinning that clear does not
reset birth_counter.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 4eba132b-bedc-47dc-86d9-9edf1c542b9f
📒 Files selected for processing (22)
.add/state.json.add/tasks/watch-cas-transactions/TASK.md.add/tooling/add.pyCHANGELOG.mdscripts/test-commands.shscripts/test-consistency.shsrc/server/conn/core.rssrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_monoio/write.rssrc/server/conn/handler_sharded/mod.rssrc/server/conn/handler_sharded/write.rssrc/server/conn/handler_single.rssrc/server/conn/mod.rssrc/server/conn/shared.rssrc/server/conn/watch.rssrc/shard/coordinator.rssrc/shard/dispatch.rssrc/shard/spsc_handler.rssrc/storage/db/accessors.rssrc/storage/db/kv_ops.rssrc/storage/db/mod.rstests/watch_cas_transactions.rs
Comment on lines
+290
to
+296
| "scope": { | ||
| "declared": [ | ||
| "src/server/conn/shared.rs", | ||
| "src/server/conn/watch.rs", | ||
| "src/server/conn/core.rs" | ||
| ], | ||
| "snapshot_md5": "fe6a109ebabb70ce61c2310ddd28f5e2" |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- relevant files ---'
git ls-files .add/state.json .add/tasks/watch-cas-transactions/TASK.md '*add.py' '*tests-to-build*' | sed -n '1,120p'
printf '%s\n' '--- add.py references ---'
rg -n -C 4 'scope\.declared|snapshot_md5|tests-to-build|out.of.scope|declared' .add scripts . 2>/dev/null | sed -n '1,240p'
printf '%s\n' '--- state scope ---'
python3 - <<'PY'
import json
from pathlib import Path
p = Path('.add/state.json')
data = json.loads(p.read_text())
scope = data.get('scope', {})
print(json.dumps(scope, indent=2))
PY
printf '%s\n' '--- task scope ---'
python3 - <<'PY'
from pathlib import Path
p = Path('.add/tasks/watch-cas-transactions/TASK.md')
lines = p.read_text().splitlines()
for i in range(338, 348):
print(f'{i+1}: {lines[i]}')
PYRepository: pilotspace/moon
Length of output: 33543
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- state task entry ---'
sed -n '250,320p' .add/state.json
printf '%s\n' '--- add.py scope implementation ---'
rg -n -C 8 'def .*scope|scope.*declared|snapshot_md5|_scope_walk|tests.*build|advance' .add/tooling/add.py | sed -n '1,320p'
printf '%s\n' '--- task scope rules and amendment ---'
sed -n '230,255p' .add/tasks/watch-cas-transactions/TASK.md
sed -n '339,365p' .add/tasks/watch-cas-transactions/TASK.md
printf '%s\n' '--- all references to this task state and scope ---'
rg -n -C 3 'watch-cas-transactions|scope\.declared|snapshot_md5' .add/tooling/add.py .add/tasks/watch-cas-transactions .add/state.json | sed -n '1,300p'Repository: pilotspace/moon
Length of output: 27762
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- complete scope resolver and guard ---'
sed -n '2938,3075p' .add/tooling/add.py
printf '%s\n' '--- scope documentation and amendment rules ---'
rg -n -C 6 'Scope AMENDED|scope.*amend|amend.*scope|tests.*build.*scope|build-scope-lock|scope gate|scope-snapshot' .add/docs .add/tooling .add/tasks/watch-cas-transactions .add/CONVENTIONS.md | sed -n '1,360p'
printf '%s\n' '--- scope sidecar status ---'
git ls-files .add/tasks/watch-cas-transactions/scope-snapshot.json
if [ -f .add/tasks/watch-cas-transactions/scope-snapshot.json ]; then
python3 - <<'PY'
import json
from pathlib import Path
p = Path('.add/tasks/watch-cas-transactions/scope-snapshot.json')
data = json.loads(p.read_text())
files = data.get('files', {})
print('sidecar_version:', data.get('version'))
print('sidecar_file_count:', len(files))
print('sidecar_contains_amended_paths:')
for name in [
'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',
]:
print(name, any(k == name or k.startswith(name) for k in files))
PY
fi
printf '%s\n' '--- working-tree changes ---'
git status --shortRepository: pilotspace/moon
Length of output: 22473
🏁 Script executed:
#!/bin/bash
set -eu
python3 - <<'PY'
import json
import re
from pathlib import Path
task = Path('.add/tasks/watch-cas-transactions/TASK.md').read_text(encoding='utf-8')
state = json.loads(Path('.add/state.json').read_text(encoding='utf-8'))
anchor = state['tasks']['watch-cas-transactions']['scope']
match = re.search(r'^\s*Scope \(may touch\):.*$', task, re.M)
if not match:
raise SystemExit('scope declaration not found')
# Model _declared_scope's relevant parsing rule: tokens come from the
# first matching line only. Resolve root-relative slash-containing tokens.
parsed = []
for token in re.findall(r'`([^`]+)`', match.group(0)):
token = token.strip()
resolved = token[2:] if token.startswith('./') else token
if resolved not in parsed:
parsed.append(resolved)
all_scope_tokens = re.findall(
r'`([^`]+)`',
task[match.start():task.find('\n\n', match.start()) if '\n\n' in task[match.start():] else len(task)]
)
print('first_scope_line:', match.group(0))
print('parsed_declared:', json.dumps(parsed))
print('state_declared:', json.dumps(anchor['declared']))
print('parsed_equals_state:', parsed == anchor['declared'])
print('tokens_in_scope_block:', len(all_scope_tokens))
print('tokens_ignored_after_first_line:', all_scope_tokens[len(parsed):])
PY
printf '%s\n' '--- heal/re-advance behavior ---'
rg -n -C 12 '_heal_or_escalate|phase.*build|re-advance through tests|scope_violation' .add/tooling/add.py | sed -n '1,260p'Repository: pilotspace/moon
Length of output: 15743
Place the complete scope declaration on the first line.
add.py parses only backticked tokens on the first Scope (may touch): line. scope.declared matches the three paths currently parsed. The paths on lines 342–345 are ignored and will not pass the scope gate.
Move all approved paths to the first declaration line, then re-establish the tests-to-build snapshot. Do not edit .add/state.json manually.
📍 Affects 2 files
.add/state.json#L290-L296(this comment).add/tasks/watch-cas-transactions/TASK.md#L341-L345
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.add/state.json around lines 290 - 296, Move every approved path from the
Scope (may touch) declaration in .add/tasks/watch-cas-transactions/TASK.md
(lines 341-345) onto its first declaration line, preserving the complete scope
for parsing. Then rerun the appropriate add.py workflow to regenerate the
tests-to-build snapshot; do not edit .add/state.json manually.
Comment on lines
+249
to
+257
| 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 |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Use one ABA mechanism throughout the frozen contract.
Lines 249-257 define the shipped Database::birth_counter creation-ticket mechanism and reject a delete epoch because TTL expiry would cause false aborts. Line 375 still prescribes a per-database delete counter. This contradiction can send a later implementation back to the rejected design. Replace every remaining delete-counter instruction with the creation-ticket mechanism before treating v2 as frozen.
Proposed wording
-5. ABA fix (pending the §3 freeze decision): per-database delete counter consulted alongside the version.
+5. ABA fix: use the per-database creation ticket (`Database::birth_counter`) as the incarnation component.Also applies to: 375-375
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.add/tasks/watch-cas-transactions/TASK.md around lines 249 - 257, Update the
frozen contract to use the per-database creation-ticket mechanism consistently:
replace the remaining per-database delete-counter instruction near the related
implementation guidance with Database::birth_counter stamped into entry
versions. Remove any wording that prescribes or implies delete-epoch updates on
deletion or TTL expiry, while preserving the documented ABA guarantee and
wraparound behavior.
Comment on lines
+477
to
+485
| - [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`). |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Preserve the watch token during disk-offload promotion.
Promotion through Database::set assigns a fresh creation ticket. A read-only promotion can therefore change the watched token and abort EXEC without a client write. Repeated retries can livelock a CAS client. Preserve the pre-spill ticket during promotion, or explicitly gate WATCH for this case and add a regression test.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.add/tasks/watch-cas-transactions/TASK.md around lines 477 - 485, Update
disk-offload promotion flows, especially promote_inflight_if_present and
promote_cold_outcome, so the promoted entry inherits its pre-spill watch/version
ticket instead of receiving a fresh one through Database::set. Preserve
WATCH/EXEC behavior without client writes, and add a regression test covering
read-only promotion and CAS retries.
Comment on lines
+1725
to
+1738
| 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 |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- target function and callers ---'
sed -n '1690,1775p' .add/tooling/add.py
rg -n -C 4 '_contract_status|contract status|Status:' .add/tooling/add.py .add/state.json .add/tasks tasks 2>/dev/null || true
printf '%s\n' '--- task files and state records ---'
if [ -d tasks ]; then
find tasks -name TASK.md -print | sort
fi
if [ -f .add/state.json ]; then
sed -n '1,240p' .add/state.json
fi
printf '%s\n' '--- deterministic decode probe ---'
python3 - <<'PY'
from pathlib import Path
import tempfile
with tempfile.TemporaryDirectory() as d:
p = Path(d) / "TASK.md"
p.write_bytes(b"Status: FROZEN\\n\\xff\\n")
try:
p.read_text(encoding="utf-8")
except Exception as exc:
print(type(exc).__name__, isinstance(exc, OSError), isinstance(exc, UnicodeError))
PYRepository: pilotspace/moon
Length of output: 50373
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- target function and callers ---'
sed -n '1690,1775p' .add/tooling/add.py
rg -n -C 4 '_contract_status|contract status|Status:' .add/tooling/add.py .add/state.json .add/tasks tasks 2>/dev/null || true
printf '%s\n' '--- task files and state records ---'
if [ -d tasks ]; then
find tasks -name TASK.md -print | sort
fi
if [ -f .add/state.json ]; then
sed -n '1,240p' .add/state.json
fi
printf '%s\n' '--- deterministic decode probe ---'
python3 - <<'PY'
from pathlib import Path
import tempfile
with tempfile.TemporaryDirectory() as d:
p = Path(d) / "TASK.md"
p.write_bytes(b"Status: FROZEN\n\xff\n")
try:
p.read_text(encoding="utf-8")
except Exception as exc:
print(type(exc).__name__, isinstance(exc, OSError), isinstance(exc, UnicodeError))
PYRepository: pilotspace/moon
Length of output: 50373
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- check aggregation and phase definitions ---'
sed -n '1,120p' .add/tooling/add.py
sed -n '1755,1815p' .add/tooling/add.py
rg -n 'PHASES|failed|checks' .add/tooling/add.py | head -80
printf '%s\n' '--- state task phases and parsed Status values ---'
python3 - <<'PY'
import json
from pathlib import Path
state = json.loads(Path(".add/state.json").read_text(encoding="utf-8"))
phases = ["specify", "scenarios", "contract", "tests", "build", "verify", "done"]
tasks = state.get("tasks", {})
for slug, record in tasks.items():
phase = record.get("phase")
if phase not in phases[phases.index("tests"):]:
continue
path = Path(".add") / "tasks" / slug / "TASK.md"
status_lines = []
try:
for line in path.read_text(encoding="utf-8").splitlines():
if line.startswith("Status:"):
status_lines.append(line)
except Exception as exc:
print(f"{slug}\tphase={phase!r}\tread_error={type(exc).__name__}")
continue
print(f"{slug}\tphase={phase!r}\tstatus={status_lines!r}")
print(f"total_tasks={len(tasks)}")
PY
printf '%s\n' '--- deterministic exception-class probe ---'
python3 - <<'PY'
from pathlib import Path
import tempfile
with tempfile.TemporaryDirectory() as d:
p = Path(d) / "TASK.md"
p.write_bytes(b"Status: FROZEN\n\xff\n")
try:
p.read_text(encoding="utf-8")
except Exception as exc:
print({
"exception": type(exc).__name__,
"is_OSError": isinstance(exc, OSError),
"is_UnicodeError": isinstance(exc, UnicodeError),
"is_caught_by_original": isinstance(exc, OSError),
"is_caught_by_proposed": isinstance(exc, (OSError, UnicodeError)),
})
PYRepository: pilotspace/moon
Length of output: 17945
Catch UnicodeError when reading TASK.md.
Malformed UTF-8 raises UnicodeDecodeError, which is not an OSError. Without this catch, add.py check raises a traceback instead of recording a failed contract check.
Proposed fix
- except OSError:
+ except (OSError, UnicodeError):📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| 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 | |
| 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, UnicodeError): | |
| return None |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.add/tooling/add.py around lines 1725 - 1738, Update _contract_status to
catch UnicodeError alongside OSError while reading and decoding TASK.md,
returning None so malformed UTF-8 is handled as a failed contract check without
propagating a traceback.
Comment on lines
+1770
to
+1780
| # 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")) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
python - <<'PY'
import json
from pathlib import Path
state = json.loads(Path(".add/state.json").read_text(encoding="utf-8"))
phases = ("tests", "build", "verify", "observe", "done")
for slug, task in state.get("tasks", {}).items():
if task.get("phase") not in phases:
continue
path = Path(".add") / "tasks" / slug / "TASK.md"
text = path.read_text(encoding="utf-8") if path.is_file() else ""
statuses = [
line[len("Status:"):].strip().split()[0].strip("*").upper()
for line in text.splitlines()
if line.startswith("Status:") and line[len("Status:"):].strip()
]
if not any(status in {"FROZEN", "INHERITED"} for status in statuses):
print(f"{slug}: {task.get('phase')} -> missing frozen contract status")
PYRepository: pilotspace/moon
Length of output: 153
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '%s\n' '--- _contract_status definition and callers ---'
rg -n -A30 -B8 'def _contract_status|_contract_status\(' .add/tooling/add.py
printf '%s\n' '--- relevant task-file encoding check ---'
python3 - <<'PY'
import json
from pathlib import Path
state = json.loads(Path(".add/state.json").read_text(encoding="utf-8"))
phases = {"tests", "build", "verify", "observe", "done"}
for slug, task in state.get("tasks", {}).items():
if task.get("phase") not in phases:
continue
path = Path(".add") / "tasks" / slug / "TASK.md"
if path.is_file():
try:
path.read_text(encoding="utf-8")
except UnicodeDecodeError as exc:
print(f"{slug}: {task.get('phase')} -> invalid UTF-8 ({exc})")
else:
print(f"{slug}: {task.get('phase')} -> missing TASK.md")
PYRepository: pilotspace/moon
Length of output: 5001
🏁 Script executed:
python3 - <<'PY'
from pathlib import Path
from tempfile import TemporaryDirectory
with TemporaryDirectory() as d:
path = Path(d) / "TASK.md"
path.write_bytes(b"Status: FROZEN\n\xff\n")
try:
path.read_text(encoding="utf-8")
except OSError:
print("caught as OSError")
except UnicodeDecodeError:
print("UnicodeDecodeError escapes the OSError handler")
PYRepository: pilotspace/moon
Length of output: 200
Handle invalid UTF-8 in _contract_status.
TASK.md with malformed UTF-8 causes add.py check to crash because UnicodeDecodeError is not caught by except OSError. Catch it and report a failed check instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.add/tooling/add.py around lines 1770 - 1780, Update _contract_status to
catch UnicodeDecodeError when reading malformed TASK.md content, alongside the
existing OSError handling. Return the same failed-check representation used for
unreadable contract files so add.py check reports the issue instead of crashing.
Comment on lines
+41
to
+62
| let keys: Vec<Bytes> = 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 }); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Preserve the earliest token when a key is watched twice.
insert overwrites an existing WatchToken for the same key. If a client runs WATCH k, then a conflicting write lands, then the client runs WATCH k again before MULTI, the second snapshot records the post-write version and the conflict is lost. Redis keeps the dependency in that case and aborts EXEC. Use an entry API so the first recorded token wins.
🐛 Proposed fix to keep the first recorded token
for (k, v) in keys.into_iter().zip(versions) {
- conn.watched_keys.insert(k, WatchToken { version: v });
+ // First token wins: a re-WATCH of the same key must not
+ // refresh a dependency that a concurrent write already broke.
+ conn.watched_keys.entry(k).or_insert(WatchToken { version: v });
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| let keys: Vec<Bytes> = 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 }); | |
| } | |
| let keys: Vec<Bytes> = 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) { | |
| // First token wins: a re-WATCH of the same key must not | |
| // refresh a dependency that a concurrent write already broke. | |
| conn.watched_keys.entry(k).or_insert(WatchToken { version: v }); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/server/conn/watch.rs` around lines 41 - 62, Update the watch-token
insertion in the WATCH handling flow to use the map entry API, inserting a new
WatchToken only when the key is not already present. Preserve the earliest
snapshot version for duplicate keys so a later WATCH does not overwrite the
existing dependency.
Comment on lines
+318
to
+331
| 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; | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Check the signature and existing labels of record_xshard_reply_timeout.
set -euo pipefail
rg -nP --type=rust -C 8 '\bfn record_xshard_reply_timeout\s*\(' src
rg -nP --type=rust -n 'record_xshard_reply_timeout\s*\(' srcRepository: pilotspace/moon
Length of output: 2071
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- coordinator symbols and targeted range ---'
ast-grep outline src/shard/coordinator.rs --match 'fn' --view compact || true
sed -n '270,345p' src/shard/coordinator.rs
echo '--- reply helper definitions and uses ---'
rg -n -C 10 'recv_reply_bounded|ReadVersionsPayload|ReadVersions' src/shard src/server
echo '--- WATCH response handling ---'
rg -n -C 12 'WATCH|watch' src/server/conn/handler_sharded src/server/conn/handler_monoio src/shard | head -n 500
echo '--- timeout metric definition and all labels ---'
sed -n '920,955p' src/admin/metrics_setup.rs
rg -n -C 8 'record_xshard_reply_timeout\(' srcRepository: pilotspace/moon
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- WATCH command paths ---'
rg -n -i -C 16 'watch' src/server/conn/handler_sharded/write.rs src/server/conn/handler_monoio/write.rs src/server/conn/handler_sharded/mod.rs src/server/conn/handler_monoio/mod.rs src/shard/coordinator.rs
echo '--- watched state and EXEC conflict checks ---'
rg -n -C 14 'watched|watch_versions|snapshot_versions|version.*mismatch|mismatch.*version|EXEC.*abort|abort.*EXEC' src/server src/shard
echo '--- bounded reply implementation ---'
sed -n '185,220p' src/shard/coordinator.rs
echo '--- relevant metric call sites ---'
sed -n '2275,2310p' src/server/conn/handler_sharded/mod.rs
sed -n '2425,2450p' src/server/conn/handler_sharded/mod.rsRepository: pilotspace/moon
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- watch module files ---'
fd -i 'watch' src/server src | head -n 50
echo '--- shared WATCH implementation ---'
rg -n -C 25 'try_handle_watch_unwatch|snapshot_versions|watched_keys' src/server/conn/watch.rs src/server/conn/shared.rs src/server/conn
echo '--- CAS check implementation ---'
rg -n -C 20 'watched|watch' src/server/conn/handler_sharded/write.rs src/server/conn/handler_monoio/write.rs src/server/conn/handler_sharded/txn.rs src/server/conn/handler_monoio/txn.rs src/shard/coordinator.rs | rg -n 'execute_transaction_sharded|watched|version|abort|conflict|snapshot' | head -n 300Repository: pilotspace/moon
Length of output: 50371
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
import re
coordinator = Path("src/shard/coordinator.rs").read_text()
watch = Path("src/server/conn/watch.rs").read_text()
shared = Path("src/server/conn/shared.rs").read_text()
metrics = Path("src/admin/metrics_setup.rs").read_text()
assert re.search(r'let mut out\s*=\s*vec!\[0u32;\s*keys\.len\(\)\]', coordinator)
assert re.search(
r'if let Ok\(versions\)\s*=\s*recv_reply_bounded\(reply_rx\)\.await\s*\{',
coordinator,
)
assert re.search(
r'for \(k,\s*v\) in keys\.into_iter\(\)\.zip\(versions\)\s*\{.*?'
r'conn\.watched_keys\.insert\(k,\s*WatchToken\s*\{\s*version:\s*v\s*\}\);',
watch,
re.S,
)
assert 'responses.push(Frame::SimpleString(Bytes::from_static(b"OK")));' in watch
assert re.search(
r'pub fn record_xshard_reply_timeout\(kind:\s*&\'static str\)',
metrics,
)
assert re.search(
r'if current_version != watched_version\.version\s*\{.*?'
r'return \(Frame::Null, Vec::new\(\), Vec::new\(\)\);',
shared,
re.S,
)
print("snapshot_versions initializes missing remote replies to zero")
print("WATCH stores those versions and returns OK")
print("owner-side EXEC rechecks the token and aborts on mismatch")
print("record_xshard_reply_timeout accepts the proposed static label")
PYRepository: pilotspace/moon
Length of output: 377
Instrument failed remote WATCH snapshots.
When recv_reply_bounded returns Err, affected slots remain 0, while WATCH returns +OK. Add a warning and record_xshard_reply_timeout("watch_versions") in the error branch.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/shard/coordinator.rs` around lines 318 - 331, In the error branch of
recv_reply_bounded within the WATCH version-fetch flow, add a warning log and
invoke record_xshard_reply_timeout with the "watch_versions" operation label.
Keep the existing successful version assignment unchanged and preserve the
current fallback where affected slots remain zero.
Comment on lines
+2127
to
+2140
| 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::<Vec<u32>>() | ||
| }); | ||
| let _ = reply_tx.send(versions); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy lift
ReadVersions is drained out of FIFO order with writes, which can lose a WATCH conflict.
drain_spsc_shared has no explicit arm for ShardMessage::ReadVersions, so it falls into the _ => other_messages.push(msg) arm at Line 306. other_messages is processed at Lines 355-383, after the whole execute_batch loop at Lines 318-348.
A ReadVersions message that arrived on the ring before a PipelineBatchSlotted / ExecuteSlotted write to the same key is therefore answered after that write is applied. The snapshot returns the post-write version, EXEC sees no change, and the transaction commits over a write that landed after WATCH. Redis aborts in that case.
Route ReadVersions through the same ordered batch as the execute messages.
🐛 Proposed fix to keep the snapshot ordered with writes
ShardMessage::Execute { .. }
| ShardMessage::PipelineBatch { .. }
| ShardMessage::MultiExecute { .. }
| ShardMessage::TxnExecute(_)
+ | ShardMessage::ReadVersions(_)
| ShardMessage::ExecuteSlotted { .. }
| ShardMessage::PipelineBatchSlotted { .. }Run the following script to confirm that no other arm re-orders ReadVersions:
#!/bin/bash
# Description: Locate every routing and handling site for ReadVersions.
rg -n -C 6 'ReadVersions' --type=rust
# Confirm the two-phase drain order in the SPSC drain.
rg -n -C 4 'other_messages.push|execute_batch.push|for msg in (execute_batch|other_messages)' src/shard/spsc_handler.rs🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/shard/spsc_handler.rs` around lines 2127 - 2140, The ReadVersions
handling currently bypasses FIFO ordering by falling into other_messages, which
are processed after writes. Update drain_spsc_shared to explicitly route
ShardMessage::ReadVersions into the ordered execute_batch alongside
PipelineBatchSlotted and ExecuteSlotted, while preserving the existing
ReadVersions snapshot and reply behavior.
Comment on lines
+1
to
+36
| //! 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 |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Update the module header and the wc6 comment: they describe the pre-fix state as current.
The header says WATCH replies -ERR unknown command, that WATCH exists only in handler_single.rs, and lists eight tests as "Expected RED on main". This PR implements all of it, so the text now describes behaviour that no longer exists. The same applies to the comment at Lines 351-354, which states that DEL + re-SET returns the key to version 1 — the per-db creation ticket added in src/storage/db/kv_ops.rs removes that.
Restate both as the invariant each test protects.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/watch_cas_transactions.rs` around lines 1 - 36, Update the module
header in watch_cas_transactions.rs to describe the implemented WATCH behavior
and the invariants protected by the test suite, removing stale claims about
unknown commands, missing production paths, and tests expected to fail. Revise
the wc6 comment to state that delete-and-recreate changes the key’s
identity/version and must abort a transaction, reflecting the per-database
creation ticket in kv_ops.rs.
Comment on lines
+121
to
+135
| fn read_reply(s: &mut TcpStream) -> Vec<u8> { | ||
| // 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") | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
read_reply can return a partial or empty reply and turn a real failure into a pass.
The helper sleeps 60 ms and then performs one read. Three problems follow:
- A short read returns a truncated reply, and every later
cmdin that test reads the leftover bytes. All subsequent assertions in the test are then misaligned. readreturningOk(0)(peer closed, or the server crashed mid-test) yields an empty buffer.is_null(&[])isfalse, sowc2,wc4, andwc5— which assert!is_null(&exec)— pass on no reply at all. That is a false green on the exact behaviour under test.- The fixed 60 ms sleep adds roughly 60 ms per command to the whole suite.
Read until a complete RESP reply is framed, and fail on EOF.
💚 Proposed fix: read until a terminator, and reject EOF
fn read_reply(s: &mut TcpStream) -> Vec<u8> {
- // 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}"),
- }
+ // Read until the reply is terminated. A single bounded read can return a
+ // partial reply, which desynchronizes every later command on this socket,
+ // and an EOF read returns empty bytes that `is_null` reports as "not a
+ // null" — a false pass on the abort assertions.
+ let mut out: Vec<u8> = Vec::with_capacity(4096);
+ let mut chunk = [0u8; 8192];
+ loop {
+ match s.read(&mut chunk) {
+ Ok(0) => panic!("connection closed before a reply arrived: {:?}", text(&out)),
+ Ok(n) => {
+ out.extend_from_slice(&chunk[..n]);
+ // Every reply this suite issues is a single RESP value and
+ // ends with CRLF; a multi-element EXEC array arrives whole
+ // because the server writes one batch per command.
+ if out.ends_with(b"\r\n") {
+ return out;
+ }
+ }
+ Err(e) => panic!("read reply: {e}"),
+ }
+ }
}The RESP3 null _\r\n also ends with CRLF, so is_null keeps working.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/watch_cas_transactions.rs` around lines 121 - 135, Update read_reply to
remove the fixed sleep and read repeatedly until a complete RESP reply is framed
by its terminating CRLF, preserving all bytes needed for that single reply.
Treat Ok(0) as a panic/failure instead of returning an empty buffer, while
continuing to panic on read errors; keep is_null unchanged and return only the
complete framed response.
TinDang97
added a commit
that referenced
this pull request
Aug 12, 2026
…#471) * docs(add): close watch-cas-transactions at PASS, record what the scope gate did not prove The code shipped in #470; this closes the task record. The first `gate PASS` was refused with `scope_violation` naming 7020 files, including `.github/workflows/*.yml` and `Cargo.lock` — paths this task never touched. Cause: the scope baseline is a whole-tree byte snapshot taken at the tests->build crossing, with no git awareness. Between that snapshot and the gate, five dependabot PRs merged into main (#420, #417, #346, #358, #359) and target/ was rebuilt, so every unrelated change read as "this task touched it". Rather than re-snapshot quietly, §6 now records: - the task's ACTUAL change set from `git show --name-only 151a185 1c12af0`, every path of which is inside the declared §5 Scope; - the two entries that needed naming instead of glossing — `conn/mod.rs` (one line, `pub mod watch;`, undeclared) and `.add/tooling/add.py` (25 lines of engine fix carried in from an earlier session, genuinely outside §5); - that the re-taken baseline is the post-merge tree, so the scope walk at this gate compares the tree against itself and proves nothing. The evidence that the build stayed in scope is the git file list, not the green gate. Also ticks the §6 verify checklist against evidence already gathered: the stubbed-counter A/B (reverting only next_birth_version turns 4 new tests red, so the green is earned), the 41->1 tokio failure delta traced to a diskfull $TMPDIR rather than to the change, and the 9/9 dispatched matrix including the Windows/macOS/console jobs no PR run executes. Adds .add/.gitignore, which `add.py init` scaffolds but this project never got. Without it the two 40MB scope-snapshot.json sidecars are commit candidates; they are regenerable working state, and the durable scope declaration is the state.json anchor. One 436K sidecar is already tracked from an older task and is left alone. Method delta for the next loop: as built, any task gated after unrelated merges land inherits an unfalsifiable scope violation. The anchor wants to be a git tree-ish, or the walk should skip gitignored paths and diff the merge-base. author: Tin Dang * feat(command): derive COMMAND/ROLE/RESET/HELLO from real server state Moon's client-identity surface answered from compile-time constants instead of from the server. The failures were invisible by eye because redis-cli renders `:0` and `*0` identically as "0", so every assertion here is on raw RESP bytes. COMMAND and COMMAND COUNT each returned the OTHER'S TYPE: bare COMMAND replied `:0` (an Integer where an Array belongs) and COMMAND COUNT replied `*0` (an Array where an Integer belongs). COMMAND INFO/DOCS/LIST/GETKEYS all replied an empty array. A driver that builds its command map at connect time does not read that as "unsupported" — it reads it as a protocol violation. All six now derive from the COMMAND_META phf registry (263 commands), so registering a command is what makes it introspectable and there is no second table to drift. ROLE and RESET were unknown commands. RESET was registered in the metadata table with full flags while dispatch rejected it — the same advertise-then- reject class as WATCH/UNWATCH before v0.8.6 — and a partial RESET existed only inside handler_sharded's subscribe-mode loop, so it worked if you happened to be subscribed on one runtime and nowhere else. RESET's "default state" is taken from restore_migrated_state(None, ..), the same function ConnectionState::new uses, so it cannot drift from what a fresh connection means by default. HELLO contradicted INFO replication on the same connection: hello_acl built `mode`/`standalone` and `role`/`master` as literals, so a replica announced itself a master. Both fields now read ReplicationState/ClusterState. Redis deliberately uses three vocabularies for this one fact (HELLO -> replica, INFO -> slave, ROLE -> slave); that was measured against a live replica pair on redis-server 8.6.1, and it corrected this task's own frozen contract, which had said "slave" everywhere. The test was made stronger rather than matched to the code: ci12 asserts all three surfaces agree, each in its own vocabulary, plus a negative assertion that HELLO no longer claims master. CLIENT INFO / CLIENT LIST reported the literal `laddr=127.0.0.1:0` from inside the format string, so every client on every listener showed port 0. It now carries the real local address. Wired on all three connection handlers plus the inline fast path. ROLE and RESET cannot ride the shared dispatch table — their answers live on ConnectionContext/ReplicationState, not in the Database that dispatch() receives — so each handler needs its own intercept, and handler_single had neither. That handler is reachable only via listener::run_with_shutdown (an in-process tokio API used by a few test suites; main.rs and embedded.rs both route through run_sharded), which is precisely why it had drifted, and why an A/B caught the gap: reverting the fix there left ci12 green. ci14 drives that handler in-process and goes red with `-ERR unknown command 'ROLE'` without it. Tests: 14 raw-RESP scenarios in tests/client_identity_introspection.rs, with parity legs on the monoio and sharded handlers and the inline path. Startup is serialised across the suite's threads — 13 servers initialising data dirs at once left one unable to answer PING inside 30s at ~1 run in 8; a longer timeout would have been slower and still flaky, and a shared OnceLock server would have leaked a live moon past exit because statics are never dropped. 32/32 green at --test-threads=13 after the fix. Found and filed, not folded in: response batches are serialised at flush time using the FINAL protocol version, so a protocol-changing command retro-encodes earlier replies in the same batch (`HELLO 3` alone -> `%7`; `HELLO 3` + `HELLO 2` in one write -> `*14`). Pre-existing — that reproducer touches none of this code — but RESET, which reverts the protocol by contract, adds a second trigger. Filed as ADD task batch-protocol-version-fidelity with the measurement table. MONITOR was split out to monitor-command-feed at freeze: it is a stream rather than a reply, the only item touching the per-command hot path, and the only one exposing other clients' credentials. Gates: cargo fmt --check, clippy (default + tokio/jemalloc, --all-targets), cargo test --release, and the full tokio CI-parity suite all green. author: Tin Dang * fix(command): queue ROLE inside MULTI, and stop the compat harness needing a live bug Two findings from the client-compat harness, which is the check that failed on PR #471 — both real, neither visible to the 14 raw-RESP tests in this task's own suite. 1. ROLE executed at QUEUE time inside MULTI. The first cut intercepted ROLE at the connection layer, ahead of the MULTI queueing step, because its answer lives on ConnectionContext.repl_state rather than in the Database that dispatch() receives. So `MULTI; ROLE; EXEC` replied the role array immediately and EXEC then returned `*0`. The damage is worse than a wrong reply: the command silently vanishes from the EXEC array, so every LATER result shifts down one index and a client reads another command's answer as this one's. Redis queues ROLE like any other command. Fixed by answering ROLE from the shared dispatch table instead, reading the process-global replication handle that INFO already uses and that every entry point registers (main.rs, listener.rs, embedded.rs). That is the only placement where a queued ROLE can work at all, since EXEC replays the queue through dispatch() — and it deleted all three per-handler intercepts, so ROLE now lives in exactly two places instead of five. Verified on the wire: `MULTI; ROLE; PING; EXEC` returns `*2` with the role array first and +PONG still last, and a live replica still reports `slave` through the global handle. ci15 pins the alignment; it fails under both prior states (`*0`, and `*1` with an unknown- command error). Nothing in this task's suite exercised a connection-layer command INSIDE a transaction, which is the coverage lesson: a new intercept has to be tested in MULTI as well as standalone, because its POSITION relative to queueing is the thing that can be wrong. 2. The harness's own divergence test needed a live Moon defect to pass. test_a_diverging_entry_exits_one_and_names_the_divergence borrowed a real bug as its fixture, so it FAILED whenever someone FIXED that bug — a test that punishes the fix. Three fixtures had already been burned this way (GET-inside-MULTI #457, SISMEMBER RESP3 #463, and COMMAND COUNT, retired by this very task, which is what turned the PR red). The durable fix was already designed and filed as #461, so it is implemented here rather than rotating to a fourth defect: a test-only `inject_moon_reply` hook fabricates the divergence, with a guard test asserting the shipped manifest never uses one. Proven load-bearing by disabling the hook, which turns the test red. Manifest: the identity_command_count and identity_role waivers are retired as fixed, and COMMAND INFO / COMMAND GETKEYS / RESET are added as live entries. COMMAND INFO carries a new, accurate waiver — its 10-field SHAPE now matches, but acl_categories is thin (@string where Redis says @READ @string @fast), key_specs is empty, and under RESP3 Redis types flags/acl_categories as Sets and key_specs entries as Maps where Moon emits Arrays. Owners recorded. Harness result against redis-server 8.6.1: PASS=181 FAIL=0 WAIVED=19 (was PASS=179 FAIL=8 before this commit). Gates: fmt, clippy (default + tokio/jemalloc, --all-targets), the identity suite (15 scenarios), and both compat suites (34 differ + 20 e2e) green. Note the unreachable-pattern warning caught during this work: (4, b'r') already existed in dispatch_inner for RPOP, so the first ROLE arm was dead code and EXEC still answered unknown command until it was merged into the existing arm. author: Tin Dang
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
WATCH/UNWATCHparsed, answered+OK, and recorded key versions — and then the two dispatchpaths clients actually reach (
handler_monoio,handler_sharded) never consulted the watch set atEXEC. Only the embeddedhandler_singlere-checked. A conflicting write from another clientcommitted anyway, so every check-and-set built on
WATCH— inventory decrement, balance transfer,leader election — silently degraded to last-writer-wins against a server reporting success.
Measured on
origin/main@8b1153b4before the fix (tmp/probe_watch.py, two connections):B's write was clobbered by a transaction that had declared a dependency on
k.Four defects, fixed together
A partial fix here is indistinguishable from none, so all four land in one change:
EXECon the monoio and sharded handlers.EXECnowaborts with a RESP null array on any version mismatch, and clears the watch set on both
outcomes by construction (
mem::take) rather than aclear()each exit path must remember.compared version belonged to an unrelated key or to nothing.
WATCHnow snapshots versionswhere the keys live via a new
ShardMessage::ReadVersions(grouped per owner: one hop pershard, not per key), the tokens travel with the body across the SPSC hop so the CAS check runs
where the body commits, and a watch set spanning shards is refused
CROSSSLOTlike across-shard
MULTIbody already was.WATCHinsideMULTIwas queued instead of refused;WATCHwith no arguments answered"unknown command" instead of an arity error.
every incarnation started at
INITIAL_VERSION:DEL k+SET khanded the watcher back theexact token it recorded and
EXECcommitted on a key destroyed and rebuilt underneath it, whereRedis aborts. Entries now draw from a per-database creation ticket. Restored keys draw tickets
too — versions are not persisted, so otherwise the first key created after a restart would
collide with the entire restored population.
Residual risk (stated, not buried)
The ticket shares the entry's 24-bit version field and 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..EXECwindow and hit the one watched key: ~1 in 16.7M, againstthe pre-fix certainty of 1.0. Only a wider
Entry(an incarnation field) retires it;WatchTokenstays a named struct so adding one later does not churn the call sites.
Rejected alternative: a per-db delete epoch. 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. This was the mechanism the contract
originally froze; the amendment and its reasoning are recorded in the task bundle (
FROZEN @ v2).Evidence
cargo fmt --check--all-targets-D warningsBench — moon-dev, fat-LTO both legs, 6 alternating rounds, 1M req
-c 50 -P 16after warm:Worst within-leg CV 7.7%. The untouched control moved as much as the leg under test, so both are
drift. A first
-P 1pass was discarded as uninformative: 13.9% noise floor with the control movingmore than SET.
Tests
tests/watch_cas_transactions.rs— 10 wire-level tests, raw-byte assertions (the abort signal ISthe type byte, so a reader rendering replies to text would hide the thing under test), two real
connections for conflicts, every case at
--shards 1and--shards 4. Five storage unit tests forthe creation ticket, each verified RED against a stubbed-constant counter before being accepted
green. Nine-combination coverage of the locality merge lattice. WATCH/CAS entries added to
scripts/test-consistency.shandscripts/test-commands.sh, driving the open transaction over/dev/tcpbecauseredis-clione-shot mode cannot hold a connection across an interleaved client.The two handlers' WATCH arms came out byte-identical at 57 lines, so both now call one
src/server/conn/watch.rs. Two copies is how these paths drifted in the first place.Out of scope, found while testing (filed, not fixed here)
tests/durability/*hardcodeCommand::new("./target/release/moon"), ignoringMOON_BIN— thesuite silently tests whatever binary is lying at that path.
durability::backup_restore::tests::backup_restore_parityassertsdump.rdb, which the snapshotwriter no longer produces (it writes
shard-N/shard-N.rrdshard). Last touched in24ee60eb(v0.1.3). Both are
#[ignore]d, so CI never runs them — which is how a durability gate rots intoproving nothing.
COMMAND INFO/COUNTreply*0; owned byinfo-observability, unchanged here.Known gap left open
With disk-offload enabled (opt-in), a watched key that spills cold and is promoted returns through
Database::setand draws a fresh ticket, so an eviction the client never asked for aborts itstransaction. Fails safe, but a CAS loop on a memory-pressured keyspace can livelock on eviction
rather than contention. Fixing it requires persisting versions. Recorded in the task's §7.
Summary by CodeRabbit
New Features
WATCH/UNWATCHtransactions.WATCH/UNWATCHusage and cross-shard watch conflicts.Bug Fixes
EXECandDISCARD.Tests