Skip to content

fix(transactions): WATCH/UNWATCH now actually guard EXEC on the production dispatch paths #470

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TinDang97 merged 2 commits into from
Aug 11, 2026
Merged

fix(transactions): WATCH/UNWATCH now actually guard EXEC on the production dispatch paths #470

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
151a185
fix(transactions): WATCH/UNWATCH now actually guard EXEC on the produ…
TinDang97 Aug 11, 2026
1c12af0
docs(add): amend watch-cas-transactions contract to FROZEN @ v2 — rec…
TinDang97 Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions .add/state.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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": {
Expand Down Expand Up @@ -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"
Comment on lines +290 to +296

@coderabbitai coderabbitai Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ 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]}')
PY

Repository: 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 --short

Repository: 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.

},
"flag_verified": true
},
"protocol-error-lifetime": {
"title": "Protocol errors reply and close cleanly, never stall or eat the valid prefix",
Expand Down Expand Up @@ -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",
Expand Down
482 changes: 403 additions & 79 deletions .add/tasks/watch-cas-transactions/TASK.md
View file
Open in desktop

Large diffs are not rendered by default.

28 changes: 28 additions & 0 deletions .add/tooling/add.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment on lines +1725 to +1738

@coderabbitai coderabbitai Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 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))
PY

Repository: 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))
PY

Repository: 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)),
        })
PY

Repository: 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.

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)
Expand Down Expand Up @@ -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"))
Comment on lines +1770 to +1780

@coderabbitai coderabbitai Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ 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")
PY

Repository: 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")
PY

Repository: 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")
PY

Repository: 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.

# drift: milestone + dependency references must resolve
ms = t.get("milestone")
if ms is not None:
Expand Down
46 changes: 46 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Comment on lines +105 to +111

@coderabbitai coderabbitai Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Document the disk-offload WATCH limitation.

The task records that promotion assigns a fresh creation ticket and can abort EXEC without a client write. This user-visible limitation is absent from the changelog entry. Add a short note for deployments with disk offload, or qualify the WATCH guarantee for that feature.

🤖 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 `@CHANGELOG.md` around lines 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.


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
Expand Down
51 changes: 51 additions & 0 deletions scripts/test-commands.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

# ===========================================================================
Expand Down
82 changes: 82 additions & 0 deletions scripts/test-consistency.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Comment on lines +597 to +613

@coderabbitai coderabbitai Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

No barrier between WATCH and the interleaved write in four CAS helpers. Each helper writes WATCH to the held /dev/tcp connection and then immediately launches a separate redis-cli write. Nothing proves the server processed the WATCH before that write lands. When the write wins the race, the snapshot records the post-write version, EXEC commits, and a correct server reports the wrong final value. Each helper already uses an ECHO barrier after EXEC; the same technique is needed after WATCH.

  • scripts/test-consistency.sh#L597-L613: append ECHO cas-armed to the first printf in watch_cas_outcome and drain fd 3 until that token returns, before the conditional redis-cli SET cas:k from-other.
  • scripts/test-consistency.sh#L624-L637: append ECHO cas-armed to the first printf in watch_cas_aba_outcome and drain until it returns, before the DEL aba:k / SET aba:k rebuilt pair.
  • scripts/test-consistency.sh#L643-L655: append ECHO cas-armed to the first printf in watch_unwatch_outcome and drain until it returns, before SET uw:k from-other; this also proves the UNWATCH was applied.
  • scripts/test-commands.sh#L732-L748: apply the identical change to watch_cas_outcome, and echo a distinct sentinel such as __WATCH_BARRIER_TIMEOUT__ when the drain times out, so a lost barrier reports itself instead of surfacing as a value mismatch.
📍 Affects 2 files
  • scripts/test-consistency.sh#L597-L613 (this comment)
  • scripts/test-consistency.sh#L624-L637
  • scripts/test-consistency.sh#L643-L655
  • scripts/test-commands.sh#L732-L748
🤖 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 `@scripts/test-consistency.sh` around lines 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.


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
# ===========================================================================
Expand Down
8 changes: 5 additions & 3 deletions src/server/conn/core.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -277,9 +277,11 @@ pub(crate) struct ConnectionState {
pub tracking_state: TrackingState,
pub tracking_rx: Option<channel::MpscReceiver<Frame>>,

// WATCH/EXEC optimistic locking (handler_single only)
#[allow(dead_code)] // Only used by handler_single (tokio feature)
pub watched_keys: HashMap<Bytes, u32>,
// 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<Bytes, crate::server::conn::shared::WatchToken>,

// Connection affinity (migration)
pub affinity_tracker: Option<AffinityTracker>,
Expand Down
1 change: 1 addition & 0 deletions src/server/conn/handler_monoio/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
32 changes: 29 additions & 3 deletions src/server/conn/handler_monoio/write.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Frame>,
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() {
Expand All @@ -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
Expand All @@ -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 =>
Expand All @@ -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,
)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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;
Expand Down
Loading
Loading