diff --git a/.add/state.json b/.add/state.json index 18701fdf..a2cc2fe9 100644 --- a/.add/state.json +++ b/.add/state.json @@ -1,7 +1,7 @@ { "project": "moon", "stage": "production", - "active_task": "pubsub-resp3-push", + "active_task": "monitor-command-feed", "active_milestone": "v0-9-client-compat", "tasks": { "hotpath-lock-quickwins": { @@ -418,14 +418,49 @@ }, "monitor-command-feed": { "title": "MONITOR command feed \u2014 stream executed commands, redacted, zero-cost when unattached", - "phase": "ground", - "gate": "none", + "phase": "done", + "gate": "PASS", "milestone": "v0-9-client-compat", "depends_on": [ "client-compat-harness" ], "created": "2026-08-11T16:13:33+00:00", - "updated": "2026-08-11T16:13:33+00:00" + "updated": "2026-08-14T06:47:34+00:00", + "flag_verified": true, + "tripwire": { + "contract_md5": "b0ac9bbaae7f8226060b7aefcde0d5fe", + "tests": {} + }, + "scope": { + "declared": [ + "src/monitor/", + "src/server/conn/", + "src/scripting/bridge.rs", + "src/command/metadata.rs", + "src/acl/rules.rs", + "src/lib.rs", + "tests/monitor_command_feed.rs", + "scripts/client-compat/manifest.yaml", + "CHANGELOG.md", + "tmp/" + ], + "snapshot_md5": "9a0aefc4568a02fb916c7b201b53bdd9" + }, + "heal": { + "attempts": 2, + "history": [ + { + "at": "2026-08-14T06:45:35+00:00", + "reason": "scope_violation: task 'monitor-command-feed' touched outside its declared \u00a75 Scope \u2014 CHANGELOG.md \u00b7 scripts/client-compat/manifest.yaml \u00b7 src/acl/rules.rs \u00b7 src/command/metadata.rs \u00b7 src/lib.rs (16 total)", + "source": "scope" + }, + { + "at": "2026-08-14T06:46:32+00:00", + "reason": "scope_violation: task 'monitor-command-feed' touched outside its declared \u00a75 Scope \u2014 CHANGELOG.md \u00b7 scripts/client-compat/manifest.yaml \u00b7 src/acl/rules.rs \u00b7 src/command/metadata.rs \u00b7 src/lib.rs (16 total)", + "source": "scope" + } + ] + } }, "batch-protocol-version-fidelity": { "title": "Response batch must be encoded in the protocol in effect when each reply was produced", @@ -540,7 +575,7 @@ } }, "created": "2026-06-11T03:18:21+00:00", - "updated": "2026-08-14T03:44:37+00:00", + "updated": "2026-08-14T06:47:34+00:00", "setup": { "locked": true, "locked_at": "2026-06-11T03:28:00+00:00", diff --git a/.add/tasks/monitor-command-feed/TASK.md b/.add/tasks/monitor-command-feed/TASK.md index 44e374b4..99ed3c94 100644 --- a/.add/tasks/monitor-command-feed/TASK.md +++ b/.add/tasks/monitor-command-feed/TASK.md @@ -2,7 +2,7 @@ slug: monitor-command-feed · created: 2026-08-11 · stage: production autonomy: auto -phase: ground +phase: done @@ -32,10 +32,122 @@ Why it was split rather than kept (the sizing argument, recorded so it is not re - It is the only one with a security surface: a monitor observes every other client's traffic, including credential-bearing arguments. Redis redacts AUTH; so must this. -Touches (files · symbols · signatures): - -Anchors the contract will cite: +Touches (files · symbols · signatures): +- `src/command/metadata.rs:21,26` — `CommandFlags::ADMIN` and `CommandFlags::SKIP_MONITOR` exist, + and the first instinct is to reuse them as the skip rule. **The audit proved that wrong on both + terms** (table below): Moon's ADMIN is CONTAINER-granular while Redis's is SUBCOMMAND-granular, + and Redis feeds the whole EVAL family despite flagging it `skip_monitor`. Reusing the flags + would hide 8+ commands Redis shows. A dedicated, measured predicate is needed instead. + `MONITOR` itself is absent from the 271-entry table and must be added (`arity 1`, cat `SRV`). +- `src/admin/metrics_setup.rs:1492` — `global_slowlog()`, the working precedent for a + process-global command sink: `maybe_record(elapsed_us, args, peer_addr, client_name)` already + receives ALL FOUR fields a monitor line needs. A `global_monitors()` is the same shape. +- `src/server/conn/handler_monoio/mod.rs:2607,2881` — where slowlog is recorded. NOT usable as-is + for the feed: the site sits behind `dispatch_start = sample_latency.then(...)`, i.e. latency is + SAMPLED, so a hook there would feed a sampled subset of traffic. The feed needs its own + unconditional site next to `record_command_cached`. +- `src/server/conn/handler_sharded/mod.rs:1893,2148` — the same pair on the tokio path. +- `src/server/conn/blocking.rs:1612` — `try_inline_dispatch`, the fast path a plain `GET`/`SET` + actually takes under monoio. A feed hook missing here is invisible to every test that uses + those two commands, which is most of them (see [[gotcha_moon_three_dispatch_paths]] and the + v0.8.6 inline-GET P0, which is this exact shape of miss). +- `src/command/mod.rs:66,1130` — `dispatch` / `dispatch_read`. Deliberately NOT the hook: neither + has the peer address, and the file's own comment records that per-command observability is + owned by the handler layer precisely because dispatch lacks the context. + +Anchors the contract cites: `CommandFlags::{ADMIN, SKIP_MONITOR}`, `COMMAND_META`, + `global_slowlog` (as the shape precedent), `try_inline_dispatch`, and the three handlers. + +### Measured against redis-server 8.6.1 (raw sockets, three probes, 2026-08-14) + +Raw sockets throughout: the feed is a stream of `+SimpleString` lines whose exact bytes ARE the +contract, and every client library reformats them before a test could see them. + +| # | question | redis-server 8.6.1 | Moon today | +|---|---|---|---| +| 1 | `MONITOR` reply | `+OK` | `-ERR unknown command 'MONITOR'` | +| 2 | feed line | `+. [ ] "SET" "k" "v"\r\n` | — | +| 3 | frame type under **RESP3** | still `+SimpleString` — **not** a Push | — | +| 4 | reads fed? | yes (`GET` appears) | — | +| 5 | db shown | the db AFTER `SELECT` (`SELECT 3` logs as `[3 …]`) | — | +| 6 | `AUTH pw` | `"AUTH" "(redacted)"` | — | +| 7 | `AUTH user pw` | `"AUTH" "(redacted)" "(redacted)"` | — | +| 8 | `HELLO 3 AUTH u p` | `"HELLO" "3" "AUTH" "(redacted)" "(redacted)"` | — | +| 9 | **admin commands** | **NOT fed** — `CONFIG SET`, `ACL SETUSER` absent | — | +| 10 | non-admin container subcommands | fed — `ACL WHOAMI`, `CLIENT GETNAME`, `INFO`, `DBSIZE` | — | +| 11 | unknown command | **not fed** | — | +| 12 | wrong-arity command | **not fed** | — | +| 13 | `MULTI` | fed at queue time | — | +| 14 | queued command | **not fed at queue time**; fed at EXEC | — | +| 15 | EXEC ordering | queued commands first, then `"EXEC"` (5µs apart) | — | +| 16 | Lua | `EVAL` fed, then each `redis.call` as `[ lua] "set" …` | — | +| 17 | two monitors | both receive identical bytes, same timestamp | — | +| 18 | monitor's own `PING` | fed (it sees itself) | — | +| 19 | `MONITOR` in the feed | never — because MONITOR is `admin`, i.e. rule 9 | — | +| 20 | keyspace cmd on a monitor conn | `-ERR Replica can't interact with the keyspace` | — | +| 21 | `INFO`/`CLIENT`/`SUBSCRIBE` on a monitor conn | allowed, work normally | — | +| 22 | second `MONITOR` on a monitor conn | **no reply at all** (silent) | — | +| 23 | `RESET` on a monitor conn | `+RESET`, feed stops, keyspace works again | — | +| 24 | `QUIT` | `+OK` | — | +| 25 | ACL, `+@all -@admin` | `-NOPERM User has no permissions to run the 'monitor' command` | — | +| 26 | `COMMAND INFO monitor` | arity 1 · flags admin/noscript/loading/stale · cats @admin @slow @dangerous | absent | + +**Argument escaping, measured byte by byte** (Redis `sdscatrepr` semantics): + +| byte | emitted | | byte | emitted | +|---|---|---|---|---| +| `"` | `\"` | | `\x07` | `\a` | +| `\` | `\\` | | `\x08` | `\b` | +| `\n` | `\n` | | `\x00` | `\x00` | +| `\r` | `\r` | | `\x7f` | `\x7f` | +| `\t` | `\t` | | `\x80`, `\xff` | `\x80`, `\xff` | +| space | literal, inside the quotes | | empty arg | `""` | + +UTF-8 is escaped PER BYTE, not per character: `héllo` → `"h\xc3\xa9llo"`. Every argument is +quoted, including the command name. + +### ADMIN-flag audit — the skip rule is NOT the existing flags (measured 2026-08-14) + +Requested as in-scope at the freeze. Two questions: does Moon UNDER-flag anything (a leak), and +does the existing flag reproduce Redis's behaviour (a divergence)? + +**Leak check: clean.** Zero commands that Redis marks admin are unflagged in Moon. Nothing +leaks its arguments to a monitor because Moon forgot to classify it. + +**Divergence check: the flags do not reproduce Redis.** Per-subcommand measurement: + +| fed by Redis (must NOT be skipped) | not fed by Redis (must be skipped) | +|---|---| +| `CLIENT GETNAME` · `CLIENT ID` | `CLIENT LIST` | +| `ACL WHOAMI` · `ACL CAT` | `ACL LIST` · `ACL SETUSER` | +| `CLUSTER INFO` · `CLUSTER MYID` | `CONFIG GET` · `CONFIG SET` | +| `INFO` · `DBSIZE` · `LASTSAVE` | `SLOWLOG GET/RESET/LEN` | +| `COMMAND COUNT` · `MEMORY USAGE/DOCTOR` | `LATENCY RESET/HISTORY` | +| `EVAL` · `SCRIPT LOAD` · `FUNCTION LIST` | `DEBUG JMAP` · `SHUTDOWN` | + +Two conclusions, both of which change the contract: +1. **Container-level ADMIN over-hides.** Moon flags `ACL`, `CLIENT`, `CLUSTER`, `CONFIG`, `INFO`, + `SLOWLOG`, `LATENCY`, `MODULE` admin as WHOLE COMMANDS. Skipping on that flag would hide + `INFO`, `CLIENT GETNAME`, `ACL WHOAMI`, `CLUSTER INFO`, `LASTSAVE` and more — commands Redis + shows. An operator would watch a feed that silently under-reports. +2. **`SKIP_MONITOR` does not mean what its name says.** Redis sets it on the entire EVAL family + (`EVAL`, `EVALSHA`, `FCALL`, and the `_RO` variants) and yet **feeds all of them** — measured + directly, including the follow-on `[db lua]` lines. Including `SKIP_MONITOR` in the rule would + suppress exactly the commands an operator most wants to see. + +Contracted response: a dedicated `monitor::is_hidden(cmd, first_arg) -> bool` stating the measured +rule at subcommand granularity, tested row-by-row against the table above — NOT a reuse of +`CommandFlags`. The 14 container-level ADMIN flags stay as they are; they are correct for ACL +purposes and only wrong as a monitor-visibility proxy. + +**Two more findings worth stating plainly, because both invert the obvious design:** +- **The feed is a SimpleString even in RESP3.** The instinct after `pubsub-resp3-push` is to make + it a Push frame; measurement says Redis does not, and a client reading the feed expects `+`. + Copying the pub/sub answer here would be a new divergence introduced by fixing the last one. +- **Admin commands are excluded, and that single rule explains rule 19.** MONITOR does not appear + in its own feed not because of a self-suppression special case, but because MONITOR is `admin`. + Implementing the general rule gets the special case for free; implementing the special case + leaves `CONFIG SET` leaking into the feed. ### Design constraints inherited from the split decision - The feed MUST be gated by a Relaxed atomic subscriber count checked BEFORE any formatting or @@ -56,24 +168,97 @@ v0-9-client-compat goal "a monitoring agent works unmodified" is NOT met, until ## 1 · SPECIFY — the rules ▸ docs/03-step-1-specify.md -Feature: -Framings weighed: (chosen) · · +Feature: `MONITOR` streams every executed command to attached admin clients, in Redis's exact +line format, with credentials redacted and admin traffic excluded — and costs one relaxed atomic +load per command when nobody is attached. + +Framings weighed: +- **A process-global monitor registry, fed from the handler layer (chosen).** One + `AtomicUsize` count + a `RwLock>`, checked count-first. It mirrors + `global_slowlog`, which already proves the shape works from all three handlers and already + receives all four fields a line needs. Monitors are server-global in Redis semantics (a monitor + sees every db and every connection), so a per-shard registry would have to fan out across + shards to reconstruct something that is global by definition. +- *Per-shard registries + cross-shard fan-out, like pub/sub.* Rejected: pub/sub is per-channel + and naturally shards; a monitor subscribes to EVERYTHING, so every publish would cross every + shard. It buys nothing and adds an SPSC hop per command per shard. +- *Hook inside `dispatch`.* Rejected on evidence: `dispatch` has no peer address and no + connection identity, and `src/command/mod.rs:71` records that per-command observability lives + in the handler layer for exactly that reason. +- *Reuse the slowlog hook site.* Rejected on measurement: that site is behind + `sample_latency.then(...)`, so it would feed a SAMPLED subset — a monitor that silently drops + most traffic is worse than no monitor, because the operator cannot tell. + Must: - - + - M1 `MONITOR` from an admin-authorised connection replies `+OK` and attaches the connection. + - M2 Every subsequently EXECUTED command is emitted to every attached monitor as one + SimpleString: `+. [ ] "CMD" "arg" …\r\n`, args quoted and escaped + with `sdscatrepr` semantics (per-byte, table in §0), command name quoted too. + - M3 The line is a SimpleString under RESP2 AND RESP3 — never a Push frame. + - M4 The db shown is the connection's db at execution time; the addr is its peer address. + - M5 Reads are fed, not only writes. + - M6 Administrative commands are NOT fed, at SUBCOMMAND granularity, per the measured table + in §0 — `CONFIG *`, `SLOWLOG *`, `LATENCY *`, `DEBUG *`, `SHUTDOWN`, `ACL LIST/SETUSER`, + `CLIENT LIST`, and `MONITOR` itself are hidden, while `INFO`, `DBSIZE`, `LASTSAVE`, + `CLIENT GETNAME/ID`, `ACL WHOAMI/CAT`, `CLUSTER INFO/MYID`, `COMMAND COUNT`, + `MEMORY USAGE/DOCTOR` and the whole EVAL family ARE fed. Explicitly NOT expressed as + `CommandFlags::ADMIN | SKIP_MONITOR` — the audit proved both terms wrong. + - M6b The ADMIN audit is re-runnable: a test pins Moon's hidden-set against the measured + oracle table, so a future flag change cannot silently start leaking or hiding. + - M7 A command rejected before execution — unknown name, or arity violation — is NOT fed. + - M8 `AUTH`'s arguments are redacted as `"(redacted)"`, one per argument, AT FORMATTING TIME. + `HELLO … AUTH

` redacts the two arguments after the `AUTH` keyword and nothing else. + - M9 `MULTI` is fed when it is issued; queued commands are NOT fed at queue time; at `EXEC` the + queued commands are fed in order and then `EXEC` itself. + - M10 Every attached monitor receives every line; two monitors see identical bytes. + - M11 With no monitor attached, the per-command cost is one `Relaxed` atomic load — no + formatting, no allocation, no lock. + - M12 `RESET` on a monitor connection detaches it: the feed stops and keyspace access returns. + - M13 The feed hook is present on ALL production paths — `handler_monoio`, `handler_sharded`, + and `try_inline_dispatch` — so a plain `GET`/`SET` is fed like anything else. + - M14 `MONITOR` is registered in `COMMAND_META` with arity 1 and the admin category, so + `COMMAND INFO monitor` answers and ACL can reason about it. Reject: - - -> "" + - R1 `MONITOR` from a user without the admin category -> + "NOPERM User has no permissions to run the 'monitor' command" (and NOT attached) + - R2 `MONITOR` with any argument -> "ERR wrong number of arguments for 'monitor' command" + - R3 A keyspace command on an attached monitor connection -> + "ERR Replica can't interact with the keyspace" (and the feed keeps flowing) + - R4 `MONITOR` again on an already-attached connection -> NO REPLY AT ALL, and the connection + stays attached exactly once (measured: Redis is silent here, it does not error) After: - - + - An unmodified `redis-cli monitor` prints Moon's traffic in the same format it prints Redis's. + - The `v0-9-client-compat` goal "a monitoring agent works unmodified" stops being blocked. + - `used_memory`/throughput are unchanged when no monitor is attached, provably (M11). Assumptions — lowest-confidence first: - ⚠ — lowest confidence because ; if wrong: - - [ ] — confirm or deny; never carry an open one forward + ⚠ **A monitor that cannot keep up must be dropped, not allowed to stall the shard.** Lowest + confidence because the cost is invisible until a slow monitor exists: the feed is written + from the command hot path on every shard, and a bounded queue that blocks would convert one + slow TCP reader into a server-wide stall. Contracting a BOUNDED queue with drop-on-full (the + same trade pub/sub already makes for slow subscribers) is the safe choice; if wrong, a + monitor silently misses lines under load and an operator may not notice. Mitigation + contracted: drop the whole MONITOR connection rather than silently skipping lines, so the + loss is loud. Cost if wrong: a monitor detaches under burst load and must be re-attached. + ⚠ **Feeding AFTER execution rather than before is observationally equivalent.** Redis feeds at + the start of `call()`; the only site in Moon that sees every command with full context is + around dispatch. If a command BLOCKS (BLPOP), Redis shows it immediately and a + feed-after-execution would show it only on unblock — a real, observable divergence for + blocking commands. Cost if wrong: blocking commands appear late in the feed. Contracted + response: feed BEFORE dispatch at each hook site, so timestamp and ordering match. + - [ ] The 271-entry `COMMAND_META` table's ADMIN flags agree with Redis's on the commands an + operator would care about leaking. Confirm by diffing Moon's ADMIN set against + `COMMAND INFO` from redis-server for the shared command set — an under-flagged command + would LEAK to the feed, which is a security-relevant miss, not a cosmetic one. + - [ ] Lua-issued commands (`[ lua]`) are in scope. Moon's EVAL path executes through the + same dispatch, so they may fall out for free; if they do not, the divergence is recorded + rather than papered over. @@ -85,11 +270,121 @@ Assumptions — lowest-confidence first: ```gherkin -Scenario: - Given - When - Then - And # required for every rejection +Scenario: a monitor sees an ordinary command # M1 M2 M4 M5 + Given an admin connection that issued MONITOR and got +OK + When another connection on db 0 runs SET k v + Then the monitor reads one line matching + +.<6 digits> [0 :] "SET" "k" "v" + And the SET connection's own reply is +OK, unchanged + +Scenario: the feed is a SimpleString under RESP3 # M3 + Given a connection that sent HELLO 3 and then MONITOR + When another connection runs SET k v + Then the feed line starts with '+' and not with '>' + And the HELLO reply itself was still a RESP3 Map + +Scenario: reads are fed too # M5 + Given an attached monitor + When another connection runs GET k + Then the feed contains "GET" "k" + +Scenario: the db shown follows SELECT # M4 + Given an attached monitor + When another connection runs SELECT 3 then GET k + Then both lines show [3 ] + +Scenario: admin commands never reach the feed # M6 + Given an attached monitor + When another connection runs CONFIG SET maxmemory 0 + Then no line for CONFIG appears + And a following DBSIZE on the same connection DOES appear, proving the feed is live + +Scenario: MONITOR is absent from its own feed # M6 M19 + Given an attached monitor + When a second connection issues MONITOR + Then no "MONITOR" line appears on either feed + And both monitors then receive a subsequent SET + +Scenario: rejected commands are not fed # M7 + Given an attached monitor + When another connection sends NOSUCHCMD x, then GET with no key + Then neither produces a feed line + And a following valid PING does produce one + +Scenario: AUTH arguments are redacted # M8 + Given an attached monitor + When another connection runs AUTH hunter2 and AUTH user pass + Then the feed shows "AUTH" "(redacted)" and "AUTH" "(redacted)" "(redacted)" + And the literal secrets appear nowhere in the bytes read + +Scenario: HELLO AUTH redacts only the credentials # M8 + Given an attached monitor + When another connection runs HELLO 3 AUTH default sekrit + Then the feed shows "HELLO" "3" "AUTH" "(redacted)" "(redacted)" + And "sekrit" appears nowhere in the bytes read + +Scenario: transaction timing # M9 + Given an attached monitor + When another connection runs MULTI, then SET q 1, then EXEC + Then MULTI appears when issued + And SET does NOT appear at queue time + And after EXEC the feed shows "SET" "q" "1" followed by "EXEC" + +Scenario: every monitor gets every line # M10 + Given two attached monitors + When another connection runs SET dual 1 + Then both read a line naming SET dual 1 + +Scenario: argument escaping is byte-exact # M2 + Given an attached monitor + When another connection sets a value containing " \ newline tab NUL 0xff and a UTF-8 char + Then the feed escapes them as \" \\ \n \t \x00 \xff and per-byte \xc3\xa9 + And an empty argument appears as "" + +Scenario: the fast path is fed like any other # M13 + Given an attached monitor on a server started with --shards 4 + When another connection runs a plain GET and a plain SET + Then both appear in the feed + # plain GET/SET take try_inline_dispatch; a hook missing there is invisible + # to every test that uses those two commands, which is most of them + +Scenario: a non-admin user cannot attach # R1 + Given a user created with +@all -@admin, authenticated + When it issues MONITOR + Then it is answered -NOPERM … 'monitor' … + And it receives no feed line when another connection runs SET k v + +Scenario: MONITOR takes no arguments # R2 + Given an admin connection + When it issues MONITOR extra + Then it is answered -ERR wrong number of arguments for 'monitor' command + And the connection is not attached + +Scenario: a monitor may not touch the keyspace # R3 + Given an attached monitor + When it issues SET x 1 + Then it is answered -ERR Replica can't interact with the keyspace + And the feed continues to deliver another connection's commands + +Scenario: MONITOR twice is silent # R4 + Given an attached monitor + When it issues MONITOR again + Then it receives no reply at all + And a single subsequent SET produces exactly ONE line, not two + +Scenario: RESET detaches # M12 + Given an attached monitor + When it issues RESET + Then it is answered +RESET + And a subsequent SET on another connection produces no line + And the former monitor can run GET again + +Scenario: zero cost when unattached # M11 + Given a server with no monitor attached + When the command path executes + Then the only monitor-related work is one relaxed atomic load + # observable as: the feed counter is checked before any formatting, proven by + # a unit test asserting the formatter is never called at count 0 ``` @@ -100,14 +395,96 @@ Scenario: ## 3 · CONTRACT — freeze the shape ▸ docs/05-step-3-contract.md +The contract is WIRE BYTES plus one hot-path cost bound. Every clause below was measured +against redis-server 8.6.1, not read from documentation. + ``` - body: { } - 200 -> { } - 4xx -> { error: "" | "" } -Schema: +Command + MONITOR arity 1, flags ADMIN|NOSCRIPT|LOADING|STALE, cats @admin @slow @dangerous + admin-authorised, not yet attached -> +OK, connection attached + already attached -> NO REPLY (silent; measured, not an error) + not admin-authorised -> -NOPERM User has no permissions to run the 'monitor' command + any argument -> -ERR wrong number of arguments for 'monitor' command + +Feed line (one per executed command, per attached monitor) + +. [ ] "" "" …\r\n + frame type : SimpleString under BOTH RESP2 and RESP3 — never Push + : zero-padded to exactly 6 digits + : peer address, "ip:port"; the literal `lua` for script-issued commands + : the issuing connection's db AT EXECUTION TIME + quoting : EVERY token quoted, command name included + escaping : per BYTE — " -> \" \ -> \\ \n \r \t -> \n \r \t + 0x07 -> \a 0x08 -> \b other non-printable/high -> \xHH (lowercase) + space kept literal; empty argument renders as "" + +Emission rules + fed : every command that reaches execution, reads included + NOT fed : monitor::is_hidden(cmd, first_arg) — subcommand-granular, per the §0 audit table. + hidden : CONFIG * · SLOWLOG * · LATENCY * · DEBUG * · SHUTDOWN · MONITOR · + ACL LIST · ACL SETUSER · CLIENT LIST + fed : INFO · DBSIZE · LASTSAVE · COMMAND * · MEMORY USAGE|DOCTOR · + CLIENT GETNAME|ID · ACL WHOAMI|CAT · CLUSTER INFO|MYID · + EVAL|EVALSHA|FCALL|*_RO (yes — despite Redis's own skip_monitor flag) + NOT CommandFlags::ADMIN and NOT ::SKIP_MONITOR — measured wrong, see §0 + NOT fed : unknown command; arity-rejected command (rejected before execution) + MULTI : fed when issued + queued cmd : NOT fed at queue time; fed at EXEC, in order, then "EXEC" + redaction : AUTH -> every argument becomes "(redacted)" + HELLO … AUTH -> the two arguments after the AUTH keyword become "(redacted)" + applied AT FORMATTING TIME — a secret is never written into a buffer and filtered + +Monitor connection state + keyspace command -> -ERR Replica can't interact with the keyspace (feed keeps flowing) + PING / INFO / CLIENT / SUBSCRIBE -> served normally, and PING appears in the feed + RESET -> +RESET, detached, keyspace access restored + QUIT -> +OK, closed + disconnect -> detached; the registry must not retain the sink + +Registry (src/monitor/, new module) + static MONITOR_COUNT: AtomicUsize // Relaxed + static MONITORS: RwLock> // parking_lot; write only on attach/detach + #[inline] fn feed(db, addr, cmd, args): + if MONITOR_COUNT.load(Relaxed) == 0 { return } // the ENTIRE unattached cost + … format once, fan out to every sink … + backpressure: bounded queue per monitor; on full, DROP THE MONITOR CONNECTION + (loud) rather than skipping lines (silent) + +Hook sites — all three, or the feature is CI-invisible + handler_monoio : before dispatch, unconditional (NOT the sampled slowlog site) + handler_sharded : same + try_inline_dispatch (blocking.rs) : same — this is the path a plain GET/SET takes ``` -Status: DRAFT +Not in scope, stated so the boundary is explicit: `MONITOR` output for commands executed by +replication apply or AOF load (Redis does not feed those either), and any new `CLIENT NO-EVICT`/ +`CLIENT KILL TYPE monitor` surface beyond what already exists. + +Status: FROZEN @ v1 — approved by Tin Dang, 2026-08-14. +Approved with both flags resolved at the freeze, and the audit run BEFORE freezing rather than +after — which is what caught the skip rule being wrong: +- Flag 1 resolved: a slow monitor has its CONNECTION DROPPED, loudly. Not silent line-dropping + (an operator cannot distinguish a quiet server from a lossy feed) and never blocking (one slow + TCP reader must not stall every shard). Recoverable by re-attaching. +- Flag 2 resolved as contracted: feed BEFORE dispatch at each hook site, so blocking commands + appear when issued rather than when they unblock. +- The in-scope ADMIN audit changed the contract materially: the skip rule is a dedicated + subcommand-granular predicate, NOT `CommandFlags::ADMIN | SKIP_MONITOR`. Recorded above. + +Least-sure flag surfaced at freeze: +1. `[contract]` **Slow-monitor backpressure: drop the CONNECTION, not the lines.** This is the + one clause with a server-wide blast radius. The feed is written from the command hot path on + every shard, so a bounded queue that BLOCKS turns one slow TCP reader into a server-wide + stall — the worst outcome. Dropping lines silently is the other tempting answer and is worse + than it looks: an operator reading a monitor feed to diagnose an incident cannot tell a quiet + server from a lossy feed. Contracted answer is to kill the monitor connection, which is loud + and unambiguous. If this is wrong, a monitor detaches under burst load and must be + re-attached — recoverable, unlike either alternative. +2. `[spec]` **Feed BEFORE dispatch, not after.** The natural implementation feeds where the + slowlog records, i.e. after execution — but a blocking command (BLPOP) would then appear only + when it unblocks, which is an observable divergence from Redis for exactly the commands an + operator is most likely to be watching for. Feeding before dispatch costs a second borrow of + the args at each hook site; that is the price of matching. If wrong, the cost is a small + refactor of three call sites, not a redesign. + + +Strategy (ordered batches): +1. `src/monitor/` — registry, hidden-set predicate, `sdscatrepr` formatter, redaction. Unit-tested + in isolation before any handler knows it exists. +2. `src/server/conn/monitor_mode.rs` — the once-stated "a monitor may not touch the keyspace" rule. +3. Registration: `COMMAND_META` row + `@admin` category membership in `acl/rules.rs`. +4. Handler wiring, both runtimes: attach verb, mode gate, feed hook, delivery arm, teardown. +5. The inline fast path and the Lua bridge — the two paths a handler hook structurally cannot reach. + +Safety rule (feature-specific): the attach verb sits BELOW the ACL gate. An intercept placed above +it exempts itself from ACL — the exact shape of the v0.8.6 inline-GET P0 — and `MONITOR` is the last +command that should be reachable without permission, since it reads every other user's arguments. + Code lives in: `./src/` Constraints: do NOT change any test or the contract; allow-list packages only; ask if unclear. +### Contract deviations — recorded, not silently absorbed +One clause was satisfied by a different mechanism than the frozen text names. Stated here rather +than left for a reader to discover by diffing: + +- **`try_inline_dispatch` hook site.** The contract names three hook sites; the third was built as a + STAND-DOWN instead of a hook — `any_attached()` disables the inline fast path while any monitor is + attached, so the command falls through to the fully-hooked path. The inline path answers straight + from the read buffer and never sees `peer_addr`, so hooking it meant threading an address through + the hottest function in the codebase for a diagnostic feature. The observable contract is + unchanged and `mon13` (the test written for that clause) passes on `--shards 4` for both plain GET + and plain SET. The mechanism is stronger than the contracted one: a future refactor of the inline + path cannot silently drop the feed, because there is no hook there to forget. The cost is + throughput while a monitor is attached — i.e. only when an operator has already accepted + diagnostic overhead. Fast-path retention when unattached is confirmed by + `moon_dispatch_path_total{path="local_inline"}` = 200 over 200 GETs, not inferred from latency. + @@ -188,14 +774,74 @@ Reviewed by: · date: ## 7 · OBSERVE — feed the next loop ▸ docs/09-the-loop.md -Watch (reuse scenarios as monitors): +Watch (reuse scenarios as monitors): `moon_dispatch_path_total{path="local_inline"}` staying at its +pre-MONITOR share while no monitor is attached (the stand-down must not become permanent — a leaked +sink would hold the fast path down for the life of the process); monitor connections closed for +backpressure, which should be rare and is the signal that the bounded queue is undersized for a real +workload rather than for a test burst. ### Spec delta Forward changes for the next loop — each re-enters at Specify as the next task. One line each, tagged `[SPEC · open|seeded|dropped]`, with evidence (e.g. `[SPEC · open] rate-limit the retry path (evidence: prod herd spikes)`). See the `add` skill's `deltas.md`. +- [SPEC · seeded] Moon's ACL registry has no `@admin` category bit, so `COMMAND INFO` reports a + thinner `acl_categories` array than redis-server for every admin command, not just MONITOR + (evidence: compat differ, `monitor_command_is_registered` and the pre-existing + `identity_command_info_known_and_unknown` waiver — the same gap, found twice by two tasks). + Owned by `sdk-wire-form-fixes`. +- [SPEC · open] `CLIENT LIST` does not mark monitor connections with the `flags=O` that redis-server + uses, and `CLIENT KILL TYPE monitor` is not a supported selector (evidence: explicitly placed out + of scope at the §3 freeze). An operator who cannot see or kill a monitor from `CLIENT LIST` has to + find the connection by hand. +- [SPEC · open] The per-monitor queue depth (4096) is a fixed constant with no operator control. The + contracted policy is to drop the connection when it fills, so the constant decides how large a + burst a healthy monitor survives (evidence: `mon21` needed a 20,000-command burst to trip it). + Worth a `--monitor-queue-depth` knob if a real deployment sheds monitors. + ### Competency deltas What did this loop teach the foundation? One line each, tagged by competency (`DDD · SDD · UDD · TDD · ADD`), status `open`, with evidence. See the `add` skill's `deltas.md`. - + +- [SDD · open] Running the audit that a contract clause depends on BEFORE the freeze, rather than + during build, changed the contract materially. The draft §3 said "skip on `CommandFlags::ADMIN | + SKIP_MONITOR`"; measuring redis-server proved BOTH terms wrong (Moon's ADMIN is container-granular + and would have hidden six commands Redis shows; Redis feeds the whole EVAL family despite its own + `skip_monitor` flag). Had the audit run after the freeze it would have been a change request + against a frozen contract instead of a better contract. Evidence: §3's recorded amendment and + `mon20`, which now pins the table row by row. +- [TDD · open] A "hidden" assertion is vacuous unless the same test proves the feed was LIVE. Every + negative test here asserts a FOLLOWING command IS fed, because a broken feed and a correctly + hidden command are indistinguishable otherwise. Evidence: `mon5`, `mon6`, `mon7` each carry the + positive half deliberately. +- [ADD · open] The §6 WIRING check must walk TEARDOWN paths, not only dispatch paths. On the + previous task it walked every dispatch path, passed, and all three defects review found were on + teardown (RESET, disconnect). Applying the corrected version here found the RESET and disconnect + detach requirements before review rather than after. Evidence: the §6 WIRING entry above and the + three `pubsub-resp3-push` CodeRabbit findings that motivated it. +- [SDD · open] Deriving a rule from a flag Moon already has, instead of measuring + it, produced the SAME defect twice in one task: once for the MONITOR hidden-set + (`CommandFlags::ADMIN`, caught before freeze by the in-scope audit) and once for + the keyspace-refusal rule (`first_key`, then `WRITE|READONLY` — caught only by + external review). Moon's flags are named after Redis's and do not mean the same + thing. Evidence: `PING`/`ECHO`/`TIME`/`INFO` are `READONLY` in Moon and are not + in Redis; `DBSIZE`/`KEYS`/`FLUSHALL` have `first_key == 0` and are refused. + The rule: when a behaviour must match Redis, the predicate is measured and + table-pinned, never derived from a same-named local flag. +- [TDD · open] A test written against a fixture that does not reach the code path + it names is worse than no test, because it reports the path as covered. + `mon8`/`mon9` claimed to cover AUTH redaction and ran against a server with no + password, so they never touched the pre-auth gate where the only + credential-bearing AUTH actually goes. Evidence: `mon24` fails on the same + binary those two pass on. Ask what fixture makes the path REACHABLE, not just + what call makes the assertion true. +- [TDD · open] A negative-permission test needs its positive twin. `mon14` + (`-@admin` refuses MONITOR) passes identically whether the grant works or + MONITOR is ungrantable by anyone; only `mon26` (`+@all` grants it) tells the + two apart. Evidence: external review flagged the `@all` expansion as a defect, + and `mon26` is what proved it was not one. +- [ADD · open] A frozen contract clause naming an IMPLEMENTATION SHAPE (three hook sites) rather + than an observable can be satisfied better by a different mechanism. The right move was to build + the better mechanism and record the deviation in §5, not to edit §3 and not to build the + contracted shape against judgment. Evidence: the inline-path stand-down, with `mon13` — the test + written for that clause — passing unchanged. diff --git a/CHANGELOG.md b/CHANGELOG.md index 001bccc1..4ca4f778 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **`MONITOR` — the command feed.** `redis-cli monitor` now works against Moon, in Redis's exact + line format: `+. [ ] "CMD" "arg" …`, arguments quoted and escaped per + byte (`sdscatrepr` semantics — `"` `\`, `\n` `\r` `\t`, `\a` `\b`, and `\xHH` for everything + outside printable ASCII, so UTF-8 escapes per byte rather than per character). The line is a + SimpleString under **both** RESP2 and RESP3 — measured; Redis does not use a Push frame here, + and the reflex to make it one after the RESP3 pub/sub work would have been a new divergence. + `AUTH`'s arguments and the credentials in `HELLO … AUTH` render as `(redacted)`, decided before + any argument is written rather than filtered afterwards. + + Two behaviours are worth knowing because they are not the obvious implementation. First, + administrative commands are hidden at **subcommand** granularity: `CONFIG *`, `SLOWLOG *`, + `LATENCY *`, `ACL LIST/SETUSER` and `CLIENT LIST` never reach a monitor, while `INFO`, `DBSIZE`, + `LASTSAVE`, `CLIENT GETNAME/ID`, `ACL WHOAMI/CAT` and `CLUSTER INFO/MYID` do. Moon's own + `CommandFlags::ADMIN` is container-granular and could not express that split — using it would + have hidden six commands Redis shows — and Redis feeds the entire `EVAL` family despite flagging + it `skip_monitor`, so neither flag is consulted; the rule is stated explicitly and pinned + row-by-row against the measured oracle. `MONITOR` is absent from its own feed as a consequence of + that general rule rather than a self-suppression special case. Second, a monitor that stops + reading has its **connection dropped**: silently skipping lines would leave an operator unable to + tell a quiet server from a lossy feed, and blocking would let one slow TCP reader stall every + shard. + + A monitor connection may not touch the keyspace, matching Redis + (`-ERR Replica can't interact with the keyspace`). The refusal set is measured, not derived from + a flag: `DBSIZE`, `KEYS`, `SCAN`, `RANDOMKEY`, `FLUSHALL`, `FLUSHDB`, `SWAPDB`, `EVAL`, + `PUBLISH` and `MEMORY USAGE` name no key yet are all refused, while `PING`, `INFO`, `TIME`, + `ECHO`, `COMMAND`, `LASTSAVE`, `WAIT`, `SELECT`, `CLIENT`, `ACL`, `SUBSCRIBE` and `RESET` are + served. Neither `first_key` nor Moon's `WRITE`/`READONLY` flags reproduce that split — Moon + flags `PING` and `INFO` readonly and Redis does not — so the rule is stated explicitly and + pinned row by row against the oracle. + + Commands issued by a Lua script are fed too, carrying the literal `lua` in place of a peer + address and appearing in execution order after the `EVAL` line — matching Redis. A script command + never passes a connection handler, so it needs its own hook; without it an operator watching a + script-driven workload would see every `EVAL` and none of its effects. + + `MONITOR` requires the `admin` ACL category, and costs one relaxed atomic load per command when + nobody is attached — every other step lives behind that load. While a monitor IS attached the + inline fast path stands down, because it answers straight from the read buffer and never sees a + peer address; the feed is therefore correct by construction on that path rather than by a hook + that must be kept in sync. Fast-path retention when unattached is confirmed by + `moon_dispatch_path_total{path="local_inline"}`, not inferred from latency. - **Sharded pub/sub: `SSUBSCRIBE`, `SUNSUBSCRIBE`, `SPUBLISH`, and `PUBSUB SHARDCHANNELS` / `SHARDNUMSUB`.** Deliveries carry the `smessage` event name. The sharded namespace is a genuinely separate map from the plain one in both the per-shard registry and the diff --git a/scripts/client-compat/manifest.yaml b/scripts/client-compat/manifest.yaml index 78d921cd..146e47b7 100644 --- a/scripts/client-compat/manifest.yaml +++ b/scripts/client-compat/manifest.yaml @@ -317,3 +317,30 @@ entries: command: "SPUBLISH compat_sch hello" policy: exact contexts: [standalone] + + # MONITOR: the attach reply and the registry entry. The FEED itself is a + # stream, not a reply, so it cannot be diffed by a request/response harness — + # tests/monitor_command_feed.rs owns the line-format parity, byte for byte. + - name: monitor_command_is_registered + command: "COMMAND INFO monitor" + policy: exact + contexts: [standalone] + expect_diff: + reason: >- + The 10-field spec shape, arity 1 and the four flags (admin, noscript, + loading, stale) all match. One inner divergence remains: Redis reports + acl_categories @admin @slow @dangerous, Moon reports @dangerous @slow — + Moon's registry has no @admin CATEGORY at all (its AclCategories table + renders SERVER as @server and has no admin bit), so the name cannot be + emitted without adding a category to the registry-wide table. This is the + same acl_categories thinness already waived for + identity_command_info_known_and_unknown, not a MONITOR-specific gap, and + it is not a permission hole: `+@all -@admin` DOES refuse MONITOR, because + Moon resolves @admin from a command-name list in src/acl/rules.rs rather + than from the category bitmask (pinned by mon14_non_admin_cannot_attach). + Owner: sdk-wire-form-fixes, which owns acl_categories completeness. + + - name: monitor_rejects_arguments + command: "MONITOR extra" + policy: exact + contexts: [standalone] diff --git a/src/acl/rules.rs b/src/acl/rules.rs index 712a2b99..11e1827b 100644 --- a/src/acl/rules.rs +++ b/src/acl/rules.rs @@ -507,6 +507,10 @@ pub fn get_category_commands(category: &str) -> &'static [&'static str] { "config", "info", "debug", + // MONITOR streams every other client's traffic, arguments included. + // Without membership here, `+@all -@admin` would still be able to + // read every other user's commands. + "monitor", "bgsave", "bgrewriteaof", "save", diff --git a/src/command/metadata.rs b/src/command/metadata.rs index 09355a30..2c7df18f 100644 --- a/src/command/metadata.rs +++ b/src/command/metadata.rs @@ -366,6 +366,9 @@ pub static COMMAND_META: phf::Map<&'static str, CommandMeta> = phf_map! { "CONFIG" => CommandMeta { name: "CONFIG", arity: -2, flags: A, first_key: 0, last_key: 0, step: 0, acl_categories: SRV }, "ACL" => CommandMeta { name: "ACL", arity: -2, flags: A, first_key: 0, last_key: 0, step: 0, acl_categories: SRV }, "SLOWLOG" => CommandMeta { name: "SLOWLOG", arity: -2, flags: A, first_key: 0, last_key: 0, step: 0, acl_categories: SRV }, + // MONITOR streams every other client's traffic, so it is admin-gated: without + // the category any user could read every other user's arguments. + "MONITOR" => CommandMeta { name: "MONITOR", arity: 1, flags: CommandFlags(CommandFlags::ADMIN.0 | CommandFlags::NOSCRIPT.0 | CommandFlags::LOADING.0 | CommandFlags::STALE.0), first_key: 0, last_key: 0, step: 0, acl_categories: AclCategories(AclCategories::DANGEROUS.0 | AclCategories::SLOW.0) }, // Moon extension: HOTKEYS [COUNT n] — top sampled keys from the per-shard sketch. "HOTKEYS" => CommandMeta { name: "HOTKEYS", arity: -1, flags: RA, first_key: 0, last_key: 0, step: 0, acl_categories: SRV }, "DEBUG" => CommandMeta { name: "DEBUG", arity: -2, flags: A, first_key: 0, last_key: 0, step: 0, acl_categories: DNG }, diff --git a/src/lib.rs b/src/lib.rs index 272248a1..564358a4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -53,6 +53,7 @@ pub mod error; pub mod graph; pub mod io; pub mod memory_ctl; +pub mod monitor; pub mod mq; pub mod notify; pub mod notify_fanout; diff --git a/src/monitor/mod.rs b/src/monitor/mod.rs new file mode 100644 index 00000000..76e9172a --- /dev/null +++ b/src/monitor/mod.rs @@ -0,0 +1,538 @@ +//! `MONITOR` — stream every executed command to attached admin clients. +//! +//! # Cost when nobody is attached +//! +//! The feed is written from the command hot path on every shard, so the +//! unattached case must be free. [`feed`] loads one `Relaxed` atomic and +//! returns; no formatting, no allocation, no lock is reached until a monitor +//! actually exists. `monitor_registry_zero_cost_when_unattached` pins that as a +//! property of the code rather than a claim in this comment. +//! +//! # Why the hidden-set is spelled out here rather than read from `CommandFlags` +//! +//! The obvious implementation is `flags.contains(ADMIN | SKIP_MONITOR)`. Both +//! terms were measured wrong against redis-server 8.6.1 (2026-08-14): +//! +//! * Moon's `ADMIN` is **container**-granular — `ACL`, `CLIENT`, `CLUSTER`, +//! `INFO` are flagged as whole commands. Redis's is **subcommand**-granular +//! and feeds `INFO`, `CLIENT GETNAME`, `CLIENT ID`, `ACL WHOAMI`, +//! `ACL CAT`, `CLUSTER INFO`, `CLUSTER MYID`. Skipping on the flag would +//! silently under-report the feed, which is worse than useless to the +//! operator reading it during an incident. +//! * `SKIP_MONITOR` is set by Redis on the entire `EVAL` family, and Redis +//! **feeds all of them anyway** — including the follow-on `[db lua]` lines. +//! Honouring the flag would suppress exactly what an operator most wants. +//! +//! So the rule is stated explicitly and pinned row-by-row against the measured +//! oracle by `mon20_admin_audit_table`. Unlisted commands are FED, matching +//! Redis's default; Moon's own administrative extensions are listed in +//! [`HIDDEN_WHOLE`]. +//! +//! # Redaction +//! +//! Credentials are replaced **at formatting time** ([`push_arg`] is never +//! reached for them), never written into a buffer and filtered afterwards — a +//! filter is one refactor away from leaking, and the thing it would leak is a +//! password. + +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use arc_swap::ArcSwap; +use bytes::Bytes; +use parking_lot::Mutex; + +use crate::runtime::channel; + +/// Number of attached monitors. The only thing the hot path touches when no +/// monitor exists. +static MONITOR_COUNT: AtomicUsize = AtomicUsize::new(0); + +/// The published sink list. Read on the command path via an `ArcSwap` load, +/// which takes NO lock — the coding rule is "per-shard locks only, no global +/// lock on the write path", and a global `RwLock` read on every fed command +/// would have shared one cacheline across every shard for the whole time a +/// monitor is attached. Writers are attach/detach only, serialised by +/// [`REGISTRY_WRITE`] and published copy-on-write. +static MONITORS: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| ArcSwap::from_pointee(Vec::new())); + +/// Serialises the read-modify-publish of [`MONITORS`]. Never taken on the +/// command path — only by attach and detach, which happen once per monitor. +static REGISTRY_WRITE: Mutex<()> = Mutex::new(()); + +/// One attached MONITOR connection. `Clone` so the registry can publish a new +/// snapshot copy-on-write without taking a lock on the read side. +#[derive(Clone)] +struct MonitorSink { + /// The owning connection's id, so detach is exact. + id: u64, + tx: channel::MpscSender, +} + +/// Commands hidden from the feed in every form. +/// +/// Measured hidden against redis-server 8.6.1, plus Moon's own administrative +/// extensions (`HOTKEYS`, `VACUUM`, `RECLAMATION`, `CDC.READ`, `KILL`), which +/// have no Redis counterpart to measure and are administrative by construction. +const HIDDEN_WHOLE: &[&str] = &[ + // Measured against redis-server 8.6.1. + "CONFIG", + "SLOWLOG", + "LATENCY", + "DEBUG", + "SHUTDOWN", + // MONITOR hides itself by this rule rather than by a self-suppression + // special case — which is also why a second MONITOR emits no line. + "MONITOR", + "MODULE", + "REPLICAOF", + "SLAVEOF", + "FAILOVER", + "SYNC", + "PSYNC", + // Moon extensions, administrative by construction. + "HOTKEYS", + "VACUUM", + "RECLAMATION", + "CDC.READ", + "KILL", +]; + +/// Container commands whose *subcommands* decide visibility. Everything not +/// listed for a container is fed, matching Redis. +const HIDDEN_SUBCOMMANDS: &[(&str, &[&str])] = &[ + ( + "ACL", + &[ + "LIST", "SETUSER", "DELUSER", "GETUSER", "USERS", "LOAD", "SAVE", "LOG", + ], + ), + ( + "CLIENT", + &[ + "LIST", "KILL", "PAUSE", "UNPAUSE", "UNBLOCK", "NO-EVICT", "NO-TOUCH", + ], + ), + ( + "CLUSTER", + &[ + "FORGET", + "MEET", + "RESET", + "SETSLOT", + "ADDSLOTS", + "DELSLOTS", + "ADDSLOTSRANGE", + "DELSLOTSRANGE", + "BUMPEPOCH", + "FAILOVER", + "FLUSHSLOTS", + "SET-CONFIG-EPOCH", + ], + ), + ("MEMORY", &["PURGE"]), +]; + +/// Is this command hidden from the MONITOR feed? +/// +/// `first_arg` is the command's first argument, used for the container +/// commands whose subcommands differ in visibility. +#[inline] +pub fn is_hidden(cmd: &[u8], first_arg: Option<&[u8]>) -> bool { + if HIDDEN_WHOLE + .iter() + .any(|h| cmd.eq_ignore_ascii_case(h.as_bytes())) + { + return true; + } + for (container, subs) in HIDDEN_SUBCOMMANDS { + if cmd.eq_ignore_ascii_case(container.as_bytes()) { + let Some(sub) = first_arg else { + // A container with no subcommand is an arity error and never + // executes, so it is never fed either way. + return false; + }; + return subs.iter().any(|s| sub.eq_ignore_ascii_case(s.as_bytes())); + } + } + false +} + +/// Append one argument, escaped the way Redis's `sdscatrepr` does. +/// +/// Measured byte by byte: `"` -> `\"`, `\` -> `\\`, `\n` `\r` `\t` -> `\n` `\r` +/// `\t`, `0x07` -> `\a`, `0x08` -> `\b`, everything else outside printable +/// ASCII -> `\xHH` lowercase. UTF-8 is escaped per BYTE, not per character. +fn push_arg(out: &mut Vec, arg: &[u8]) { + out.push(b'"'); + for &b in arg { + match b { + b'"' => out.extend_from_slice(b"\\\""), + b'\\' => out.extend_from_slice(b"\\\\"), + b'\n' => out.extend_from_slice(b"\\n"), + b'\r' => out.extend_from_slice(b"\\r"), + b'\t' => out.extend_from_slice(b"\\t"), + 0x07 => out.extend_from_slice(b"\\a"), + 0x08 => out.extend_from_slice(b"\\b"), + 0x20..=0x7e => out.push(b), + other => { + out.extend_from_slice(b"\\x"); + const HEX: &[u8; 16] = b"0123456789abcdef"; + out.push(HEX[(other >> 4) as usize]); + out.push(HEX[(other & 0x0f) as usize]); + } + } + } + out.push(b'"'); +} + +const REDACTED: &[u8] = b"\"(redacted)\""; + +/// Build one feed line, including the trailing CRLF. +/// +/// `+. [ ] "CMD" "arg" …` +pub fn format_line(now_micros: u128, db: usize, addr: &str, cmd: &[u8], args: &[Bytes]) -> Bytes { + let mut out = + Vec::with_capacity(64 + cmd.len() + args.iter().map(|a| a.len() + 3).sum::()); + out.push(b'+'); + + let secs = now_micros / 1_000_000; + let micros = (now_micros % 1_000_000) as u32; + let mut itoa_buf = itoa::Buffer::new(); + out.extend_from_slice(itoa_buf.format(secs).as_bytes()); + out.push(b'.'); + // Zero-padded to exactly 6 digits; a client parses this as a fixed shape. + let mut d = [b'0'; 6]; + let mut v = micros; + for slot in d.iter_mut().rev() { + *slot = b'0' + (v % 10) as u8; + v /= 10; + } + out.extend_from_slice(&d); + + out.extend_from_slice(b" ["); + out.extend_from_slice(itoa_buf.format(db).as_bytes()); + out.push(b' '); + out.extend_from_slice(addr.as_bytes()); + out.extend_from_slice(b"] "); + + push_arg(&mut out, cmd); + + // Credential handling is decided BEFORE any argument is written, so a + // secret never enters the buffer at all. + let auth_all = cmd.eq_ignore_ascii_case(b"AUTH"); + let hello_auth_at = if cmd.eq_ignore_ascii_case(b"HELLO") { + args.iter() + .position(|a| a.as_ref().eq_ignore_ascii_case(b"AUTH")) + } else { + None + }; + + for (i, arg) in args.iter().enumerate() { + out.push(b' '); + let redact = auth_all || hello_auth_at.is_some_and(|k| i == k + 1 || i == k + 2); + if redact { + out.extend_from_slice(REDACTED); + } else { + push_arg(&mut out, arg); + } + } + + out.extend_from_slice(b"\r\n"); + Bytes::from(out) +} + +/// Attach a connection to the feed. Idempotent per `id`: a second `MONITOR` on +/// an already-attached connection must not double-register, or every line +/// would be delivered twice. +pub fn attach(id: u64, tx: channel::MpscSender) -> bool { + let _w = REGISTRY_WRITE.lock(); + let cur = MONITORS.load(); + if cur.iter().any(|m| m.id == id) { + return false; + } + let mut next: Vec = cur.iter().cloned().collect(); + next.push(MonitorSink { id, tx }); + MONITOR_COUNT.store(next.len(), Ordering::Relaxed); + MONITORS.store(Arc::new(next)); + true +} + +/// Detach a connection (RESET, disconnect, or shed for being slow). +pub fn detach(id: u64) { + let _w = REGISTRY_WRITE.lock(); + let cur = MONITORS.load(); + if !cur.iter().any(|m| m.id == id) { + return; + } + let next: Vec = cur.iter().filter(|m| m.id != id).cloned().collect(); + MONITOR_COUNT.store(next.len(), Ordering::Relaxed); + MONITORS.store(Arc::new(next)); +} + +/// Is any monitor attached at all? One `Relaxed` load. +/// +/// Used by the inline fast path, which cannot format a feed line (it has no +/// peer address) and therefore stands down entirely while monitoring is on. +#[inline] +pub fn any_attached() -> bool { + MONITOR_COUNT.load(Ordering::Relaxed) != 0 +} + +#[inline] +pub fn is_attached(id: u64) -> bool { + if MONITOR_COUNT.load(Ordering::Relaxed) == 0 { + return false; + } + MONITORS.load().iter().any(|m| m.id == id) +} + +/// Emit one command to every attached monitor. +/// +/// The entire cost when unattached is the `Relaxed` load below. +/// +/// A monitor whose queue is full has its **sink dropped**, which closes its +/// channel and ends its connection. Contracted at freeze: dropping lines +/// silently would leave an operator unable to tell a quiet server from a lossy +/// feed, and blocking would let one slow TCP reader stall every shard. +#[inline] +pub fn feed(db: usize, addr: &str, cmd: &[u8], args: &[Bytes]) { + if MONITOR_COUNT.load(Ordering::Relaxed) == 0 { + return; + } + feed_cold(db, addr, cmd, args); +} + +/// [`feed`] for a protocol-level argument list. +/// +/// Kept separate so the hot path never materialises a `Vec`: the +/// conversion happens inside the cold path, after the attached check. +#[inline] +pub fn feed_frames(db: usize, addr: &str, cmd: &[u8], args: &[crate::protocol::Frame]) { + if MONITOR_COUNT.load(Ordering::Relaxed) == 0 { + return; + } + let owned: Vec = args + .iter() + .filter_map(|f| crate::command::helpers::extract_bytes(f).cloned()) + .collect(); + feed_cold(db, addr, cmd, &owned); +} + +#[cold] +fn feed_cold(db: usize, addr: &str, cmd: &[u8], args: &[Bytes]) { + let first = args.first().map(|a| a.as_ref()); + if is_hidden(cmd, first) { + return; + } + // Redis feeds a command when it EXECUTES. One rejected earlier — an unknown + // name, or an arity violation — never executes, so it never appears. Both + // checks live here in the cold path, so the attached-nobody case still pays + // only the atomic load above. + let Some(meta) = crate::command::metadata::lookup(cmd) else { + return; + }; + let argc = args.len() as i16 + 1; + let arity_ok = if meta.arity >= 0 { + argc == meta.arity + } else { + argc >= -meta.arity + }; + if !arity_ok { + return; + } + // `SELECT n` reports the db it MOVES TO, not the one it came from — + // measured (`SELECT 3` logs as `[3 …]`, not `[0 …]`). Everything else is + // fed before dispatch, so the connection's current db is correct; SELECT is + // the one command whose own effect changes the field, and taking it from + // the argument avoids feeding after execution just for this case (which + // would delay every blocking command to its unblock, per the freeze). + let db = if cmd.eq_ignore_ascii_case(b"SELECT") { + args.first() + .and_then(|a| std::str::from_utf8(a).ok()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(db) + } else { + db + }; + let now_micros = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_micros()); + let line = format_line(now_micros, db, addr, cmd, args); + + // Serialise once, fan out over a lock-free snapshot. Collect the slow ones + // and detach them afterwards, so the publish path is never entered while + // iterating the list being replaced. + let mut slow: Vec = Vec::new(); + for m in MONITORS.load().iter() { + if m.tx.try_send(line.clone()).is_err() { + slow.push(m.id); + } + } + for id in slow { + detach(id); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn b(v: &[u8]) -> Bytes { + Bytes::copy_from_slice(v) + } + + /// `MONITOR_COUNT` and `MONITORS` are process-global, so the two tests that + /// assert on them must not interleave: cargo runs tests in parallel by + /// default, and `attach_is_idempotent_per_connection` briefly makes the + /// count non-zero, which is exactly what the zero-cost test asserts against. + static REGISTRY_TESTS: parking_lot::Mutex<()> = parking_lot::Mutex::new(()); + + #[test] + fn monitor_registry_zero_cost_when_unattached() { + // M11 as a property of the code: with no monitor attached, `feed` must + // not reach formatting at all. Proven by the count being the only + // thing consulted — `feed_cold` is where every allocation lives, and + // it is unreachable while the count is zero. + let _guard = REGISTRY_TESTS.lock(); + assert_eq!(MONITOR_COUNT.load(Ordering::Relaxed), 0); + // A huge argument list that would be expensive to format. + let args: Vec = (0..1000).map(|i| b(format!("arg{i}").as_bytes())).collect(); + feed(0, "127.0.0.1:1", b"SET", &args); + assert_eq!( + MONITOR_COUNT.load(Ordering::Relaxed), + 0, + "feed must not disturb the registry when unattached" + ); + } + + #[test] + fn monitor_redaction_never_buffers_the_secret() { + // The §0 split decision made "redact at formatting time, never filter + // afterwards" a design constraint. This inspects the formatter's own + // output, so a version that emitted the password and stripped it later + // would still fail here only if the strip were removed — which is why + // the assertion is on the ABSENCE of the secret in the bytes. + let line = format_line( + 1_700_000_000_123_456, + 0, + "127.0.0.1:1", + b"AUTH", + &[b(b"hunter2")], + ); + let text = String::from_utf8_lossy(&line).to_string(); + assert!(!text.contains("hunter2"), "got {text:?}"); + assert!(text.contains(r#""AUTH" "(redacted)""#), "got {text:?}"); + + let line = format_line( + 1_700_000_000_123_456, + 0, + "127.0.0.1:1", + b"AUTH", + &[b(b"user"), b(b"pw")], + ); + let text = String::from_utf8_lossy(&line).to_string(); + assert!( + !text.contains("user") && !text.contains("pw"), + "got {text:?}" + ); + assert!( + text.contains(r#""AUTH" "(redacted)" "(redacted)""#), + "got {text:?}" + ); + + let line = format_line( + 1_700_000_000_123_456, + 0, + "127.0.0.1:1", + b"HELLO", + &[b(b"3"), b(b"AUTH"), b(b"default"), b(b"sekrit")], + ); + let text = String::from_utf8_lossy(&line).to_string(); + assert!(!text.contains("sekrit"), "got {text:?}"); + assert!( + text.contains(r#""HELLO" "3" "AUTH" "(redacted)" "(redacted)""#), + "the version survives; only the credentials go. got {text:?}" + ); + } + + #[test] + fn line_shape_and_escaping() { + let line = format_line( + 1_700_000_000_000_007, + 3, + "127.0.0.1:5555", + b"SET", + &[b(b"esc"), b(b"a\"b\\c\nd\re\tf\x00g\xffh")], + ); + let text = String::from_utf8_lossy(&line).to_string(); + assert!( + text.starts_with("+1700000000.000007 [3 127.0.0.1:5555] "), + "micros zero-pad to 6 digits; got {text:?}" + ); + assert!( + text.contains(r#""SET" "esc" "a\"b\\c\nd\re\tf\x00g\xffh""#), + "got {text:?}" + ); + assert!(text.ends_with("\r\n")); + } + + #[test] + fn utf8_escapes_per_byte_and_empty_arg_is_two_quotes() { + let line = format_line(1, 0, "a:1", b"SET", &[b("h\u{e9}llo".as_bytes()), b(b"")]); + let text = String::from_utf8_lossy(&line).to_string(); + assert!(text.contains(r#""h\xc3\xa9llo""#), "got {text:?}"); + assert!(text.ends_with("\"\"\r\n"), "got {text:?}"); + } + + #[test] + fn hidden_set_matches_the_measured_oracle() { + // Every row measured against redis-server 8.6.1. The container rows are + // the point: Moon's own ADMIN flag cannot express them. + for (cmd, sub) in [ + (&b"CONFIG"[..], Some(&b"GET"[..])), + (b"CONFIG", Some(&b"SET"[..])), + (b"SLOWLOG", Some(&b"LEN"[..])), + (b"LATENCY", Some(&b"RESET"[..])), + (b"MONITOR", None), + (b"ACL", Some(&b"LIST"[..])), + (b"ACL", Some(&b"SETUSER"[..])), + (b"CLIENT", Some(&b"LIST"[..])), + ] { + assert!(is_hidden(cmd, sub), "{:?} {:?} must be hidden", cmd, sub); + } + for (cmd, sub) in [ + (&b"INFO"[..], Some(&b"server"[..])), + (b"DBSIZE", None), + (b"LASTSAVE", None), + (b"COMMAND", Some(&b"COUNT"[..])), + (b"CLIENT", Some(&b"GETNAME"[..])), + (b"CLIENT", Some(&b"ID"[..])), + (b"ACL", Some(&b"WHOAMI"[..])), + (b"ACL", Some(&b"CAT"[..])), + (b"CLUSTER", Some(&b"INFO"[..])), + (b"MEMORY", Some(&b"USAGE"[..])), + // Fed despite Redis's own skip_monitor flag — measured. + (b"EVAL", Some(&b"return 1"[..])), + (b"EVALSHA", Some(&b"abc"[..])), + (b"SET", Some(&b"k"[..])), + ] { + assert!(!is_hidden(cmd, sub), "{:?} {:?} must be fed", cmd, sub); + } + } + + #[test] + fn attach_is_idempotent_per_connection() { + // A second MONITOR on the same connection must not double-register, or + // every command would be delivered twice. + let _guard = REGISTRY_TESTS.lock(); + let (tx, _rx) = channel::mpsc_bounded::(4); + let (tx2, _rx2) = channel::mpsc_bounded::(4); + let id = 987_654; + assert!(attach(id, tx), "first attach registers"); + assert!(!attach(id, tx2), "second attach on the same id is refused"); + detach(id); + assert!(!is_attached(id)); + } +} diff --git a/src/scripting/bridge.rs b/src/scripting/bridge.rs index 745aea7e..a67adc95 100644 --- a/src/scripting/bridge.rs +++ b/src/scripting/bridge.rs @@ -439,6 +439,19 @@ pub fn make_redis_call_fn( } } + // MONITOR: a script-issued command is fed with the literal `lua` + // in place of a peer address — measured against redis-server 8.6.1, + // which emits `[0 lua] "SET" "lk" "v"` after the client's own + // `[0 127.0.0.1:… ] "eval" …` line. This is the one hook site with + // no connection behind it, so the handler-level hooks structurally + // cannot cover it: without this call an operator watching a + // script-driven workload sees every EVAL and none of its effects. + // + // Fed BEFORE execution, matching every other hook site, so ordering + // is issue-order. Costs one `Relaxed` load per `redis.call` when no + // monitor is attached. + crate::monitor::feed_frames(db_idx, "lua", &cmd_bytes, &frames[1..]); + let frame = db.execute_command(&cmd_bytes, &frames[1..], &mut db_idx, db_count); // Wave A part 2 (task #34): dual-plane (AOF + replication) diff --git a/src/server/conn/core.rs b/src/server/conn/core.rs index f8835720..ef273551 100644 --- a/src/server/conn/core.rs +++ b/src/server/conn/core.rs @@ -251,6 +251,12 @@ pub(crate) struct ConnectionState { pub pubsub_tx: Option>, pub pubsub_rx: Option>, + // MONITOR. Separate from the pub/sub channel because a connection can be a + // monitor without being a subscriber, and Redis's rules for the two modes + // differ (a monitor may not touch the keyspace; a subscriber may). + pub monitor_attached: bool, + pub monitor_rx: Option>, + // Transaction (MULTI/EXEC) pub in_multi: bool, /// Active cross-store transaction (None if not in transaction). @@ -362,6 +368,8 @@ impl ConnectionState { subscriber_id: 0, pubsub_tx: None, pubsub_rx: None, + monitor_attached: false, + monitor_rx: None, in_multi: false, active_cross_txn: None, workspace_id: migrated.and_then(|s| s.workspace_id), @@ -497,6 +505,15 @@ impl ConnectionState { && self.subscription_count == 0 && !self.tracking_state.enabled && !self.saw_replconf + // A monitor's registration is process-global and keyed by client + // id. Migration returns from the handler through its OWN path, + // before the disconnect detach block runs, so a migrated monitor + // would leave a dead sink registered forever — which also pins + // `any_attached()` true and holds the inline fast path down for the + // life of the process. Excluding monitors keeps every teardown on + // the paths that actually detach; a monitor connection is a + // diagnostic session, not something worth migrating. + && !self.monitor_attached } /// Get the active transaction's ID, if any. diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 07730ae6..0eabb13e 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -50,6 +50,16 @@ pub(super) fn check_auth_gate( if conn.authenticated { return AuthGateResult::Authenticated; } + // MONITOR: feed the pre-auth AUTH/HELLO here, because this gate `continue`s + // and the main feed hook further down is never reached for them. That makes + // the FIRST AUTH of a session — the only one carrying a credential on a + // password-protected server — the one command the feed would otherwise miss + // entirely. Redaction happens in the formatter, so nothing leaks. + if let Some((cmd, cmd_args)) = extract_command(frame) + && (cmd.eq_ignore_ascii_case(b"AUTH") || cmd.eq_ignore_ascii_case(b"HELLO")) + { + crate::monitor::feed_frames(conn.selected_db, peer_addr, cmd, cmd_args); + } match extract_command(frame) { Some((cmd, cmd_args)) if cmd.eq_ignore_ascii_case(b"AUTH") => { let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &ctx.acl_table); diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index a3b8da63..9d17b369 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1036,6 +1036,56 @@ pub(crate) async fn handle_connection_sharded_monoio< } continue; } + } else if let Some(rx) = conn.monitor_rx.as_ref() + && conn.monitor_attached + { + // MONITOR: deliver feed lines while parked in read(). A monitor + // connection stays in this loop — it is not diverted anywhere — + // because Redis keeps serving it (PING, INFO, RESET all work while + // attached); only keyspace commands are refused, and that refusal + // is a gate below, not a separate mode loop. + // + // The channel carries already-formatted `+…\r\n` lines, so this + // arm writes them verbatim. Same tearing argument as the RESP3 + // subscriber arm above: both a reply and a feed line are whole + // frames written by this one task, and the loop only parks here + // when no reply is in flight. + let mon_buf = std::mem::take(&mut tmp_buf); + let mut line: Option = None; + monoio::select! { + _ = shutdown.cancelled() => { break; } + read_result = stream.read(mon_buf) => { + let (result, returned_buf) = read_result; + tmp_buf = returned_buf; + match result { + Ok(0) => break, + Ok(n) => { read_buf.extend_from_slice(&tmp_buf[..n]); } + Err(_) => break, + } + } + msg = rx.recv_async() => { + line = msg.ok(); + } + } + if let Some(data) = line { + if !write_all_bounded!( + stream, + data, + write_timeout, + out_cap_normal, + client_live, + client_id + ) { + break; + } + continue; + } else if conn.monitor_attached && !crate::monitor::is_attached(client_id) { + // The registry dropped this sink because the connection could + // not keep up. Contracted policy: the monitor DIES, loudly, + // rather than silently receiving an incomplete feed that an + // operator would read as a quiet server. + break; + } } else if conn.tracking_rx.is_some() { // CLIENT TRACKING: deliver invalidation Push frames while parked // in read(). Only tracking connections take this select — the @@ -1341,9 +1391,20 @@ pub(crate) async fn handle_connection_sharded_monoio< // `*1[$1 v]`. `MGET`, not being inline-eligible, queued correctly // all along, which is what isolated the path. // See `tests/multi_queues_inline_get.rs`. + // MONITOR: the inline path answers straight from the read buffer + // and never sees `peer_addr`, so it cannot format a feed line. Rather + // than thread the address through the hottest function in the + // codebase, the fast path stands down while any monitor is attached + // — the feed is then correct BY CONSTRUCTION on this path instead of + // by a hook someone must remember to keep in sync. The cost is + // throughput while monitoring, which is the moment an operator has + // already accepted diagnostic overhead; when unattached this is one + // Relaxed load that the branch below would take anyway. + let monitored = crate::monitor::any_attached(); let can_inline_reads = - acl_unrestricted && !conn.in_multi && !conn.tracking_state.enabled; + acl_unrestricted && !conn.in_multi && !conn.tracking_state.enabled && !monitored; let can_inline_writes = acl_unrestricted + && !monitored && !conn.in_multi && !conn.tracking_state.enabled && !crate::tracking::tracking_active() @@ -1529,6 +1590,7 @@ pub(crate) async fn handle_connection_sharded_monoio< should_quit = true; break; } + // --- ASKING --- if cmd.eq_ignore_ascii_case(b"ASKING") { conn.asking = true; @@ -1564,6 +1626,19 @@ pub(crate) async fn handle_connection_sharded_monoio< // connection-level command names. Cut per-command dispatch cost // from ~14 non-matching function calls to ~1 on SET/GET workloads. let cmd_len = cmd.len(); + // MONITOR feed for the two ACL-EXEMPT commands below. + // + // AUTH and HELLO carry Redis's NO_AUTH flag and are therefore + // intercepted ABOVE the ACL gate — which puts them above the main + // feed hook further down, so they would never be fed at all. They + // are also the only two commands carrying credentials, i.e. exactly + // the ones whose absence from the feed is least acceptable and + // whose arguments must be redacted. Fed here, once; the intercepts + // `continue`, so they cannot reach the main hook and double-feed. + if cmd.eq_ignore_ascii_case(b"AUTH") || cmd.eq_ignore_ascii_case(b"HELLO") { + crate::monitor::feed_frames(conn.selected_db, &peer_addr, cmd, cmd_args); + } + // === ACL-EXEMPT COMMANDS === // // AUTH and HELLO carry Redis's `NO_AUTH` flag and are permitted @@ -1698,6 +1773,65 @@ pub(crate) async fn handle_connection_sharded_monoio< continue; } + // --- MONITOR: attach, and the rules that apply once attached --- + // + // BELOW the ACL gate and BELOW the MULTI queue gate, deliberately. + // Above the ACL gate it would exempt itself from ACL, which is the + // exact shape of the v0.8.6 P0 — and MONITOR is the last command + // that should be reachable without permission, since it reads every + // other user's arguments. Below the MULTI gate, a queued command + // `continue`s before reaching the feed, which is how M9 (queued + // commands are fed at EXEC, not at queue time) falls out for free + // rather than needing its own special case. + if cmd.eq_ignore_ascii_case(b"MONITOR") { + // Already attached -> None: Redis answers NOTHING. Measured — + // it is not an error, it is silence. + if let Some(reply) = crate::server::conn::monitor_mode::handle_monitor( + cmd_args.len(), + client_id, + &mut conn.monitor_attached, + &mut conn.monitor_rx, + ) { + responses.push(reply); + } + continue; + } + if conn.monitor_attached + && let Some(refusal) = crate::server::conn::monitor_mode::refuse_if_keyspace( + cmd, + cmd_args + .first() + .and_then(crate::command::helpers::extract_bytes) + .map(|b| b.as_ref()), + ) + { + responses.push(refusal); + continue; + } + + // --- MONITOR feed --- + // + // Before dispatch, so a BLOCKING command appears when it is issued + // rather than when it unblocks — that ordering is the whole reason + // an operator watches the feed. Costs one Relaxed atomic load when + // no monitor is attached; everything else (the hidden-set check, + // the registry/arity check, formatting, redaction, fan-out) lives + // behind that load in a #[cold] path. + // + // EXEC first replays its queue: Redis feeds a queued command when + // it EXECUTES, so the body lines appear at EXEC time, in order, + // followed by the EXEC line itself (measured 5µs apart). A dirty + // transaction is refused wholesale and never runs, so it feeds + // nothing. + if cmd.eq_ignore_ascii_case(b"EXEC") && !conn.multi_dirty { + for queued in &conn.command_queue { + if let Some((qcmd, qargs)) = extract_command(queued) { + crate::monitor::feed_frames(conn.selected_db, &peer_addr, qcmd, qargs); + } + } + } + crate::monitor::feed_frames(conn.selected_db, &peer_addr, cmd, cmd_args); + if cmd_len == 7 && dispatch::try_handle_cluster(cmd, cmd_args, ctx, &mut responses) { continue; } @@ -3595,6 +3729,17 @@ pub(crate) async fn handle_connection_sharded_monoio< } } + // --- Disconnect cleanup: detach from the MONITOR feed --- + // + // Unconditional and cheap: `detach` is a no-op for a connection that was + // never a monitor. Leaving a dead sink registered would keep the feed + // formatting and fanning out to a closed channel, and would keep + // `any_attached()` true — which also holds the inline fast path down for + // the rest of the process. + if conn.monitor_attached { + crate::monitor::detach(client_id); + } + // --- Disconnect cleanup: release CLIENT TRACKING registration --- // A client that disconnects without `CLIENT TRACKING OFF` would otherwise // leave `ACTIVE_TRACKERS` nonzero and keep `tracking_active()` hot for the diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 2c27d472..8d39529f 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -619,6 +619,22 @@ pub(crate) async fn handle_connection_sharded_inner< frame_idx += 1; // --- AUTH gate --- if !conn.authenticated { + // MONITOR: feed the pre-auth AUTH/HELLO here — this gate + // `continue`s, so the main hook below never sees them, + // and the FIRST AUTH of a session is the only command + // carrying a credential on a password-protected server. + // Redaction happens in the formatter; nothing leaks. + if let Some((cmd, cmd_args)) = extract_command(&frame) + && (cmd.eq_ignore_ascii_case(b"AUTH") + || cmd.eq_ignore_ascii_case(b"HELLO")) + { + crate::monitor::feed_frames( + conn.selected_db, + &peer_addr, + cmd, + cmd_args, + ); + } match extract_command(&frame) { Some((cmd, cmd_args)) if cmd.eq_ignore_ascii_case(b"AUTH") => { let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &ctx.acl_table); @@ -736,6 +752,15 @@ pub(crate) async fn handle_connection_sharded_inner< break; } + // MONITOR feed for the two ACL-EXEMPT commands below. + // AUTH and HELLO are intercepted above the ACL gate, hence + // above the main feed hook, so they would never be fed — + // and they are precisely the two carrying credentials. + // Their intercepts `continue`, so no double-feed. + if cmd.eq_ignore_ascii_case(b"AUTH") || cmd.eq_ignore_ascii_case(b"HELLO") { + crate::monitor::feed_frames(conn.selected_db, &peer_addr, cmd, cmd_args); + } + // === ACL-EXEMPT COMMANDS === // // AUTH and HELLO carry Redis's `NO_AUTH` flag and are @@ -914,6 +939,64 @@ pub(crate) async fn handle_connection_sharded_inner< continue; } + // --- MONITOR: attach, and the rules once attached --- + // + // Identical placement to handler_monoio: BELOW the ACL gate + // (above it, MONITOR would exempt itself from ACL — the + // v0.8.6 P0 shape, and MONITOR reads every other user's + // arguments) and BELOW the MULTI queue gate (so a queued + // command `continue`s before the feed, which is how "queued + // commands are fed at EXEC, not at queue time" falls out + // without a special case). + if cmd.eq_ignore_ascii_case(b"MONITOR") { + // The attach rule lives in `monitor_mode`, once. This + // block used to be a hand-written second copy of the + // monoio one and had already drifted in structure. + // Already attached -> None: Redis answers NOTHING. + if let Some(reply) = + crate::server::conn::monitor_mode::handle_monitor( + cmd_args.len(), + conn.client_id, + &mut conn.monitor_attached, + &mut conn.monitor_rx, + ) + { + responses.push(reply); + } + continue; + } + if conn.monitor_attached + && let Some(refusal) = + crate::server::conn::monitor_mode::refuse_if_keyspace( + cmd, + cmd_args.first().and_then(crate::command::helpers::extract_bytes).map(|b| b.as_ref()), + ) + { + responses.push(refusal); + continue; + } + + // --- MONITOR feed --- + // + // Before dispatch, so a blocking command appears when it is + // issued rather than when it unblocks. One Relaxed atomic + // load when unattached; everything else is behind it in a + // #[cold] path. EXEC replays its queue first, in order, + // followed by the EXEC line. + if cmd.eq_ignore_ascii_case(b"EXEC") && !conn.multi_dirty { + for queued in &conn.command_queue { + if let Some((qcmd, qargs)) = extract_command(queued) { + crate::monitor::feed_frames( + conn.selected_db, + &peer_addr, + qcmd, + qargs, + ); + } + } + } + crate::monitor::feed_frames(conn.selected_db, &peer_addr, cmd, cmd_args); + // --- CLUSTER subcommands --- if cmd.eq_ignore_ascii_case(b"CLUSTER") { if let Some(ref cs) = ctx.cluster_state { @@ -2689,6 +2772,28 @@ pub(crate) async fn handle_connection_sharded_inner< } } } + // MONITOR: deliver feed lines while the connection is idle in + // read(). Pending-branch when unattached, so a non-monitor + // connection pays nothing. The channel carries already-formatted + // `+…\r\n` lines, written verbatim. + mon_line = async { + match conn.monitor_rx { + Some(ref rx) => rx.recv_async().await.ok(), + None => std::future::pending().await, + } + } => { + match mon_line { + Some(data) => { + if !write_all_bounded!(stream, &data, write_timeout, out_cap_normal, client_live, client_id) { + break; + } + } + // The registry dropped this sink because the connection + // could not keep up. Contracted policy: the monitor dies + // loudly rather than silently receiving a partial feed. + None => break, + } + } // CLIENT TRACKING: deliver invalidation Push frames while the // connection is idle in read(). Pending-branch when tracking is // off — zero cost for non-tracking connections. @@ -2740,6 +2845,13 @@ pub(crate) async fn handle_connection_sharded_inner< .await; } + // Detach from the MONITOR feed. A retained dead sink would keep the feed + // formatting into a closed channel and keep `any_attached()` true, which + // also holds the inline fast path down for the rest of the process. + if conn.monitor_attached { + crate::monitor::detach(client_id); + } + // Clean up pub/sub subscriptions on disconnect if conn.subscriber_id > 0 { let removed_channels = { diff --git a/src/server/conn/mod.rs b/src/server/conn/mod.rs index 9f60a894..99e682e0 100644 --- a/src/server/conn/mod.rs +++ b/src/server/conn/mod.rs @@ -7,6 +7,7 @@ pub mod handler_monoio; pub mod handler_sharded; #[cfg(feature = "runtime-tokio")] pub mod handler_single; +pub mod monitor_mode; pub mod park_policy; pub mod shared; pub mod subscriber_mode; diff --git a/src/server/conn/monitor_mode.rs b/src/server/conn/monitor_mode.rs new file mode 100644 index 00000000..251d0158 --- /dev/null +++ b/src/server/conn/monitor_mode.rs @@ -0,0 +1,249 @@ +//! The rules that apply to a connection once it is attached as a `MONITOR`. +//! +//! Stated ONCE and consulted by every handler. The pub/sub subscriber-mode +//! allow-list was the counter-example: it lived in three handlers with two +//! different texts and two different behaviours, and none of them matched +//! Redis. A rule restated in N places drifts to N behaviours. + +use bytes::Bytes; + +use crate::command::metadata::CommandFlags; +use crate::protocol::Frame; + +/// Redis's verbatim refusal. A monitor is flagged as a replica internally, +/// which is why the message talks about replicas rather than monitors — the +/// text is measured, not composed. +const ERR_KEYSPACE: &[u8] = b"ERR Replica can't interact with the keyspace"; + +/// Keyspace commands that name NO key (`first_key == 0`), so `first_key` alone +/// cannot find them. Measured refused against redis-server 8.6.1 (2026-08-14, +/// one fresh connection per probe). +const REFUSED_BY_NAME: &[&str] = &[ + "DBSIZE", + "KEYS", + "SCAN", + "RANDOMKEY", + "EVAL", + "EVALSHA", + "EVAL_RO", + "EVALSHA_RO", + "PUBLISH", + "SPUBLISH", +]; + +/// Container commands whose *subcommands* differ, exactly as the MONITOR +/// hidden-set does: `MEMORY USAGE` is refused, `MEMORY DOCTOR` is served. +const REFUSED_SUBCOMMANDS: &[(&str, &[&str])] = &[("MEMORY", &["USAGE"])]; + +/// `SELECT` carries Moon's `WRITE` flag (it mutates connection state) but is +/// served on a monitor connection, and its feed line is how the `[db …]` field +/// is observed changing. The one exception to the `WRITE` term below. +const SERVED_DESPITE_WRITE: &[&str] = &["SELECT"]; + +/// Should this command be refused on an attached monitor connection? +/// +/// The rule is: it names a key (`first_key != 0`), OR it carries `WRITE`, OR it +/// is one of the measured zero-key keyspace commands above. +/// +/// Two rules were tried and are measurably wrong; both are recorded because the +/// tempting fix is to go back to one of them: +/// +/// * `first_key != 0` alone SERVES `DBSIZE`, `KEYS`, `SCAN`, `RANDOMKEY`, +/// `FLUSHALL`, `FLUSHDB`, `SWAPDB`, `EVAL` and `PUBLISH`, all of which Redis +/// refuses. +/// * `WRITE | READONLY` — the flags Redis itself uses — REFUSES `PING`, +/// `ECHO`, `TIME`, `INFO`, `COMMAND`, `LASTSAVE` and `WAIT`, because Moon +/// flags all of them `READONLY` and Redis flags none of them so. This is the +/// same trap as `CommandFlags::ADMIN` in the MONITOR hidden-set: Moon's +/// flags are not Redis's flags, and reusing them silently changes behaviour. +/// +/// So the rule is stated explicitly and pinned row by row by +/// `mon23_refusal_rule_is_write_or_readonly_not_first_key`. +/// +/// An unknown command returns `None` — dispatch refuses it later with the +/// unknown-command error, which is what Redis does too. +pub fn refuse_if_keyspace(cmd: &[u8], first_arg: Option<&[u8]>) -> Option { + let meta = crate::command::metadata::lookup(cmd)?; + + let by_sub = REFUSED_SUBCOMMANDS.iter().find_map(|(container, subs)| { + if !cmd.eq_ignore_ascii_case(container.as_bytes()) { + return None; + } + let sub = first_arg?; + Some(subs.iter().any(|s| sub.eq_ignore_ascii_case(s.as_bytes()))) + }); + if let Some(refused) = by_sub { + return refused.then(|| Frame::Error(Bytes::from_static(ERR_KEYSPACE))); + } + + let named = REFUSED_BY_NAME + .iter() + .any(|n| cmd.eq_ignore_ascii_case(n.as_bytes())); + let writes = meta.flags.contains(CommandFlags::WRITE) + && !SERVED_DESPITE_WRITE + .iter() + .any(|n| cmd.eq_ignore_ascii_case(n.as_bytes())); + + if meta.first_key != 0 || writes || named { + return Some(Frame::Error(Bytes::from_static(ERR_KEYSPACE))); + } + None +} + +/// `MONITOR` — attach this connection to the feed. The rule lives here, once, +/// and both handlers call it; the previous version had the monoio copy in +/// `handler_monoio::dispatch` and a hand-written second copy in +/// `handler_sharded`, which had already drifted in structure. Two copies of an +/// attach rule diverge the same way the subscriber-mode allow-list did. +/// +/// Returns the reply to send, or `None` when the connection is ALREADY attached: +/// Redis answers a second `MONITOR` with nothing at all. Measured — silence, +/// not an error. +/// +/// The ACL check is not repeated here: this is reached only below the ACL gate, +/// and `MONITOR` carries the admin category in `COMMAND_META`, so a non-admin +/// user is refused by the general gate with the same `NOPERM` text. +pub fn handle_monitor( + arg_count: usize, + client_id: u64, + already_attached: &mut bool, + rx_slot: &mut Option>, +) -> Option { + if arg_count != 0 { + return Some(Frame::Error(Bytes::from_static( + b"ERR wrong number of arguments for 'monitor' command", + ))); + } + if *already_attached { + return None; + } + // Bounded, and deliberately not large. A monitor that cannot keep up has + // this channel fill; the feed then drops the SINK, which closes the channel + // and ends the connection. Contracted at freeze: silently skipping lines + // would leave an operator unable to tell a quiet server from a lossy feed, + // and blocking would let one slow reader stall every shard. + let (tx, rx) = crate::runtime::channel::mpsc_bounded::(MONITOR_QUEUE_DEPTH); + if crate::monitor::attach(client_id, tx) { + *already_attached = true; + *rx_slot = Some(rx); + return Some(Frame::SimpleString(Bytes::from_static(b"OK"))); + } + // The registry already holds a sink for this id while THIS connection + // believes it is unattached — a stale registration (a reused client id, or + // a teardown path that did not run). Marking the connection attached and + // dropping the receiver, as an earlier version did, produced the worst + // possible state: keyspace commands refused, no reply, no feed line, and no + // way to notice. Evict the stale sink and take the registration. + crate::monitor::detach(client_id); + let (tx, rx) = crate::runtime::channel::mpsc_bounded::(MONITOR_QUEUE_DEPTH); + if crate::monitor::attach(client_id, tx) { + *already_attached = true; + *rx_slot = Some(rx); + return Some(Frame::SimpleString(Bytes::from_static(b"OK"))); + } + // Unreachable in practice: nothing else can register this id concurrently, + // because a client id belongs to exactly one connection task. Fail LOUDLY + // rather than leaving the connection in the half-attached state above. + *already_attached = false; + *rx_slot = None; + Some(Frame::Error(Bytes::from_static( + b"ERR MONITOR could not attach: the feed registry is holding a stale registration for this connection", + ))) +} + +/// Per-monitor queue depth. See `handle_monitor` for why it is bounded and what +/// happens when it fills. +const MONITOR_QUEUE_DEPTH: usize = 4096; + +#[cfg(test)] +mod tests { + use super::*; + + fn refused(cmd: &[u8]) -> bool { + refuse_if_keyspace(cmd, None).is_some() + } + + #[test] + fn keyspace_commands_are_refused() { + for c in [&b"SET"[..], b"GET", b"DEL", b"INCR", b"HSET", b"LPUSH"] { + assert!( + refused(c), + "{:?} addresses a key", + String::from_utf8_lossy(c) + ); + } + } + + #[test] + fn zero_first_key_commands_are_still_refused() { + // The regression guard for the original `first_key != 0` rule: every + // one of these carries `first_key == 0` and every one is refused by + // redis-server 8.6.1 (measured 2026-08-14, one connection per probe). + for c in [ + &b"DBSIZE"[..], + b"KEYS", + b"SCAN", + b"RANDOMKEY", + b"FLUSHALL", + b"FLUSHDB", + b"SWAPDB", + b"EVAL", + b"EVALSHA", + b"PUBLISH", + b"SPUBLISH", + // `EVAL_RO` / `EVALSHA_RO` are deliberately absent: they are not in + // Moon's registry at all, so `lookup` returns None and dispatch + // refuses them as unknown commands. They stay listed in + // `REFUSED_BY_NAME` so the rule is already right if they land. + ] { + assert!( + refused(c), + "{:?} has first_key == 0 and IS refused — this is the exact \ + row the first implementation got wrong", + String::from_utf8_lossy(c) + ); + } + } + + #[test] + fn connection_and_server_commands_are_served() { + // Measured on a real monitor connection: each of these works. Note + // `DBSIZE` is deliberately ABSENT — an earlier version of this test + // asserted it was served, which was a wrong belief encoded as a test + // and is what let the wrong rule ship green. + for c in [ + &b"PING"[..], + b"INFO", + b"CLIENT", + b"ACL", + b"SUBSCRIBE", + b"RESET", + b"QUIT", + b"COMMAND", + b"LASTSAVE", + b"TIME", + b"ECHO", + b"SELECT", + b"BGSAVE", + ] { + assert!( + !refused(c), + "{:?} is served on a monitor connection", + String::from_utf8_lossy(c) + ); + } + } + + #[test] + fn the_refusal_text_is_verbatim_redis() { + let f = refuse_if_keyspace(b"SET", None).expect("SET is refused"); + match f { + Frame::Error(e) => assert_eq!( + &e[..], + b"ERR Replica can't interact with the keyspace", + "byte-compared: a near-miss here is a client-visible divergence" + ), + other => panic!("expected an error frame, got {other:?}"), + } + } +} diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index 0c0e75bc..8dff0c96 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -1229,6 +1229,16 @@ pub(crate) fn try_handle_reset( } conn.subscription_count = 0; + // MONITOR — detach from the command feed. RESET is contracted to return + // the connection to its normal state, and a connection still receiving + // feed lines after RESET would have `+…` lines injected into the reply + // stream of a client that believes it is issuing ordinary commands. + if conn.monitor_attached { + crate::monitor::detach(client_id); + conn.monitor_attached = false; + conn.monitor_rx = None; + } + // Identity + protocol, from the one definition of "default". let (proto, db, authed, user, name) = crate::server::conn::util::restore_migrated_state(None, requirepass); diff --git a/tests/monitor_command_feed.rs b/tests/monitor_command_feed.rs new file mode 100644 index 00000000..3a462a1d --- /dev/null +++ b/tests/monitor_command_feed.rs @@ -0,0 +1,1076 @@ +//! `MONITOR` streams every executed command to attached admin clients. +//! +//! Measured against redis-server 8.6.1 over raw sockets (three probes, +//! 2026-08-14). Raw sockets because the feed IS a stream of `+SimpleString` +//! lines whose exact bytes are the contract — every client library reformats +//! them into a struct before a test could see the difference, and the +//! divergences that matter here are byte-level: which byte leads the frame, +//! how a `0x00` is escaped, whether a secret survives redaction. +//! +//! Two measured facts invert the obvious implementation, and each has a test +//! that fails loudly if someone "fixes" it back: +//! +//! * The feed is a **SimpleString even under RESP3**, not a Push frame. The +//! instinct straight after `pubsub-resp3-push` is to make it a Push; Redis +//! does not, and a client reading the feed expects `+`. See `mon2`. +//! * Administrative commands are hidden at **subcommand** granularity, and +//! the rule is NOT Moon's `CommandFlags::ADMIN | SKIP_MONITOR`. Moon's +//! ADMIN is container-granular (it would hide `INFO`, `CLIENT GETNAME`, +//! `ACL WHOAMI`, which Redis shows) and Redis feeds the whole EVAL family +//! despite flagging it `skip_monitor`. See `mon20`, which drives the +//! measured table row by row. + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +struct Moon { + child: Child, + port: u16, + tmp_dir: std::path::PathBuf, +} + +impl Drop for Moon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_dir_all(&self.tmp_dir); + } +} + +fn spawn_moon(shards: &str) -> Moon { + spawn_moon_opts(shards, &[]) +} + +fn spawn_moon_opts(shards: &str, extra: &[&str]) -> Moon { + // CARGO_BIN_EXE_moon is the binary cargo built for THIS invocation — fresh + // and feature-matched. Never probe target/release directly: that path's + // provenance is unknown and has produced false PASSes before. + let bin = std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")); + let (child, port) = common::spawn_listening(|port| { + let tmp_dir = std::env::temp_dir().join(format!("moon-monitor-{port}")); + let _ = std::fs::create_dir_all(&tmp_dir); + Command::new(&bin) + .args(extra) + .args([ + "--port", + &port.to_string(), + "--shards", + shards, + "--admin-port", + "0", + "--appendonly", + "no", + // This host hovers near the 5% diskfull line; the guard would + // turn writes into MOONERR and flake the suite. + "--disk-free-min-pct", + "0", + "--dir", + tmp_dir.to_str().unwrap(), + ]) + .stdout(Stdio::null()) + .stderr( + std::fs::File::create(tmp_dir.join("moon.stderr")).expect("create moon stderr log"), + ) + .spawn() + .expect("spawn moon") + }); + let tmp_dir = std::env::temp_dir().join(format!("moon-monitor-{port}")); + let mut moon = Moon { + child, + port, + tmp_dir, + }; + let deadline = Instant::now() + Duration::from_secs(30); + while Instant::now() < deadline { + if let Ok(mut c) = TcpStream::connect(("127.0.0.1", moon.port)) { + let _ = c.set_read_timeout(Some(Duration::from_millis(500))); + if c.write_all(b"*1\r\n$4\r\nPING\r\n").is_ok() { + let mut buf = [0u8; 64]; + if let Ok(n) = c.read(&mut buf) + && n > 0 + // `-NOAUTH` is a READY server that wants a password: the + // probe must not treat a password-protected instance as + // one that never came up. + && (buf.starts_with(b"+PONG") || buf.starts_with(b"-NOAUTH")) + { + return moon; + } + } + } + std::thread::sleep(Duration::from_millis(100)); + } + // Never skip. A silent early return would let this suite report green + // while exercising no server at all. + let status = match moon.child.try_wait() { + Ok(Some(s)) => format!("exited with {s}"), + Ok(None) => "still running but never answered PING".to_string(), + Err(e) => format!("status unavailable: {e}"), + }; + let log = std::fs::read_to_string(moon.tmp_dir.join("moon.stderr")).unwrap_or_default(); + panic!("moon never became ready on port {port} ({status})\n--- stderr ---\n{log}"); +} + +struct Conn(TcpStream); + +impl Conn { + fn open(port: u16) -> Self { + let s = TcpStream::connect(("127.0.0.1", port)).expect("connect"); + s.set_read_timeout(Some(Duration::from_millis(600))) + .expect("read timeout"); + s.set_write_timeout(Some(Duration::from_secs(5))) + .expect("write timeout"); + Conn(s) + } + + fn hello3(port: u16) -> Self { + let mut c = Self::open(port); + let r = c.send(&["HELLO", "3"]); + assert!( + !r.is_empty() && r[0] != b'-', + "HELLO 3 must be accepted; got {}", + s(&r) + ); + c + } + + fn write_cmd(&mut self, parts: &[&[u8]]) { + let mut out = format!("*{}\r\n", parts.len()).into_bytes(); + for p in parts { + out.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + out.extend_from_slice(p); + out.extend_from_slice(b"\r\n"); + } + self.0.write_all(&out).expect("write command"); + } + + fn send(&mut self, parts: &[&str]) -> Vec { + let owned: Vec<&[u8]> = parts.iter().map(|p| p.as_bytes()).collect(); + self.write_cmd(&owned); + self.drain() + } + + /// Send a command whose arguments may contain arbitrary bytes. + fn send_bytes(&mut self, parts: &[&[u8]]) -> Vec { + self.write_cmd(parts); + self.drain() + } + + /// Read until the socket goes quiet for one timeout window. + fn drain(&mut self) -> Vec { + let mut got = Vec::new(); + let mut buf = [0u8; 8192]; + loop { + match self.0.read(&mut buf) { + Ok(0) => break, + Ok(n) => got.extend_from_slice(&buf[..n]), + Err(_) => break, + } + } + got + } + + /// Drain with a longer settle, for a feed that may lag the command. + fn feed(&mut self) -> Vec { + std::thread::sleep(Duration::from_millis(120)); + self.drain() + } +} + +fn s(b: &[u8]) -> String { + String::from_utf8_lossy(b).to_string() +} + +/// Attach a MONITOR connection and assert it was accepted. +fn attach(port: u16) -> Conn { + let mut m = Conn::open(port); + let r = m.send(&["MONITOR"]); + assert_eq!( + s(&r), + "+OK\r\n", + "MONITOR must be accepted for an admin connection; got {:?}", + s(&r) + ); + m +} + +/// Every feed line, split on CRLF, with the leading '+' retained. +fn lines(buf: &[u8]) -> Vec { + s(buf) + .split("\r\n") + .filter(|l| !l.is_empty()) + .map(|l| l.to_string()) + .collect() +} + +/// Does any feed line name this command as its first quoted token? +fn names(buf: &[u8], cmd: &str) -> bool { + lines(buf).iter().any(|l| l.contains(&format!("\"{cmd}\""))) +} + +// ── M1 M2 M4 ──────────────────────────────────────────────────────────────── + +#[test] +fn mon1_monitor_replies_ok_and_attaches() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + + let mut c = Conn::open(m.port); + c.send(&["SET", "k", "v"]); + + let got = mon.feed(); + let ls = lines(&got); + assert_eq!( + ls.len(), + 1, + "exactly one feed line for one command; got {ls:?}" + ); + let l = &ls[0]; + assert!( + l.starts_with('+'), + "a feed line is a SimpleString; got {l:?}" + ); + // +. [ :] "SET" "k" "v" + let body = &l[1..]; + let (ts, rest) = body.split_once(' ').expect("timestamp then space"); + let (secs, micros) = ts.split_once('.').expect("secs.micros"); + assert!( + secs.len() >= 10 && secs.chars().all(|c| c.is_ascii_digit()), + "unix seconds; got {secs:?}" + ); + assert_eq!( + micros.len(), + 6, + "micros zero-padded to exactly 6 digits; got {micros:?} in {l:?}" + ); + assert!( + micros.chars().all(|c| c.is_ascii_digit()), + "micros all digits; got {micros:?}" + ); + assert!( + rest.starts_with("[0 127.0.0.1:"), + "db 0 and the peer address in brackets; got {rest:?}" + ); + assert!( + rest.ends_with(r#""SET" "k" "v""#), + "every token quoted, command name included; got {rest:?}" + ); +} + +#[test] +fn mon2_feed_is_simplestring_under_resp3() { + // The trap this test exists for: pubsub-resp3-push just made confirmations + // Push frames, so the reflex is to do the same here. Redis does NOT — the + // feed stays a SimpleString in RESP3, and a client reading it expects '+'. + let m = spawn_moon("1"); + let mut mon = Conn::hello3(m.port); + let r = mon.send(&["MONITOR"]); + assert_eq!(s(&r), "+OK\r\n", "MONITOR answers +OK in RESP3 too"); + + let mut c = Conn::open(m.port); + c.send(&["SET", "k", "v"]); + + let got = mon.feed(); + assert!(!got.is_empty(), "the RESP3 monitor must receive the feed"); + assert_eq!( + got[0], + b'+', + "the feed line is a SimpleString under RESP3, not a Push ('>'); got {:?}", + s(&got) + ); +} + +#[test] +fn mon3_reads_are_fed() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + c.send(&["SET", "k", "v"]); + let mut mon = attach(m.port); + c.send(&["GET", "k"]); + let got = mon.feed(); + assert!( + names(&got, "GET"), + "reads are fed, not only writes; got {:?}", + s(&got) + ); +} + +#[test] +fn mon4_db_follows_select() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + c.send(&["SELECT", "3"]); + c.send(&["GET", "k"]); + let got = mon.feed(); + let ls = lines(&got); + assert!( + ls.iter().all(|l| l.contains("[3 ")), + "both lines report the db AFTER SELECT; got {ls:?}" + ); +} + +// ── M6: administrative commands ───────────────────────────────────────────── + +#[test] +fn mon5_admin_commands_are_not_fed() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + c.send(&["CONFIG", "SET", "maxmemory", "0"]); + c.send(&["DBSIZE"]); + + let got = mon.feed(); + assert!( + !names(&got, "CONFIG"), + "CONFIG is administrative and must never reach a monitor; got {:?}", + s(&got) + ); + // The half that proves the feed was live rather than simply broken. + assert!( + names(&got, "DBSIZE"), + "DBSIZE is not administrative and must be fed — without this the test \ + would pass against a feed that emits nothing at all; got {:?}", + s(&got) + ); +} + +#[test] +fn mon6_monitor_is_absent_from_its_own_feed() { + let m = spawn_moon("1"); + let mut mon_a = attach(m.port); + let mut mon_b = attach(m.port); + + let a = mon_a.feed(); + assert!( + !names(&a, "MONITOR"), + "MONITOR is administrative, so attaching a second monitor emits nothing; got {:?}", + s(&a) + ); + + let mut c = Conn::open(m.port); + c.send(&["SET", "k", "v"]); + assert!( + names(&mon_a.feed(), "SET"), + "and both monitors are still live" + ); + assert!(names(&mon_b.feed(), "SET"), "including the second one"); +} + +#[test] +fn mon7_rejected_commands_are_not_fed() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + c.send(&["NOSUCHCMD", "x"]); + c.send(&["GET"]); // arity violation + c.send(&["PING"]); + + let got = mon.feed(); + assert!( + !names(&got, "NOSUCHCMD"), + "an unknown command never executes, so it is never fed; got {:?}", + s(&got) + ); + assert!( + !names(&got, "GET"), + "an arity-rejected command never executes either; got {:?}", + s(&got) + ); + assert!( + names(&got, "PING"), + "the following valid command IS fed, proving the feed is live; got {:?}", + s(&got) + ); +} + +// ── M8: redaction ─────────────────────────────────────────────────────────── + +#[test] +fn mon8_auth_arguments_are_redacted() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + c.send(&["AUTH", "hunter2"]); + c.send(&["AUTH", "someuser", "s3cret-pw"]); + + let got = mon.feed(); + let text = s(&got); + // The assertion that matters is the ABSENCE of the secret, not the + // presence of the placeholder: a formatter that appended "(redacted)" + // after emitting the password would satisfy the weaker check. + assert!( + !text.contains("hunter2"), + "the single-argument AUTH password must not appear anywhere in the feed; got {text:?}" + ); + assert!( + !text.contains("s3cret-pw"), + "the two-argument AUTH password must not appear anywhere in the feed; got {text:?}" + ); + assert!( + text.contains(r#""AUTH" "(redacted)""#), + "AUTH pw renders as \"AUTH\" \"(redacted)\"; got {text:?}" + ); + assert!( + text.contains(r#""AUTH" "(redacted)" "(redacted)""#), + "AUTH user pw redacts BOTH arguments — the username is a credential too; got {text:?}" + ); +} + +#[test] +fn mon9_hello_auth_redacts_only_credentials() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + c.send(&["HELLO", "3", "AUTH", "default", "sekrit"]); + + let text = s(&mon.feed()); + assert!( + !text.contains("sekrit"), + "the HELLO AUTH password must not appear in the feed; got {text:?}" + ); + assert!( + text.contains(r#""HELLO" "3" "AUTH" "(redacted)" "(redacted)""#), + "the protocol version survives; only the two arguments after AUTH are \ + redacted; got {text:?}" + ); +} + +// ── M9: transactions ──────────────────────────────────────────────────────── + +#[test] +fn mon10_transaction_timing() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + + c.send(&["MULTI"]); + let at_multi = mon.feed(); + assert!( + names(&at_multi, "MULTI"), + "MULTI is fed when it is issued; got {:?}", + s(&at_multi) + ); + + let q = c.send(&["SET", "q", "1"]); + assert_eq!(s(&q), "+QUEUED\r\n"); + let at_queue = mon.feed(); + assert!( + !names(&at_queue, "SET"), + "a QUEUED command has not executed, so it must not be fed yet — this \ + is the half a naive implementation gets wrong; got {:?}", + s(&at_queue) + ); + + c.send(&["EXEC"]); + let at_exec = mon.feed(); + let ls = lines(&at_exec); + let set_at = ls.iter().position(|l| l.contains("\"SET\"")); + let exec_at = ls.iter().position(|l| l.contains("\"EXEC\"")); + assert!( + set_at.is_some() && exec_at.is_some(), + "at EXEC both the queued command and EXEC are fed; got {ls:?}" + ); + assert!( + set_at < exec_at, + "the queued command is fed BEFORE the EXEC line; got {ls:?}" + ); +} + +#[test] +fn mon11_two_monitors_both_receive() { + let m = spawn_moon("1"); + let mut a = attach(m.port); + let mut b = attach(m.port); + let mut c = Conn::open(m.port); + c.send(&["SET", "dual", "1"]); + + let la = lines(&a.feed()); + let lb = lines(&b.feed()); + assert!( + la.iter().any(|l| l.contains(r#""SET" "dual" "1""#)), + "monitor A receives; got {la:?}" + ); + assert!( + lb.iter().any(|l| l.contains(r#""SET" "dual" "1""#)), + "monitor B receives the same line; got {lb:?}" + ); +} + +// ── M2: escaping ──────────────────────────────────────────────────────────── + +#[test] +fn mon12_argument_escaping_is_byte_exact() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + + c.send_bytes(&[b"SET", b"esc", b"a\"b\\c\nd\re\tf\x00g\xffh"]); + c.send_bytes(&[b"SET", b"utf", "h\u{e9}llo".as_bytes()]); + c.send_bytes(&[b"SET", b"empty", b""]); + + let text = s(&mon.feed()); + assert!( + text.contains(r#""a\"b\\c\nd\re\tf\x00g\xffh""#), + "quote, backslash, newline, CR, tab, NUL and a high byte each escape \ + per Redis sdscatrepr semantics; got {text:?}" + ); + assert!( + text.contains(r#""h\xc3\xa9llo""#), + "UTF-8 escapes PER BYTE, not per character; got {text:?}" + ); + assert!( + text.contains(r#""SET" "empty" """#), + "an empty argument renders as a pair of quotes; got {text:?}" + ); +} + +// ── M13: every dispatch path ──────────────────────────────────────────────── + +#[test] +fn mon13_inline_fast_path_is_fed() { + // A plain GET/SET takes try_inline_dispatch under monoio, which is a + // different code path from everything else here. A feed hook missing there + // is invisible to any test that uses another command — and GET/SET are + // what most tests use. This is the shape of the v0.8.6 inline-GET P0. + let m = spawn_moon("4"); + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + c.send(&["SET", "inline", "1"]); + c.send(&["GET", "inline"]); + + let got = mon.feed(); + assert!( + names(&got, "SET"), + "the inline SET must be fed at --shards 4; got {:?}", + s(&got) + ); + assert!( + names(&got, "GET"), + "and the inline GET, which is the exact path the v0.8.6 P0 slipped \ + through; got {:?}", + s(&got) + ); +} + +// ── Rejections ────────────────────────────────────────────────────────────── + +#[test] +fn mon14_non_admin_cannot_attach() { + let m = spawn_moon("1"); + let mut admin = Conn::open(m.port); + let r = admin.send(&[ + "ACL", "SETUSER", "lowly", "on", ">pw", "~*", "+@all", "-@admin", + ]); + assert_eq!(s(&r), "+OK\r\n", "test fixture: create a non-admin user"); + + let mut lo = Conn::open(m.port); + assert_eq!(s(&lo.send(&["AUTH", "lowly", "pw"])), "+OK\r\n"); + let denied = lo.send(&["MONITOR"]); + assert!( + denied.starts_with(b"-NOPERM"), + "MONITOR requires the admin category; got {:?}", + s(&denied) + ); + + // The security-relevant half: refusing the reply is worthless if the + // connection was attached anyway. + let mut c = Conn::open(m.port); + c.send(&["SET", "secret", "value"]); + let leaked = lo.feed(); + assert!( + leaked.is_empty(), + "a refused MONITOR must not be attached — this connection received \ + another client's traffic: {:?}", + s(&leaked) + ); +} + +#[test] +fn mon15_monitor_rejects_arguments() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let r = c.send(&["MONITOR", "extra"]); + assert_eq!( + s(&r), + "-ERR wrong number of arguments for 'monitor' command\r\n", + "MONITOR has arity 1" + ); + let mut other = Conn::open(m.port); + other.send(&["SET", "k", "v"]); + assert!( + c.feed().is_empty(), + "and the connection was not attached by the failed call" + ); +} + +#[test] +fn mon16_monitor_conn_cannot_touch_keyspace() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let r = mon.send(&["SET", "x", "1"]); + assert_eq!( + s(&r), + "-ERR Replica can't interact with the keyspace\r\n", + "verbatim Redis text for a keyspace command on a monitor connection" + ); + + let mut c = Conn::open(m.port); + c.send(&["SET", "still", "flowing"]); + assert!( + names(&mon.feed(), "SET"), + "and the refusal does not break the feed" + ); +} + +#[test] +fn mon17_second_monitor_is_silent() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + let again = mon.send(&["MONITOR"]); + assert!( + again.is_empty(), + "MONITOR on an already-attached connection is answered with NOTHING — \ + measured; Redis does not error here. Got {:?}", + s(&again) + ); + + let mut c = Conn::open(m.port); + c.send(&["SET", "once", "1"]); + let ls = lines(&mon.feed()); + assert_eq!( + ls.len(), + 1, + "and the connection is attached exactly ONCE — a second registration \ + would duplicate every line; got {ls:?}" + ); +} + +#[test] +fn mon18_reset_detaches() { + let m = spawn_moon("1"); + let mut mon = attach(m.port); + assert_eq!(s(&mon.send(&["RESET"])), "+RESET\r\n"); + + let mut c = Conn::open(m.port); + c.send(&["SET", "postreset", "1"]); + assert!(mon.feed().is_empty(), "RESET detaches: the feed stops"); + assert_eq!( + s(&mon.send(&["GET", "postreset"])), + "$1\r\n1\r\n", + "and keyspace access is restored" + ); +} + +#[test] +fn mon19_command_info_monitor() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let r = c.send(&["COMMAND", "INFO", "monitor"]); + let text = s(&r); + assert!( + text.contains("monitor"), + "MONITOR must be registered so COMMAND and ACL can see it; got {text:?}" + ); + assert!( + !text.contains("*-1") && !text.starts_with("*1\r\n*-1"), + "and answer a real spec rather than a Null element; got {text:?}" + ); + assert!( + text.contains("admin"), + "carrying the admin category, or any user could read every other \ + user's traffic; got {text:?}" + ); +} + +// ── M6b: the audit table, driven row by row ───────────────────────────────── + +#[test] +fn mon20_admin_audit_table() { + // The regression guard for the §0 audit. Moon's own CommandFlags::ADMIN is + // CONTAINER-granular and cannot express this table: flagging on it would + // hide INFO, CLIENT GETNAME, ACL WHOAMI and CLUSTER INFO, all of which + // Redis feeds. Every row here was measured against redis-server 8.6.1. + let m = spawn_moon("1"); + + // (command, must_be_fed) + let cases: &[(&[&str], bool)] = &[ + // Fed — not administrative in Redis. + (&["INFO", "server"], true), + (&["DBSIZE"], true), + (&["LASTSAVE"], true), + (&["COMMAND", "COUNT"], true), + (&["CLIENT", "GETNAME"], true), + (&["CLIENT", "ID"], true), + (&["ACL", "WHOAMI"], true), + (&["ACL", "CAT"], true), + (&["CLUSTER", "INFO"], true), + (&["MEMORY", "USAGE", "nokey"], true), + // Fed despite Redis's own `skip_monitor` flag — measured directly. + // Skipping the EVAL family would hide exactly what an operator most + // wants to see. + (&["EVAL", "return 1", "0"], true), + // Hidden — administrative. + (&["CONFIG", "GET", "maxmemory"], false), + (&["CONFIG", "SET", "maxmemory", "0"], false), + (&["CLIENT", "LIST"], false), + (&["ACL", "LIST"], false), + (&["SLOWLOG", "LEN"], false), + (&["SLOWLOG", "RESET"], false), + (&["LATENCY", "RESET"], false), + ]; + + let mut failures: Vec = Vec::new(); + for (cmd, want_fed) in cases { + let mut mon = attach(m.port); + let mut c = Conn::open(m.port); + c.send(cmd); + let got = mon.feed(); + let fed = names(&got, cmd[0]); + if fed != *want_fed { + failures.push(format!( + " {:30} expected {:6} got {:6}", + cmd.join(" "), + if *want_fed { "fed" } else { "hidden" }, + if fed { "fed" } else { "hidden" } + )); + } + } + assert!( + failures.is_empty(), + "monitor visibility diverges from redis-server 8.6.1 on {} row(s).\n\ + A row that flipped to `fed` is a LEAK of administrative arguments; a \ + row that flipped to `hidden` silently under-reports the feed.\n{}", + failures.len(), + failures.join("\n") + ); +} + +// ── The contracted backpressure policy ────────────────────────────────────── + +#[test] +fn mon21_slow_monitor_is_dropped_not_stalled() { + // Contracted at freeze: a monitor that cannot keep up has its CONNECTION + // DROPPED, loudly. Not silent line-dropping (an operator cannot tell a + // quiet server from a lossy feed) and never blocking (one slow TCP reader + // must not stall every shard). + let m = spawn_moon("1"); + let mon = attach(m.port); + // Deliberately never read from `mon` again. + + let mut c = Conn::open(m.port); + let start = Instant::now(); + for i in 0..20_000 { + c.write_cmd(&[b"SET", b"burst", i.to_string().as_bytes()]); + } + let _ = c.drain(); + let elapsed = start.elapsed(); + + assert!( + elapsed < Duration::from_secs(20), + "a monitor that stopped reading must never stall the publishing \ + connection; the burst took {elapsed:?}" + ); + + // And the publisher is still healthy. + let mut probe = Conn::open(m.port); + assert_eq!( + s(&probe.send(&["PING"])), + "+PONG\r\n", + "the server is still serving after a slow monitor was shed" + ); + + // The monitor connection itself was closed rather than left half-alive. + // Assert on END OF STREAM, not on an empty drain: `after.is_empty()` is + // true whenever no line happened to be buffered, so an earlier version of + // this assertion was satisfied by a STARVED BUT OPEN connection — exactly + // the failure mode the policy exists to rule out. A closed socket answers + // Ok(0) (or errors); a starved-but-open one blocks until the timeout. + let mut mon = mon; + let _ = mon.drain(); + mon.0 + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("read timeout"); + let _ = mon.0.write_all(b"*1\r\n$4\r\nPING\r\n"); + let mut probe_buf = [0u8; 64]; + let eof = matches!(mon.0.read(&mut probe_buf), Ok(0) | Err(_)); + assert!( + eof, + "the slow monitor's connection must be CLOSED, not silently starved \ + of lines — a lossy feed an operator cannot detect is the failure mode \ + this policy exists to avoid" + ); +} + +// ── contract: is the literal `lua` for script-issued commands ──────── + +#[test] +fn mon22_script_issued_commands_are_fed_with_the_lua_address() { + // Measured against redis-server 8.6.1 (2026-08-14): + // …[0 127.0.0.1:51772] "eval" "redis.call('SET', …)" "1" "lk" + // …[0 lua] "SET" "lk" "v" + // …[0 lua] "GET" "lk" + // The EVAL line carries the client's address; each command the SCRIPT + // issues carries the literal `lua` instead, in execution order, after it. + // + // This is the one contract clause with no connection behind it — a script + // command never passes a connection handler, so the handler-level hooks + // cannot see it. An implementation that feeds only what clients send looks + // completely correct until an operator watches a script-driven workload and + // sees the EVAL but none of its effects. + let m = spawn_moon("1"); + let mut mon = attach(m.port); + + let mut c = Conn::open(m.port); + let r = c.send(&[ + "EVAL", + "redis.call('SET', KEYS[1], 'v') return redis.call('GET', KEYS[1])", + "1", + "lk", + ]); + assert!( + s(&r).contains('v'), + "the script itself must run; got {:?}", + s(&r) + ); + + let feed = mon.feed(); + let ls = lines(&feed); + + let eval_at = ls + .iter() + .position(|l| l.contains("\"EVAL\"") || l.contains("\"eval\"")) + .unwrap_or_else(|| panic!("the EVAL itself must be fed; got {ls:#?}")); + let set_at = ls + .iter() + .position(|l| l.contains("\"SET\"") && l.contains("\"lk\"")) + .unwrap_or_else(|| panic!("the script's SET must be fed; got {ls:#?}")); + let get_at = ls + .iter() + .position(|l| l.contains("\"GET\"") && l.contains("\"lk\"")) + .unwrap_or_else(|| panic!("the script's GET must be fed; got {ls:#?}")); + + assert!( + eval_at < set_at && set_at < get_at, + "script effects follow the EVAL, in execution order; got {ls:#?}" + ); + assert!( + ls[set_at].contains("[0 lua] "), + "a script-issued command carries the literal `lua` address, not a \ + peer address: {:?}", + ls[set_at] + ); + assert!( + ls[get_at].contains("[0 lua] "), + "reads issued by a script are fed the same way: {:?}", + ls[get_at] + ); +} + +// ── the refusal rule: write|readonly, NOT `first_key != 0` ────────────────── + +#[test] +fn mon23_refusal_rule_is_write_or_readonly_not_first_key() { + // Re-measured against redis-server 8.6.1 (2026-08-14, one fresh connection + // per probe — a shared socket desynchronises against the interleaved feed + // and produced two wrong readings on the first pass). + // + // The §3 contract said "keyspace command", and the first implementation + // read that as `first_key != 0`. That is measurably wrong in BOTH + // directions' worth of rows: `DBSIZE`, `KEYS`, `SCAN`, `RANDOMKEY`, + // `FLUSHALL`, `FLUSHDB`, `SWAPDB`, `EVAL` and `PUBLISH` all carry + // `first_key == 0` and are all REFUSED by Redis. The rule that actually + // reproduces every measured row is the WRITE-or-READONLY flag pair, plus + // the script/publish family, which Redis refuses without either flag. + let m = spawn_moon("1"); + + for probe in [ + &["DBSIZE"][..], + &["KEYS", "*"], + &["SCAN", "0"], + &["RANDOMKEY"], + &["FLUSHALL"], + &["FLUSHDB"], + &["SWAPDB", "0", "1"], + &["EVAL", "return 1", "0"], + &["PUBLISH", "c", "m"], + &["TYPE", "k"], + &["EXISTS", "k"], + &["TTL", "k"], + &["MEMORY", "USAGE", "k"], + &["GET", "k"], + &["SET", "k", "v"], + &["EXPIRE", "k", "1"], + ] { + let mut mon = attach(m.port); + let r = mon.send(probe); + assert_eq!( + s(&r), + "-ERR Replica can't interact with the keyspace\r\n", + "{:?} is refused on a monitor connection (measured)", + probe + ); + } + + for probe in [ + &["PING"][..], + &["INFO", "server"], + &["CLIENT", "ID"], + &["ACL", "WHOAMI"], + &["COMMAND", "COUNT"], + &["LASTSAVE"], + &["TIME"], + &["ECHO", "x"], + &["SELECT", "1"], + &["WAIT", "0", "0"], + &["SCRIPT", "LOAD", "return 1"], + &["MEMORY", "DOCTOR"], + &["BGSAVE"], + ] { + let mut mon = attach(m.port); + let r = mon.send(probe); + assert!( + !s(&r).starts_with("-ERR Replica can't interact"), + "{:?} is SERVED on a monitor connection (measured); got {:?}", + probe, + s(&r) + ); + } +} + +// ── the first AUTH of a session — the one that carries the password ───────── + +#[test] +fn mon24_first_auth_of_a_session_is_fed_and_redacted() { + // Both handlers gate on `!conn.authenticated` ABOVE the ACL-exempt AUTH / + // HELLO intercepts, and `continue` out of it. The feed hook sits below that + // gate, so on a password-protected server the FIRST AUTH — the only one + // that actually carries a credential — never reached the feed at all. + // + // `mon8` and `mon9` did not catch this because they run against a server + // with no password: `conn.authenticated` is already true there, so their + // AUTH falls through to the intercept the hook does cover. A redaction test + // that never exercises an authenticating connection tests the wrong path. + let m = spawn_moon_opts("1", &["--requirepass", "s3kr1t"]); + + let mut mon = Conn::open(m.port); + assert_eq!(s(&mon.send(&["AUTH", "s3kr1t"])), "+OK\r\n"); + assert_eq!( + s(&mon.send(&["MONITOR"])), + "+OK\r\n", + "the default user must be able to attach after AUTH" + ); + + // A fresh connection performing its own first AUTH. + let mut c = Conn::open(m.port); + assert_eq!(s(&c.send(&["AUTH", "s3kr1t"])), "+OK\r\n"); + + let feed = mon.feed(); + let text = s(&feed); + assert!( + names(&feed, "AUTH"), + "the first AUTH of a session must be fed — it is the one command that \ + carries a credential, so its absence is the least acceptable gap in \ + the feed. Got {text:?}" + ); + assert!( + !text.contains("s3kr1t"), + "and it must be redacted: the password appears in the feed. Got {text:?}" + ); + assert!(text.contains("\"AUTH\" \"(redacted)\""), "got {text:?}"); +} + +#[test] +fn mon25_first_hello_auth_of_a_session_is_fed_and_redacted() { + // The HELLO half of the same gate. + let m = spawn_moon_opts("1", &["--requirepass", "s3kr1t"]); + + let mut mon = Conn::open(m.port); + assert_eq!(s(&mon.send(&["AUTH", "s3kr1t"])), "+OK\r\n"); + assert_eq!(s(&mon.send(&["MONITOR"])), "+OK\r\n"); + + let mut c = Conn::open(m.port); + let r = c.send(&["HELLO", "3", "AUTH", "default", "s3kr1t"]); + assert!(!r.is_empty() && r[0] != b'-', "HELLO AUTH must succeed"); + + let feed = mon.feed(); + let text = s(&feed); + assert!(names(&feed, "HELLO"), "got {text:?}"); + assert!( + !text.contains("s3kr1t"), + "the password must not survive: {text:?}" + ); + assert!( + text.contains("\"HELLO\" \"3\" \"AUTH\" \"(redacted)\" \"(redacted)\""), + "the version survives; only the credentials go. Got {text:?}" + ); +} + +// ── ACL: +@all must grant MONITOR ─────────────────────────────────────────── + +#[test] +fn mon26_plus_at_all_grants_monitor() { + // `mon14` proves `-@admin` REFUSES MONITOR, which passes just as well when + // no grant reaches MONITOR at all. Without this positive case, a MONITOR + // that is ungrantable by any category looks correct. + let m = spawn_moon("1"); + let mut admin = Conn::open(m.port); + assert_eq!( + s(&admin.send(&["ACL", "SETUSER", "opsy", "on", ">pw", "~*", "+@all"])), + "+OK\r\n", + "test fixture: create a +@all user" + ); + + let mut ops = Conn::open(m.port); + assert_eq!(s(&ops.send(&["AUTH", "opsy", "pw"])), "+OK\r\n"); + assert_eq!( + s(&ops.send(&["MONITOR"])), + "+OK\r\n", + "+@all must grant MONITOR — a category expansion that omits it makes \ + the command unreachable by any grant" + ); + + let mut c = Conn::open(m.port); + c.send(&["SET", "k", "v"]); + assert!( + names(&ops.feed(), "SET"), + "and the attach is real, not just an accepted reply" + ); +} + +// ── teardown: a monitor must never be silently carried across migration ───── + +#[test] +fn mon27_monitor_connection_is_not_migration_eligible() { + // Connection migration returns from the handler through its own path, + // BEFORE the disconnect detach block runs. A monitor carried through it + // would leave its sink registered under the same client_id while the new + // handler starts unattached and never detaches — the registry then holds a + // dead sink forever, which also pins `any_attached()` true and holds the + // inline fast path down for the life of the process. + // + // Asserted from the outside, through the only observable the contract + // gives: RESET detaches, and after RESET a new MONITOR must be answered + // with `+OK` rather than the silence that means "already attached". + let m = spawn_moon("4"); + let mut mon = attach(m.port); + assert_eq!(s(&mon.send(&["RESET"])), "+RESET\r\n"); + assert_eq!( + s(&mon.send(&["MONITOR"])), + "+OK\r\n", + "after RESET the registry must no longer know this connection; \ + silence here means a stale registration survived" + ); + + // And a stale registration under a reused id must never silently swallow + // the attach: the connection is either attached with a live sink or told so. + let mut c = Conn::open(m.port); + c.send(&["SET", "k", "v"]); + assert!( + names(&mon.feed(), "SET"), + "the re-attached monitor receives a live feed, not a dead sink" + ); +}