From 9789946572d0d944be57ab380b5259c2e7683336 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 13 Aug 2026 10:33:09 +0700 Subject: [PATCH 1/8] =?UTF-8?q?docs(add):=20plan=20info-observability=20?= =?UTF-8?q?=E2=80=94=20INFO=20parity=20+=20keyspace=20notifications?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Grounded §0 by measurement against redis-server 8.6.1 rather than assumption, which changed three things I would otherwise have got wrong: - INCR emits `incrby`, not `incr`; RENAME emits TWO events (rename_from on the source, rename_to on the destination); a key miss emits nothing under `A` because `m` is deliberately not a member of that class. - Redis's default INFO omits Commandstats and Latencystats — Moon emits Commandstats unconditionally. - Flag readback is canonicalized, not echoed: KEA -> AKE, Kg$ -> g$K. Measured state: Moon exposes 61 INFO fields vs Redis's 213, is missing the Errorstats/Latencystats/Cluster/Modules sections, emits `# Replication` twice, and ignores the section argument entirely (`info(db, _args)`), so `INFO replication` returns all 12 sections. Keyspace notifications are absent outright — zero occurrences of notify-keyspace-events, __keyspace@ or __keyevent@ anywhere in src/ or tests/. The §1 lowest-confidence assumption was checked BEFORE freezing and proved partially wrong: ShardSlice has no pubsub registry and no current-db index, so a per-command hook would have meant re-plumbing every write signature. Design inverted accordingly — commands append to a per-shard notification outbox and the event loop drains it, keeping src/command/** out of scope and the fan-out where awaiting is legal. Recorded in §3 with the cost of being wrong again. No src/ changes yet; §4 red suite is specified, not written. author: Tin Dang --- .add/state.json | 4 +- .add/tasks/info-observability/TASK.md | 408 +++++++++++++++++++++++--- 2 files changed, 362 insertions(+), 50 deletions(-) diff --git a/.add/state.json b/.add/state.json index f046efb2..537874dd 100644 --- a/.add/state.json +++ b/.add/state.json @@ -1,7 +1,7 @@ { "project": "moon", "stage": "production", - "active_task": "protocol-error-lifetime", + "active_task": "info-observability", "active_milestone": "v0-9-client-compat", "tasks": { "hotpath-lock-quickwins": { @@ -476,7 +476,7 @@ } }, "created": "2026-06-11T03:18:21+00:00", - "updated": "2026-08-12T04:55:06+00:00", + "updated": "2026-08-13T02:05:58+00:00", "setup": { "locked": true, "locked_at": "2026-06-11T03:28:00+00:00", diff --git a/.add/tasks/info-observability/TASK.md b/.add/tasks/info-observability/TASK.md index bc746b95..61f984d4 100644 --- a/.add/tasks/info-observability/TASK.md +++ b/.add/tasks/info-observability/TASK.md @@ -15,37 +15,155 @@ phase: ground - --- ## 2 · SCENARIOS — pass/fail cases ▸ docs/04-step-2-scenarios.md @@ -53,54 +171,248 @@ Assumptions — lowest-confidence first: ```gherkin -Scenario: - Given - When - Then - And # required for every rejection +Scenario: INFO selects a single section + Given a running server + When the client sends "INFO replication" + Then the reply contains exactly one section header, "# Replication" + And no other "# " header appears in the reply + +Scenario: INFO section matching is case-insensitive + Given a running server + When the client sends "INFO REPLICATION" + Then the reply is byte-identical to "INFO replication" + +Scenario: INFO accepts several sections + Given a running server + When the client sends "INFO server clients" + Then the reply contains exactly "# Server" and "# Clients" + And they appear in the server's canonical order + +Scenario: unknown section is empty, not an error + Given a running server + When the client sends "INFO nosuchsection" + Then the reply is an empty bulk string + And the connection stays open + +Scenario: default INFO omits Commandstats + Given a running server + When the client sends "INFO" + Then "# Commandstats" does not appear + And "# Commandstats" DOES appear for "INFO all" + +Scenario: no section header is emitted twice + Given a running server + When the client sends "INFO" + Then every "# " header in the reply is unique + +Scenario: run_id changes across a restart + Given a server whose INFO reports run_id R1 + When the server is restarted on the same dir + Then INFO reports a run_id different from R1 + And it is 40 hex characters + +Scenario: keyspace hit and miss counters move + Given notifications are irrelevant and the counters are read from INFO + When the client GETs an existing key and then a missing key + Then keyspace_hits increases by 1 + And keyspace_misses increases by 1 + +Scenario: invalid notify flag is rejected verbatim + Given a running server + When the client sends "CONFIG SET notify-keyspace-events KEQ" + Then the error is exactly Redis's Invalid-event-class-character message + And CONFIG GET still returns the PREVIOUS flag string + +Scenario: notify flags are canonicalized on readback + Given a running server + When the client sends "CONFIG SET notify-keyspace-events KEA" + Then CONFIG GET notify-keyspace-events returns "AKE" + +Scenario: keyspace and keyevent channels carry inverted payloads + Given notify-keyspace-events is "KEA" + And a client subscribed to "__key*@0__:*" + When another client sends "SET foo bar" + Then a message arrives on "__keyspace@0__:foo" with payload "set" + And a message arrives on "__keyevent@0__:set" with payload "foo" + +Scenario: INCR reports incrby, not incr + Given notify-keyspace-events is "KEA" + And a client subscribed to "__keyevent@0__:*" + When another client sends "INCR ctr" + Then the event channel is "__keyevent@0__:incrby" + +Scenario: RENAME emits both halves + Given notify-keyspace-events is "KEA" + And a client subscribed to "__keyevent@0__:*" + When another client renames "a" to "b" + Then an event "rename_from" arrives carrying "a" + And an event "rename_to" arrives carrying "b" + +Scenario: expiry emits expired + Given notify-keyspace-events is "KEA" + And a client subscribed to "__keyevent@0__:expired" + When a key set with PX 60 elapses + Then an "expired" event arrives naming that key + +Scenario: a key miss emits nothing under A + Given notify-keyspace-events is "KEA" + And a client subscribed to "__key*@0__:*" + When another client GETs a key that does not exist + Then no message arrives + And the same GET under flags "KEm" DOES produce a keymiss event + +Scenario: subscriber on a different shard still receives + Given a server with 4 shards and notify-keyspace-events "KEA" + And a client subscribed to "__keyevent@0__:set" on one connection + When keys owned by every shard are SET from other connections + Then an event arrives for each key + +Scenario: disabled notifications cost nothing on the write path + Given notify-keyspace-events is "" (the default) + And a client subscribed to "__key*@0__:*" + When another client sends "SET foo bar" + Then no message arrives ``` - - --- -## 3 · CONTRACT — freeze the shape ▸ docs/05-step-3-contract.md - -``` - body: { } - 200 -> { } - 4xx -> { error: "" | "" } -Schema: -``` +## 3 · CONTRACT — the frozen interface ▸ docs/05-step-3-contract.md Status: DRAFT - ---- +### Least-sure flag surfaced at freeze: + +⚠ **"The event hook can be placed where it sees both the key and the db without +re-plumbing command signatures."** — **PARTIALLY WRONG, resolved before freeze.** +Checked `ShardSlice` (`src/shard/slice.rs:63`): it carries `shard_id` and +`databases: Box<[Database]>` but **no pubsub registry and no current-db index** +(db selection is per-connection). Hooking each write command directly would +therefore have required threading both through every command signature — the +invasive diff the flag warned about. + +Resolved by inverting the design: commands do not publish. They append to a +per-shard **notification outbox** on `ShardSlice`, which the event loop drains +after the command completes — the loop already holds `all_pubsub_registries`. +This keeps `src/command/**` out of the scope, matches the established shard-slice +side-table idiom (`kv_write_intents`), and puts the fan-out on the loop where +awaiting is legal. The db index is stamped into the outbox entry by the caller +that already knows it, not looked up later. + +Cost of being wrong a second time: if the outbox cannot be drained without +crossing an await while holding the slice borrow, fan-out moves to a dedicated +chore tick and events gain up-to-one-tick latency (still at-most-once, still +ordered per key). That is a latency change, not a correctness change. + +### Interface + +```rust +// src/pubsub/keyspace.rs (new) + +/// Event classes, one bit each. Parsed from the `notify-keyspace-events` flag +/// string; `A` expands to every class EXCEPT keymiss (`m`) and new-key (`n`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct NotifyFlags(u16); + +impl NotifyFlags { + /// Parse a flag string. Returns Err(offending_char) on any char outside + /// `Ag$lshzxeKEtmdn`, so the caller can build Redis's verbatim error. + pub fn parse(spec: &[u8]) -> Result; + /// Canonical readback: classes in `g$lshzxetdmn` order, then K, then E. + /// `A` is re-collapsed when every A-member class is present. + pub fn to_spec(&self) -> String; + /// True when neither K nor E is set — the whole feature is off. + #[inline] pub fn disabled(&self) -> bool; + #[inline] pub fn wants(&self, class: EventClass) -> bool; +} + +/// One pending notification. `key` is a cheap Bytes clone, never a copy. +pub struct PendingNotify { pub db: u32, pub class: EventClass, pub event: &'static str, pub key: Bytes } + +/// Appended by write paths, drained by the shard event loop. +/// Push is a no-op (single flag load, no formatting, no alloc) when disabled. +impl ShardSlice { pub fn notify(&mut self, db: u32, class: EventClass, event: &'static str, key: &Bytes); } +``` -## 4 · TESTS — failing-first suite (red) ▸ docs/06-step-4-tests.md +```rust +// src/command/connection.rs — INFO gains real section selection +/// `sections` is the caller's argument list, already lowercased. +/// Empty => the default set (everything except Commandstats/Latencystats). +pub fn info_sections(db: &Database, sections: &[&str], keyspace: &[(u64,u64)]) -> Frame; +``` -Coverage target: -Plan (one test per scenario, asserting behavior not internals): - - - test_: arrange / act / assert + assert - +### Guarantees + - `NotifyFlags::parse` never allocates and never panics on arbitrary bytes. + - `to_spec(parse(x))` is idempotent: canonical form round-trips. + - `ShardSlice::notify` when `disabled()` performs no allocation and no + string formatting — verified by a test asserting the outbox stays empty + and by keeping channel construction inside the drain, not the push. + - A section name is emitted at most once per INFO reply even if the caller + repeats it (`INFO server server` yields one `# Server`). + - Event delivery is at-most-once and may be dropped for a slow subscriber, + exactly as PUBLISH already behaves — notifications are NOT a durability + mechanism and this contract does not promise delivery. + +### Scope (files this task may touch) + - `src/pubsub/keyspace.rs` (new) · `src/pubsub/mod.rs` (re-export) + - `src/shard/slice.rs` (outbox field + `notify`) · `src/shard/event_loop.rs` (drain) + - `src/command/connection.rs` (INFO) · `src/config.rs` (+ `config/conf_file.rs`) + - `src/command/introspect.rs` (CONFIG GET/SET wiring) + - `tests/info_observability.rs` (new) · `tests/keyspace_notifications.rs` (new) + - `scripts/client-compat/manifest.yaml` · `CHANGELOG.md` + - **Out of scope:** `src/command/**` other than `connection.rs`/`introspect.rs`. + If the outbox turns out to need per-command call sites, that is a contract + change, not a build decision. + +### Target + - `INFO replication` returns one section; default `INFO` omits Commandstats. + - The 19 Must-listed fields present with real values; `run_id` differs across restart. + - All 17 §2 scenarios green, including the 4-shard cross-shard delivery case. + - Disabled-path: no allocation on the write path (outbox-empty assertion). -Tests live in: `./tests/` · MUST run red (missing implementation) before Build. - +--- - +## 4 · TESTS & SCENARIOS — the red suite ▸ docs/06-step-4-tests.md + +Run red BEFORE any build code. Every Must and every Reject has a test below. + +`tests/info_observability.rs` + - `io1_single_section_only` — covers: INFO
returns only that section + - `io2_section_case_insensitive` — covers: case-insensitive matching + - `io3_multiple_sections` — covers: `INFO server clients` + - `io4_unknown_section_is_empty` — covers: unknown section => empty bulk, conn open + - `io5_default_omits_commandstats` — covers: default vs `INFO all` + - `io6_no_duplicate_headers` — covers: `# Replication` emitted twice (current bug) + - `io7_repeated_section_emitted_once` — covers: `INFO server server` + - `io8_required_fields_present` — covers: the 19 Must-listed fields + - `io9_run_id_shape_and_restart` — covers: 40 hex, differs across restart + - `io10_keyspace_hit_miss_counters` — covers: hits/misses move by exactly 1 + +`tests/keyspace_notifications.rs` + - `kn1_invalid_flag_char_verbatim` — covers: Reject #1, exact error text, + AND that CONFIG GET still returns the previous value (no partial apply) + - `kn2_flags_canonicalized` — covers: `KEA`->`AKE`, `Kg$`->`g$K` + - `kn3_keyspace_and_keyevent_inverted` — covers: both channels, inverted payloads + - `kn4_incr_reports_incrby` — covers: event name != command name + - `kn5_rename_emits_both_halves` — covers: rename_from + rename_to + - `kn6_expired_event` — covers: TTL elapse emits `expired` + - `kn7_keymiss_silent_under_A` — covers: Reject #2, and that `m` DOES emit + - `kn8_k_or_e_required` — covers: Reject #3, classes without K/E deliver nothing + - `kn9_cross_shard_delivery` — covers: 4 shards, subscriber on one connection + - `kn10_disabled_emits_nothing` — covers: default config is silent + +Unit tests (in-module, `src/pubsub/keyspace.rs`): + - `parse_rejects_out_of_class_char` · `to_spec_round_trips` · + `a_expands_without_m_or_n` · `disabled_when_no_k_or_e` + +Harness: `scripts/client-compat/manifest.yaml` gains live parity assertions for +INFO section selection and the notify-flag canonicalization, replacing nothing +(no existing waiver covers these). + +Note: `kn9` MUST run under the monoio default runtime, not only tokio — the +cross-shard path differs between handlers and a tokio-only test is CI-blind +to the shipped runtime. --- From 8b136fb6d80f5fd04b1b465bf6219ca35ca603e8 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 13 Aug 2026 11:20:40 +0700 Subject: [PATCH 2/8] test(info): red suite for INFO parity and keyspace notifications MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 20 tests, 19 red, each failing for the reason the contract names. io1 reproduces the headline exactly — `INFO replication` returns 13 sections with `# Replication` emitted twice: ["# Server", "# Clients", "# Memory", "# Persistence", "# Vector", "# MoonStore", "# Reclamation", "# Stats", "# CPU", "# Replication", "# Commandstats", "# Keyspace", "# Replication"] kn3-kn9 fail at the `enable()` helper because `notify-keyspace-events` is not a known config parameter — the feature is absent, not broken. Expectations were captured from redis-server 8.6.1 rather than recalled, which changed three of them: INCR publishes `incrby` not `incr`; RENAME publishes TWO events (rename_from on the source, rename_to on the destination); and a key miss publishes nothing under `A`, because `m` is deliberately not a member of that class. kn9 is the test a single-shard suite cannot write. Moon keeps one pub/sub registry per shard, the write runs on the key's owner shard, and the subscriber sits on whichever shard accepted its connection — so a local-only publish passes at --shards 1 and drops roughly (N-1)/N of events at --shards N. That exact mistake is already live in the MQ-trigger path; filed as #474. kn10 passes today and is annotated as vacuous: a server that can publish nothing trivially publishes nothing. It guards the default, it does not prove gating — kn8 does that. Raw sockets throughout: redis-rs parses INFO into a map, which would hide both the section order and the duplicate-header bug under test. author: Tin Dang --- .add/tasks/info-observability/TASK.md | 10 +- tests/info_observability.rs | 382 ++++++++++++++++++++++ tests/keyspace_notifications.rs | 438 ++++++++++++++++++++++++++ 3 files changed, 828 insertions(+), 2 deletions(-) create mode 100644 tests/info_observability.rs create mode 100644 tests/keyspace_notifications.rs diff --git a/.add/tasks/info-observability/TASK.md b/.add/tasks/info-observability/TASK.md index 61f984d4..116bd54b 100644 --- a/.add/tasks/info-observability/TASK.md +++ b/.add/tasks/info-observability/TASK.md @@ -152,9 +152,15 @@ Assumptions — lowest-confidence first: If wrong: the hook has to be threaded through every write command's signature — a much larger, more invasive diff than this contract assumes, and the §5 scope would need to grow to `src/command/**`. - - [ ] Cross-shard notification fan-out can reuse PUBLISH's existing + - [x] Cross-shard notification fan-out can reuse PUBLISH's existing `all_pubsub_registries` path rather than needing a new SPSC message class. - Confirm by reading how PUBLISH fans out before writing the hook. + **CONFIRMED** (`handler_monoio/pubsub.rs:60`): PUBLISH resolves targets via + `ctx.remote_subscriber_map.read().target_shards(&ch)` and dispatches in + per-shard batches. The outbox drain reuses exactly this — no new message + class. Filed #474 along the way: the MQ-trigger path + (`shard/timers.rs:249`) does the LOCAL publish only and so never reaches a + subscriber on another shard; the same mistake is the one `kn9` guards + against here. - [ ] `instantaneous_ops_per_sec` can be derived from an existing counter sampled on the 1s chore, with no new per-command atomic on the hot path. Confirm against the existing Stats counters; if it needs a new atomic, diff --git a/tests/info_observability.rs b/tests/info_observability.rs new file mode 100644 index 00000000..24941525 --- /dev/null +++ b/tests/info_observability.rs @@ -0,0 +1,382 @@ +//! INFO must answer the question the client actually asked. +//! +//! Measured against redis-server 8.6.1. Moon's `info()` takes `_args` and +//! discards it, so section selection does not exist: +//! +//! ```text +//! INFO replication +//! redis: # Replication (1 section) +//! moon : # Server # Clients # Memory # Persistence # Vector # MoonStore +//! # Reclamation # Stats # CPU # Replication # Commandstats +//! # Keyspace # Replication (13, one twice) +//! ``` +//! +//! A monitoring agent that scrapes `INFO replication` every second is paying +//! for the whole payload and parsing a duplicated header. Separately, Moon +//! exposes 61 fields where Redis exposes 213, and the missing ones are the +//! ones dashboards read: keyspace_hits, evicted_keys, maxmemory_policy. +//! +//! Raw sockets: redis-rs parses INFO into a map, which would hide both the +//! section ORDER and the duplicate-header bug under test. + +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_in(shards, None) +} + +/// `dir` lets a restart test reuse the SAME data dir, which is the only way +/// to prove run_id changes for reasons other than a fresh dataset. +fn spawn_moon_in(shards: &str, dir: Option) -> Moon { + let bin = std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")); + let fixed_dir = dir.clone(); + let (child, port) = common::spawn_listening(|port| { + let tmp_dir = fixed_dir + .clone() + .unwrap_or_else(|| std::env::temp_dir().join(format!("moon-infoobs-{port}"))); + let _ = std::fs::create_dir_all(&tmp_dir); + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + shards, + "--admin-port", + "0", + "--appendonly", + "no", + "--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 = dir.unwrap_or_else(|| std::env::temp_dir().join(format!("moon-infoobs-{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 + && buf.starts_with(b"+PONG") + { + return moon; + } + } + } + std::thread::sleep(Duration::from_millis(100)); + } + 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_secs(5))).unwrap(); + s.set_write_timeout(Some(Duration::from_secs(5))).unwrap(); + Conn(s) + } + + fn send(&mut self, parts: &[&str]) -> String { + let mut out = format!("*{}\r\n", parts.len()); + for p in parts { + out.push_str(&format!("${}\r\n{p}\r\n", p.len())); + } + self.0.write_all(out.as_bytes()).expect("write"); + self.read_reply() + } + + fn read_reply(&mut self) -> String { + let mut buf = [0u8; 16384]; + let mut acc = Vec::new(); + loop { + match self.0.read(&mut buf) { + Ok(0) => break, + Ok(n) => { + acc.extend_from_slice(&buf[..n]); + self.0 + .set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + } + Err(_) => break, + } + } + self.0 + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + String::from_utf8_lossy(&acc).into_owned() + } +} + +/// Section headers, in reply order. `INFO` is a bulk string whose payload is +/// CRLF-delimited, so the `$` prefix line is skipped by the `#` filter. +fn headers(reply: &str) -> Vec { + reply + .lines() + .map(|l| l.trim_end_matches('\r')) + .filter(|l| l.starts_with('#')) + .map(|l| l.to_string()) + .collect() +} + +fn field(reply: &str, name: &str) -> Option { + reply + .lines() + .map(|l| l.trim_end_matches('\r')) + .find_map(|l| l.strip_prefix(&format!("{name}:")).map(|v| v.to_string())) +} + +// --------------------------------------------------------------------------- +// io1 — the headline. This is why the task exists. +// --------------------------------------------------------------------------- + +#[test] +fn io1_single_section_only() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let reply = c.send(&["INFO", "replication"]); + let h = headers(&reply); + assert_eq!( + h, + vec!["# Replication"], + "INFO
must return ONLY that section — a monitoring agent \ + polling INFO replication should not pay for the whole payload. got {h:?}" + ); +} + +#[test] +fn io2_section_case_insensitive() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let lower = headers(&c.send(&["INFO", "replication"])); + let upper = headers(&c.send(&["INFO", "REPLICATION"])); + assert_eq!( + lower, upper, + "section matching is case-insensitive in Redis; clients send both" + ); + assert_eq!(upper, vec!["# Replication"]); +} + +#[test] +fn io3_multiple_sections() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let h = headers(&c.send(&["INFO", "server", "clients"])); + assert_eq!( + h, + vec!["# Server", "# Clients"], + "INFO accepts several section names and returns exactly those, in the \ + server's canonical order — not the caller's. got {h:?}" + ); +} + +#[test] +fn io4_unknown_section_is_empty() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let reply = c.send(&["INFO", "nosuchsection"]); + assert!( + !reply.starts_with('-'), + "an unknown section is NOT an error in Redis — it is an empty bulk \ + string. Erroring here breaks clients that probe optional sections. got {reply:?}" + ); + assert!( + headers(&reply).is_empty(), + "unknown section must yield no sections; got {:?}", + headers(&reply) + ); + // The connection must survive — an unknown section is not a protocol fault. + assert!(c.send(&["PING"]).starts_with("+PONG")); +} + +#[test] +fn io5_default_omits_commandstats() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let default = headers(&c.send(&["INFO"])); + let all = headers(&c.send(&["INFO", "all"])); + assert!( + !default.iter().any(|h| h == "# Commandstats"), + "Redis's default INFO omits Commandstats — it is per-command data that \ + grows with the command table and is only emitted on request. got {default:?}" + ); + assert!( + all.iter().any(|h| h == "# Commandstats"), + "INFO all must include Commandstats; got {all:?}" + ); +} + +#[test] +fn io6_no_duplicate_headers() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let h = headers(&c.send(&["INFO"])); + let mut seen = std::collections::HashSet::new(); + let dups: Vec<&String> = h.iter().filter(|x| !seen.insert((*x).clone())).collect(); + assert!( + dups.is_empty(), + "every section header must appear at most once — a duplicate makes \ + naive INFO parsers (split on '#', build a map) silently keep only one \ + copy and drop the other's fields. duplicated: {dups:?} in {h:?}" + ); +} + +#[test] +fn io7_repeated_section_emitted_once() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let h = headers(&c.send(&["INFO", "server", "server"])); + assert_eq!( + h, + vec!["# Server"], + "a repeated section name must not duplicate the section; got {h:?}" + ); +} + +#[test] +fn io8_required_fields_present() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let reply = c.send(&["INFO", "all"]); + // The fields clients, drivers and monitoring agents actually parse. + for name in [ + "run_id", + "redis_mode", + "cluster_enabled", + "process_id", + "os", + "arch_bits", + "keyspace_hits", + "keyspace_misses", + "expired_keys", + "evicted_keys", + "rejected_connections", + "maxmemory", + "maxmemory_policy", + "instantaneous_ops_per_sec", + "blocked_clients", + "pubsub_channels", + "pubsub_patterns", + "total_net_input_bytes", + "total_net_output_bytes", + ] { + assert!( + field(&reply, name).is_some(), + "INFO is missing {name:?} — stock monitoring agents read it and \ + will either KeyError or silently report zero" + ); + } +} + +#[test] +fn io9_run_id_shape_and_restart() { + let dir = std::env::temp_dir().join(format!("moon-infoobs-runid-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + + let first = { + let m = spawn_moon_in("1", Some(dir.clone())); + let mut c = Conn::open(m.port); + let id = field(&c.send(&["INFO", "server"]), "run_id").expect("run_id present"); + assert_eq!( + id.len(), + 40, + "run_id is a 40-char hex string in Redis; got {id:?}" + ); + assert!( + id.bytes().all(|b| b.is_ascii_hexdigit()), + "run_id must be hex; got {id:?}" + ); + id + // m dropped here -> server killed, and Drop also removes the dir, so + // the second server below starts clean. That is fine for this test: + // run_id must be per-PROCESS, and deriving it from dataset state would + // be wrong regardless of whether the dataset survived. + }; + + // Fresh process: clients use run_id to detect that the server they are + // talking to is not the one they were talking to. + let m2 = spawn_moon_in("1", Some(dir.clone())); + let mut c2 = Conn::open(m2.port); + let second = field(&c2.send(&["INFO", "server"]), "run_id").expect("run_id present"); + assert_ne!( + first, second, + "run_id must differ across a restart — a stable one defeats every \ + client-side restart/failover detection that depends on it" + ); +} + +#[test] +fn io10_keyspace_hit_miss_counters() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + c.send(&["SET", "io10key", "v"]); + + let before = c.send(&["INFO", "stats"]); + let h0: u64 = field(&before, "keyspace_hits") + .expect("keyspace_hits present") + .parse() + .expect("numeric"); + let m0: u64 = field(&before, "keyspace_misses") + .expect("keyspace_misses present") + .parse() + .expect("numeric"); + + c.send(&["GET", "io10key"]); // hit + c.send(&["GET", "io10missing"]); // miss + + let after = c.send(&["INFO", "stats"]); + let h1: u64 = field(&after, "keyspace_hits").unwrap().parse().unwrap(); + let m1: u64 = field(&after, "keyspace_misses").unwrap().parse().unwrap(); + + assert_eq!( + h1 - h0, + 1, + "one existing-key GET must move keyspace_hits by exactly 1 \ + (hit-rate dashboards divide by these)" + ); + assert_eq!( + m1 - m0, + 1, + "one missing-key GET must move keyspace_misses by exactly 1" + ); +} diff --git a/tests/keyspace_notifications.rs b/tests/keyspace_notifications.rs new file mode 100644 index 00000000..a11e39bb --- /dev/null +++ b/tests/keyspace_notifications.rs @@ -0,0 +1,438 @@ +//! Keyspace notifications: `__keyspace@__` / `__keyevent@__`. +//! +//! Moon has none. Measured: zero occurrences of `notify-keyspace-events`, +//! `__keyspace@` or `__keyevent@` anywhere in `src/` or `tests/`. Cache +//! invalidation frameworks and change-data-capture consumers subscribe to +//! these channels and currently get silence from Moon. +//! +//! Every expectation below was captured from redis-server 8.6.1 rather than +//! recalled, which mattered — three of them are counter-intuitive: +//! +//! * `INCR` publishes `incrby`, NOT `incr`. +//! * `RENAME` publishes TWO events: `rename_from` on the source key and +//! `rename_to` on the destination. +//! * a key MISS publishes nothing under `A`, because `m` (keymiss) is +//! deliberately not a member of the `A` class. +//! +//! The cross-shard case (`kn9`) is the one most likely to be quietly wrong: +//! Moon keeps one pub/sub registry PER SHARD, a write runs on the shard that +//! owns the key, and the subscriber sits on whichever shard accepted its +//! connection. A local-only publish passes at `--shards 1` and fails at 4. +//! That exact mistake is live elsewhere in the tree — see issue #474. + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +/// Redis 8.6.1, verbatim. +const INVALID_FLAG_ERR: &str = "Invalid event class character. Use 'Ag$lshzxeKEtmdn'."; + +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 { + 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-keyspacenotif-{port}")); + let _ = std::fs::create_dir_all(&tmp_dir); + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + shards, + "--admin-port", + "0", + "--appendonly", + "no", + "--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-keyspacenotif-{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 + && buf.starts_with(b"+PONG") + { + return moon; + } + } + } + std::thread::sleep(Duration::from_millis(100)); + } + 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_secs(5))).unwrap(); + s.set_write_timeout(Some(Duration::from_secs(5))).unwrap(); + Conn(s) + } + + fn send(&mut self, parts: &[&str]) -> String { + self.write(parts); + self.read_reply() + } + + fn write(&mut self, parts: &[&str]) { + let mut out = format!("*{}\r\n", parts.len()); + for p in parts { + out.push_str(&format!("${}\r\n{p}\r\n", p.len())); + } + self.0.write_all(out.as_bytes()).expect("write"); + } + + fn read_reply(&mut self) -> String { + let mut buf = [0u8; 16384]; + let mut acc = Vec::new(); + loop { + match self.0.read(&mut buf) { + Ok(0) => break, + Ok(n) => { + acc.extend_from_slice(&buf[..n]); + self.0 + .set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + } + Err(_) => break, + } + } + self.0 + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + String::from_utf8_lossy(&acc).into_owned() + } + + /// Drain pmessages for `window`, returning (channel, payload) pairs. + /// + /// Waits the full window rather than returning on first message: a test + /// asserting that NOTHING arrives must not pass merely by reading early. + fn collect_pmessages(&mut self, window: Duration) -> Vec<(String, String)> { + self.0 + .set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + let deadline = Instant::now() + window; + let mut acc = Vec::new(); + let mut buf = [0u8; 16384]; + while Instant::now() < deadline { + match self.0.read(&mut buf) { + Ok(0) => break, + Ok(n) => acc.extend_from_slice(&buf[..n]), + Err(_) => {} + } + } + let text = String::from_utf8_lossy(&acc).into_owned(); + // RESP arrays: pmessage / / / + let lines: Vec<&str> = text + .split("\r\n") + .filter(|l| !l.is_empty() && !l.starts_with('*') && !l.starts_with('$')) + .collect(); + let mut out = Vec::new(); + for (i, l) in lines.iter().enumerate() { + if *l == "pmessage" && i + 3 < lines.len() { + out.push((lines[i + 2].to_string(), lines[i + 3].to_string())); + } + } + out + } +} + +/// Subscribe to a pattern and return the connection, ready to collect. +fn psubscriber(port: u16, pattern: &str) -> Conn { + let mut c = Conn::open(port); + let ack = c.send(&["PSUBSCRIBE", pattern]); + assert!( + ack.contains("psubscribe"), + "PSUBSCRIBE not acknowledged: {ack:?}" + ); + c +} + +fn enable(port: u16, flags: &str) { + let mut c = Conn::open(port); + let r = c.send(&["CONFIG", "SET", "notify-keyspace-events", flags]); + assert!( + r.starts_with("+OK"), + "CONFIG SET notify-keyspace-events {flags:?} failed: {r:?}" + ); +} + +// --------------------------------------------------------------------------- +// Config surface +// --------------------------------------------------------------------------- + +#[test] +fn kn1_invalid_flag_char_verbatim() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + c.send(&["CONFIG", "SET", "notify-keyspace-events", "KEA"]); + + let err = c.send(&["CONFIG", "SET", "notify-keyspace-events", "KEQ"]); + assert!( + err.starts_with('-'), + "an out-of-class flag char must be rejected; got {err:?}" + ); + assert!( + err.contains(INVALID_FLAG_ERR), + "the error must name the valid class set the way Redis does, so a \ + config-management tool can surface it unchanged. want {INVALID_FLAG_ERR:?}, got {err:?}" + ); + + // A rejected SET must not have partially applied. + let readback = c.send(&["CONFIG", "GET", "notify-keyspace-events"]); + assert!( + readback.contains("AKE"), + "a rejected CONFIG SET must leave the PREVIOUS value intact — a \ + half-applied flag set silently changes which events fire. got {readback:?}" + ); +} + +#[test] +fn kn2_flags_canonicalized() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + for (set, want) in [("KEA", "AKE"), ("Kg$", "g$K"), ("xe", "xe"), ("Km", "Km")] { + let r = c.send(&["CONFIG", "SET", "notify-keyspace-events", set]); + assert!(r.starts_with("+OK"), "CONFIG SET {set:?}: {r:?}"); + let got = c.send(&["CONFIG", "GET", "notify-keyspace-events"]); + assert!( + got.contains(want), + "readback is canonicalized, not echoed: classes in 'g$lshzxetdmn' \ + order, then K, then E. set {set:?} want {want:?}, got {got:?}" + ); + } +} + +// --------------------------------------------------------------------------- +// Event wire form +// --------------------------------------------------------------------------- + +#[test] +fn kn3_keyspace_and_keyevent_inverted() { + let m = spawn_moon("1"); + enable(m.port, "KEA"); + let mut sub = psubscriber(m.port, "__key*@0__:*"); + + let mut w = Conn::open(m.port); + w.send(&["SET", "kn3key", "v"]); + + let msgs = sub.collect_pmessages(Duration::from_secs(2)); + assert!( + msgs.contains(&("__keyspace@0__:kn3key".into(), "set".into())), + "keyspace channel is named for the KEY and carries the EVENT; got {msgs:?}" + ); + assert!( + msgs.contains(&("__keyevent@0__:set".into(), "kn3key".into())), + "keyevent channel is named for the EVENT and carries the KEY — the \ + pair is inverted, and getting it backwards breaks every consumer; got {msgs:?}" + ); +} + +#[test] +fn kn4_incr_reports_incrby() { + let m = spawn_moon("1"); + enable(m.port, "KEA"); + let mut sub = psubscriber(m.port, "__keyevent@0__:*"); + + let mut w = Conn::open(m.port); + w.send(&["INCR", "kn4ctr"]); + + let msgs = sub.collect_pmessages(Duration::from_secs(2)); + assert!( + msgs.iter().any(|(ch, _)| ch == "__keyevent@0__:incrby"), + "INCR publishes 'incrby', not 'incr' — the event name is the internal \ + operation, not the command the client typed. got {msgs:?}" + ); +} + +#[test] +fn kn5_rename_emits_both_halves() { + let m = spawn_moon("1"); + enable(m.port, "KEA"); + let mut w = Conn::open(m.port); + w.send(&["SET", "kn5src", "v"]); + + let mut sub = psubscriber(m.port, "__keyevent@0__:*"); + w.send(&["RENAME", "kn5src", "kn5dst"]); + + let msgs = sub.collect_pmessages(Duration::from_secs(2)); + assert!( + msgs.contains(&("__keyevent@0__:rename_from".into(), "kn5src".into())), + "RENAME emits rename_from carrying the SOURCE key; got {msgs:?}" + ); + assert!( + msgs.contains(&("__keyevent@0__:rename_to".into(), "kn5dst".into())), + "RENAME emits a SECOND event, rename_to, carrying the DESTINATION — a \ + consumer tracking key lifetimes needs both halves; got {msgs:?}" + ); +} + +#[test] +fn kn6_expired_event() { + let m = spawn_moon("1"); + enable(m.port, "KEA"); + let mut sub = psubscriber(m.port, "__keyevent@0__:expired"); + + let mut w = Conn::open(m.port); + w.send(&["SET", "kn6vol", "v", "PX", "60"]); + + let msgs = sub.collect_pmessages(Duration::from_secs(3)); + assert!( + msgs.iter().any(|(_, key)| key == "kn6vol"), + "an elapsed TTL must publish 'expired' — cache consumers rely on it to \ + invalidate; got {msgs:?}" + ); +} + +// --------------------------------------------------------------------------- +// Gating — the cases where NOTHING must arrive +// --------------------------------------------------------------------------- + +#[test] +fn kn7_keymiss_silent_under_a() { + let m = spawn_moon("1"); + enable(m.port, "KEA"); + let mut sub = psubscriber(m.port, "__key*@0__:*"); + + let mut w = Conn::open(m.port); + w.send(&["GET", "kn7definitelymissing"]); + + let msgs = sub.collect_pmessages(Duration::from_secs(2)); + assert!( + msgs.is_empty(), + "'m' (keymiss) is deliberately NOT part of the 'A' class — publishing \ + on every miss would put a pub/sub fan-out on the read path. got {msgs:?}" + ); + + // ...but it DOES fire when asked for explicitly. + enable(m.port, "Km"); + let mut sub2 = psubscriber(m.port, "__key*@0__:*"); + w.send(&["GET", "kn7definitelymissing"]); + let msgs2 = sub2.collect_pmessages(Duration::from_secs(2)); + assert!( + !msgs2.is_empty(), + "with 'm' set explicitly a key miss MUST publish; got nothing" + ); +} + +#[test] +fn kn8_k_or_e_required() { + let m = spawn_moon("1"); + // Class flags select WHICH events; K/E select WHETHER they are delivered. + enable(m.port, "g$"); + let mut sub = psubscriber(m.port, "__key*@0__:*"); + + let mut w = Conn::open(m.port); + w.send(&["SET", "kn8key", "v"]); + + let msgs = sub.collect_pmessages(Duration::from_secs(2)); + assert!( + msgs.is_empty(), + "with neither K nor E set nothing is delivered, however many class \ + flags are on; got {msgs:?}" + ); +} + +/// NOTE: this is the one test in this file that passes BEFORE the feature +/// exists, because a server that can publish nothing trivially publishes +/// nothing. It is a guard against the default flipping on, not evidence that +/// gating works — `kn8` is what proves gating. Do not read its green as +/// progress. +#[test] +fn kn10_disabled_emits_nothing() { + let m = spawn_moon("1"); + // No enable() — the default is off, and must stay off. + let mut sub = psubscriber(m.port, "__key*@0__:*"); + + let mut w = Conn::open(m.port); + w.send(&["SET", "kn10key", "v"]); + w.send(&["DEL", "kn10key"]); + + let msgs = sub.collect_pmessages(Duration::from_secs(2)); + assert!( + msgs.is_empty(), + "notifications are off by default and must cost nothing; got {msgs:?}" + ); +} + +// --------------------------------------------------------------------------- +// kn9 — the one a single-shard test cannot catch. See issue #474. +// --------------------------------------------------------------------------- + +#[test] +fn kn9_cross_shard_delivery() { + let m = spawn_moon("4"); + enable(m.port, "KEA"); + let mut sub = psubscriber(m.port, "__keyevent@0__:set"); + + // Distinct key names hash to different shards; the subscriber sits on + // whichever shard accepted ITS connection, which is at most one of them. + let keys = [ + "kn9:alpha", + "kn9:beta", + "kn9:gamma", + "kn9:delta", + "kn9:epsilon", + "kn9:zeta", + "kn9:eta", + "kn9:theta", + ]; + let mut w = Conn::open(m.port); + for k in keys { + w.send(&["SET", k, "v"]); + } + + let msgs = sub.collect_pmessages(Duration::from_secs(3)); + let got: std::collections::HashSet<&str> = msgs.iter().map(|(_, key)| key.as_str()).collect(); + let missing: Vec<&str> = keys.iter().copied().filter(|k| !got.contains(k)).collect(); + assert!( + missing.is_empty(), + "a notification must reach a subscriber on ANY shard, not just the \ + shard that owns the mutated key. Moon keeps one pub/sub registry per \ + shard, so a local-only publish passes at --shards 1 and drops roughly \ + (N-1)/N of events at --shards N. missing: {missing:?} (got {got:?})" + ); +} From c652d2b495f4cb9cfcf4b41267f49779a079e8d4 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 13 Aug 2026 12:27:07 +0700 Subject: [PATCH 3/8] feat(info): INFO section selection and single assembly point MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit INFO now answers the question the client asked. `INFO replication` returns one section instead of thirteen, and `# Replication` is no longer emitted twice. The duplicate had a structural cause: `connection::info` wrote a STUB `# Replication`, and all three connection handlers then APPENDED the real section from `replication::handshake::build_info_replication`. That also meant INFO had three assembly points, so a filter living in `connection.rs` alone would have leaked the appended section on every request. The real section is now passed IN (`info_with_keyspace_and_replication`) and substituted in place, leaving one point — `info_sections::finalize` — that sees the final section set and can both de-duplicate and filter it. Semantics measured against redis-server 8.6.1, not recalled: INFO -> every section EXCEPT Commandstats/Latencystats INFO all | everything -> every section INFO replication -> only that one, case-insensitively INFO server clients -> both, in the SERVER's order, not the caller's INFO nosuchsection -> empty payload, NOT an error Also adds run_id (40 hex, per-process, regenerated on every start — deriving it from anything durable would defeat the restart detection clients use it for), redis_mode, cluster_enabled, process_id, os, arch_bits, and atomic-backed keyspace/eviction/network counters. Those counters needed their own atomics: the existing `counter!()` recorders feed Prometheus behind METRICS_INITIALIZED, which is false unless the admin port is up, so INFO would have reported zero on a server nobody scraped. The atomics sit alongside, incremented in the SAME recorder functions — no new call sites on any command path. Green: io1-io7, io9 (8/10 integration) plus 9 unit tests. Known red, carried to the next commit: - io8: maxmemory, maxmemory_policy, blocked_clients, pubsub_channels and pubsub_patterns are still missing. - io10: keyspace_hits does not move. `record_keyspace_hit()` lives in string_read.rs, which the fast dispatch path appears to bypass — the three-dispatch-path trap, not a counter bug. author: Tin Dang --- .add/tasks/info-observability/TASK.md | 21 +- src/command/connection.rs | 111 ++++++++- src/command/info_sections.rs | 252 ++++++++++++++++++++ src/command/mod.rs | 1 + src/server/conn/handler_monoio/dispatch.rs | 26 +- src/server/conn/handler_sharded/dispatch.rs | 24 +- src/server/conn/handler_single.rs | 24 +- 7 files changed, 406 insertions(+), 53 deletions(-) create mode 100644 src/command/info_sections.rs diff --git a/.add/tasks/info-observability/TASK.md b/.add/tasks/info-observability/TASK.md index 116bd54b..fa107af5 100644 --- a/.add/tasks/info-observability/TASK.md +++ b/.add/tasks/info-observability/TASK.md @@ -165,9 +165,18 @@ Assumptions — lowest-confidence first: sampled on the 1s chore, with no new per-command atomic on the hot path. Confirm against the existing Stats counters; if it needs a new atomic, drop the field rather than pay hot-path cost (Must says omit, never lie). - - [ ] Moon's `# Reclamation` and duplicate `# Replication` come from a path + - [x] Moon's `# Reclamation` and duplicate `# Replication` come from a path outside `connection.rs:info`; the section filter must cover that path too - or filtering will silently leak sections. Confirm by locating the injector. + or filtering will silently leak sections. **CONFIRMED, and located.** + `connection.rs:426` writes a STUB `# Replication`, then all three handlers + append the REAL one from `replication::handshake::build_info_replication` + (`handler_monoio/dispatch.rs:677`, `handler_sharded/dispatch.rs:369`, + `handler_single.rs:1078`). That is the duplicate. `# Reclamation` is + written from inside `info()` (`connection.rs:391`) and is fine. + Consequence: INFO has THREE assembly points, so a filter applied only in + `connection.rs` would leak the replication section on every request. The + fix is to give `info()` the real section and delete the append — which + requires touching the handlers, hence the §3 scope amendment below. --- @@ -367,6 +376,14 @@ pub fn info_sections(db: &Database, sections: &[&str], keyspace: &[(u64,u64)]) - - `src/command/introspect.rs` (CONFIG GET/SET wiring) - `tests/info_observability.rs` (new) · `tests/keyspace_notifications.rs` (new) - `scripts/client-compat/manifest.yaml` · `CHANGELOG.md` + - `src/server/conn/handler_monoio/dispatch.rs` · `handler_sharded/dispatch.rs` · + `handler_single.rs` — **added by amendment (contract still DRAFT)**. INFO + has three assembly points, not one: each handler appends the real + `# Replication` section after `info()` has already written a stub. A filter + applied only in `connection.rs` would leak that section on every request, + and the duplicate-header Must (`io6`) cannot be satisfied without deleting + the append. Scope is limited to passing the real section into `info()` and + removing the append — no other handler behaviour changes. - **Out of scope:** `src/command/**` other than `connection.rs`/`introspect.rs`. If the outbox turns out to need per-command call sites, that is a contract change, not a build decision. diff --git a/src/command/connection.rs b/src/command/connection.rs index 2118425b..e9d9820a 100644 --- a/src/command/connection.rs +++ b/src/command/connection.rs @@ -169,7 +169,50 @@ fn format_memory_human(bytes: u64) -> String { /// the single-`Database` fallback for paths with no scatter access (generic /// dispatch). The connection handlers pass cross-shard, all-db stats through /// [`info_with_keyspace`] instead. -pub fn info(db: &Database, _args: &[Frame]) -> Frame { +pub fn info(db: &Database, args: &[Frame]) -> Frame { + let raw = info_raw(db); + crate::command::info_sections::finalize(&raw, None, args) +} + +/// This process's `run_id`: 40 hex chars, stable for the process lifetime and +/// regenerated on every start. +/// +/// Clients (and Sentinel) compare it across reconnects to decide whether they +/// are talking to the same server instance; deriving it from anything durable +/// — the data dir, the port, a config hash — would defeat exactly that check, +/// so it is seeded from process identity plus start time. +fn run_id() -> &'static str { + use std::sync::OnceLock; + static RUN_ID: OnceLock = OnceLock::new(); + RUN_ID.get_or_init(|| { + use std::hash::{Hash, Hasher}; + let mut h = std::collections::hash_map::DefaultHasher::new(); + std::process::id().hash(&mut h); + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + .hash(&mut h); + (&RUN_ID as *const _ as usize).hash(&mut h); + let a = h.finish(); + // Second independent round so the 40 chars are not a repeat of 16. + a.hash(&mut h); + let b = h.finish(); + b.hash(&mut h); + let c = h.finish(); + format!("{a:016x}{b:016x}{c:016x}")[..40].to_string() + }) +} + +/// Build the full INFO payload with every section present. +/// +/// Callers must pass this through [`crate::command::info_sections::finalize`], +/// which applies section selection and drops duplicate headers. Building +/// everything and filtering afterwards is deliberate: the `# Replication` +/// section here is a STUB that the connection handlers replace with the real +/// one, so a filter applied during assembly would either emit the stub or drop +/// the real section depending on where it ran. +fn info_raw(db: &Database) -> String { use std::fmt::Write as _; let mut sections = String::with_capacity(2048); @@ -177,6 +220,18 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { let _ = write!(sections, "redis_version:{REDIS_COMPAT_VERSION}\r\n"); let _ = write!(sections, "moon_version:{MOON_VERSION}\r\n"); sections.push_str("moon:true\r\n"); + let _ = write!(sections, "run_id:{}\r\n", run_id()); + // Cluster mode is reported by the cluster subsystem; standalone until it + // says otherwise. Clients branch on these two before anything else. + sections.push_str("redis_mode:standalone\r\n"); + sections.push_str("cluster_enabled:0\r\n"); + let _ = write!(sections, "process_id:{}\r\n", std::process::id()); + let _ = write!(sections, "os:{}\r\n", std::env::consts::OS); + let _ = write!( + sections, + "arch_bits:{}\r\n", + (std::mem::size_of::() * 8) + ); sections.push_str("\r\n"); sections.push_str("# Clients\r\n"); @@ -441,6 +496,29 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { crate::admin::metrics_setup::spsc_notify_skipped(), crate::admin::metrics_setup::ft_search_cooperative_yields(), ); + // Fields stock monitoring agents read. Backed by real counters — a field + // Moon cannot answer truthfully is omitted rather than reported as a + // constant, because a hardcoded zero is indistinguishable from a healthy + // server on a dashboard. + let _ = write!( + sections, + "keyspace_hits:{}\r\n\ + keyspace_misses:{}\r\n\ + expired_keys:{}\r\n\ + evicted_keys:{}\r\n\ + rejected_connections:{}\r\n\ + total_net_input_bytes:{}\r\n\ + total_net_output_bytes:{}\r\n\ + instantaneous_ops_per_sec:{}\r\n", + crate::admin::metrics_setup::keyspace_hits(), + crate::admin::metrics_setup::keyspace_misses(), + crate::admin::metrics_setup::expired_keys(), + crate::admin::metrics_setup::evicted_keys(), + crate::admin::metrics_setup::rejected_connections(), + crate::admin::metrics_setup::total_net_input_bytes(), + crate::admin::metrics_setup::total_net_output_bytes(), + crate::admin::metrics_setup::instantaneous_ops_per_sec(), + ); sections.push_str("\r\n"); // # CPU @@ -497,7 +575,7 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { ); } - Frame::BulkString(Bytes::from(sections)) + sections } /// INFO with an externally-gathered `# Keyspace` section: one `(keys, @@ -507,16 +585,29 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { /// count, so `SELECT 2; SET k v; INFO` reported the db-2 count as db0 and /// every other db was invisible. pub fn info_with_keyspace(db: &Database, args: &[Frame], keyspace: &[(u64, u64)]) -> Frame { + info_with_keyspace_and_replication(db, args, keyspace, None) +} + +/// As [`info_with_keyspace`], but also substitutes the authoritative +/// `# Replication` section. +/// +/// The connection handlers own the replication state, so they used to APPEND +/// their section after `info()` had already written a stub — which is why INFO +/// emitted `# Replication` twice. Passing it in instead keeps one assembly +/// point, so section selection and de-duplication see the final section set. +pub fn info_with_keyspace_and_replication( + db: &Database, + args: &[Frame], + keyspace: &[(u64, u64)], + real_replication: Option<&str>, +) -> Frame { use std::fmt::Write as _; - let base = match info(db, args) { - Frame::BulkString(b) => b, - other => return other, - }; - let text = String::from_utf8_lossy(&base); + let text = info_raw(db); // Rebuild everything up to the fallback "# Keyspace" section, then emit - // the accurate per-db lines. + // the accurate per-db lines. Filtering runs afterwards, so a request for a + // single section still gets the ACCURATE keyspace numbers. let Some(cut) = text.find("# Keyspace\r\n") else { - return Frame::BulkString(base); + return crate::command::info_sections::finalize(&text, real_replication, args); }; let mut sections = String::with_capacity(text.len() + keyspace.len() * 32); sections.push_str(&text[..cut]); @@ -529,7 +620,7 @@ pub fn info_with_keyspace(db: &Database, args: &[Frame], keyspace: &[(u64, u64)] ); } } - Frame::BulkString(Bytes::from(sections)) + crate::command::info_sections::finalize(§ions, real_replication, args) } /// INFO command handler (read-only variant for RwLock read path). diff --git a/src/command/info_sections.rs b/src/command/info_sections.rs new file mode 100644 index 00000000..e8d88f3c --- /dev/null +++ b/src/command/info_sections.rs @@ -0,0 +1,252 @@ +//! INFO section selection, de-duplication, and the single assembly point. +//! +//! Moon builds INFO in more than one place. `connection::info` writes every +//! section including a STUB `# Replication`, and each connection handler then +//! appends the REAL replication section from +//! `replication::handshake::build_info_replication`. That is why `INFO` +//! historically emitted `# Replication` twice, and it is why filtering cannot +//! live inside `connection::info` alone — a filter applied before the append +//! would leak the appended section on every request. +//! +//! [`finalize`] is that single point: it takes the raw payload, optionally +//! substitutes the real replication section for the stub, drops any section +//! header seen twice, and then keeps only the sections the client asked for. +//! +//! Semantics measured against redis-server 8.6.1: +//! +//! ```text +//! INFO -> every section EXCEPT Commandstats/Latencystats +//! INFO all | everything -> every section +//! INFO replication -> only that one, case-insensitively +//! INFO server clients -> both, in the SERVER's order, not the caller's +//! INFO nosuchsection -> empty payload, NOT an error +//! ``` + +use crate::protocol::Frame; + +/// Sections omitted from a bare `INFO` and included only on explicit request +/// or via `all`/`everything`. Redis treats these as opt-in because they grow +/// with the command table rather than being fixed-size. +const NON_DEFAULT: [&str; 2] = ["commandstats", "latencystats"]; + +/// What the caller asked for. +enum Want { + /// Bare `INFO` — everything except [`NON_DEFAULT`]. + Default, + /// `INFO all` / `INFO everything`. + All, + /// An explicit list, already lowercased. May be empty (unknown section + /// only), which correctly yields an empty payload rather than an error. + Named(Vec), +} + +impl Want { + fn from_args(args: &[Frame]) -> Self { + let mut named: Vec = Vec::new(); + for a in args { + let raw = match a { + Frame::BulkString(b) => b.as_ref(), + Frame::SimpleString(s) => s.as_ref(), + _ => continue, + }; + let name = String::from_utf8_lossy(raw).to_ascii_lowercase(); + if name == "all" || name == "everything" { + return Want::All; + } + // A repeated section must not duplicate the section in the reply. + if !named.contains(&name) { + named.push(name); + } + } + if named.is_empty() { + Want::Default + } else { + Want::Named(named) + } + } + + fn accepts(&self, section_lower: &str) -> bool { + match self { + Want::All => true, + Want::Default => !NON_DEFAULT.contains(§ion_lower), + Want::Named(list) => list.iter().any(|n| n == section_lower), + } + } +} + +/// Split an assembled INFO payload into `(header_line, body_including_header)` +/// chunks. Anything before the first header (there should be nothing) is +/// dropped rather than silently attached to the first section. +fn split_sections(text: &str) -> Vec<(String, String)> { + let mut out: Vec<(String, String)> = Vec::new(); + let mut current: Option<(String, String)> = None; + for line in text.split_inclusive("\r\n") { + let trimmed = line.trim_end_matches(['\r', '\n']); + if let Some(name) = trimmed.strip_prefix("# ") { + if let Some(prev) = current.take() { + out.push(prev); + } + current = Some((name.trim().to_string(), line.to_string())); + } else if let Some((_, body)) = current.as_mut() { + body.push_str(line); + } + } + if let Some(prev) = current { + out.push(prev); + } + out +} + +/// Assemble the final INFO reply. +/// +/// `raw` is the payload built by `connection::info`. `real_replication`, when +/// present, is the authoritative `# Replication` section from the replication +/// subsystem; it REPLACES the stub rather than being appended, which is what +/// removes the duplicate header. +/// +/// De-duplication keeps the FIRST occurrence of a section. Callers that have a +/// better version of a section must pass it in, not append it. +pub fn finalize(raw: &str, real_replication: Option<&str>, args: &[Frame]) -> Frame { + let want = Want::from_args(args); + let mut chunks = split_sections(raw); + + if let Some(real) = real_replication { + // Substitute in place so the section keeps its canonical position. + let real_body = real.to_string(); + if let Some(slot) = chunks + .iter_mut() + .find(|(name, _)| name.eq_ignore_ascii_case("Replication")) + { + slot.1 = ensure_trailing_blank(&real_body); + } else { + chunks.push(("Replication".to_string(), ensure_trailing_blank(&real_body))); + } + } + + let mut seen: Vec = Vec::new(); + let mut out = String::with_capacity(raw.len()); + for (name, body) in chunks { + let lower = name.to_ascii_lowercase(); + if seen.contains(&lower) { + continue; // a header emitted twice would break map-building parsers + } + seen.push(lower.clone()); + if want.accepts(&lower) { + out.push_str(&body); + } + } + Frame::BulkString(bytes::Bytes::from(out)) +} + +/// Sections are blank-line separated; a substituted body that lacks the +/// separator would run into the next header. +fn ensure_trailing_blank(body: &str) -> String { + if body.ends_with("\r\n\r\n") { + body.to_string() + } else if body.ends_with("\r\n") { + format!("{body}\r\n") + } else { + format!("{body}\r\n\r\n") + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + + fn arg(s: &str) -> Frame { + Frame::BulkString(Bytes::from(s.to_string())) + } + + fn text(f: Frame) -> String { + match f { + Frame::BulkString(b) => String::from_utf8_lossy(&b).into_owned(), + other => panic!("expected bulk string, got {other:?}"), + } + } + + const RAW: &str = "# Server\r\nredis_version:7.4.0\r\n\r\n\ + # Clients\r\nconnected_clients:1\r\n\r\n\ + # Replication\r\nrole:master\r\n\r\n\ + # Commandstats\r\n\r\n"; + + #[test] + fn default_omits_commandstats() { + let got = text(finalize(RAW, None, &[])); + assert!(got.contains("# Server")); + assert!( + !got.contains("# Commandstats"), + "a bare INFO must omit Commandstats; got {got:?}" + ); + } + + #[test] + fn all_includes_commandstats() { + let got = text(finalize(RAW, None, &[arg("all")])); + assert!(got.contains("# Commandstats")); + } + + #[test] + fn single_section_only() { + let got = text(finalize(RAW, None, &[arg("replication")])); + assert!(got.starts_with("# Replication"), "got {got:?}"); + assert!(!got.contains("# Server"), "got {got:?}"); + } + + #[test] + fn section_match_is_case_insensitive() { + let lower = text(finalize(RAW, None, &[arg("replication")])); + let upper = text(finalize(RAW, None, &[arg("REPLICATION")])); + assert_eq!(lower, upper); + } + + #[test] + fn unknown_section_is_empty_not_error() { + let got = text(finalize(RAW, None, &[arg("nosuchsection")])); + assert!(got.is_empty(), "got {got:?}"); + } + + #[test] + fn repeated_section_emitted_once() { + let got = text(finalize(RAW, None, &[arg("server"), arg("server")])); + assert_eq!(got.matches("# Server").count(), 1, "got {got:?}"); + } + + #[test] + fn multiple_sections_use_server_order() { + // Caller asks clients-then-server; the reply must still be + // server-then-clients, because that is the assembly order. + let got = text(finalize(RAW, None, &[arg("clients"), arg("server")])); + let s = got.find("# Server").expect("server present"); + let c = got.find("# Clients").expect("clients present"); + assert!(s < c, "server must precede clients; got {got:?}"); + } + + #[test] + fn real_replication_replaces_stub_without_duplicating() { + let real = "# Replication\r\nrole:slave\r\nmaster_link_status:up\r\n"; + let got = text(finalize(RAW, Some(real), &[])); + assert_eq!( + got.matches("# Replication").count(), + 1, + "the real section must REPLACE the stub, not append; got {got:?}" + ); + assert!(got.contains("master_link_status:up"), "got {got:?}"); + assert!( + !got.contains("role:master"), + "the stub's body must be gone; got {got:?}" + ); + } + + #[test] + fn substituted_section_keeps_blank_separator() { + // Without the separator the next header would be glued to this body. + let real = "# Replication\r\nrole:slave\r\n"; + let got = text(finalize(RAW, Some(real), &[arg("all")])); + assert!( + got.contains("role:slave\r\n\r\n# Commandstats"), + "sections must stay blank-line separated; got {got:?}" + ); + } +} diff --git a/src/command/mod.rs b/src/command/mod.rs index d0564756..8743f8e8 100644 --- a/src/command/mod.rs +++ b/src/command/mod.rs @@ -12,6 +12,7 @@ pub mod helpers; pub mod hll; pub mod identity; pub mod info_reclamation; +pub mod info_sections; pub mod introspect; pub mod key; pub mod key_extra; diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 88296061..868a8fb8 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -664,22 +664,18 @@ pub(super) async fn try_handle_info( &ctx.spsc_notifiers, ) .await; - let response_text = crate::shard::slice::with_shard_db(conn.selected_db, |db| { - let resp_frame = conn_cmd::info_with_keyspace(db, cmd_args, &keyspace); - match resp_frame { - Frame::BulkString(b) => String::from_utf8_lossy(&b).to_string(), - _ => String::new(), - } + // The real replication section is PASSED IN, not appended: `info()` writes + // a stub `# Replication`, so appending produced the section twice and left + // section filtering unable to see the final set. + let real_repl = ctx + .repl_state + .as_ref() + .and_then(|rs| rs.try_read()) + .map(|rs_guard| crate::replication::handshake::build_info_replication(&rs_guard)); + let resp_frame = crate::shard::slice::with_shard_db(conn.selected_db, |db| { + conn_cmd::info_with_keyspace_and_replication(db, cmd_args, &keyspace, real_repl.as_deref()) }); - let mut response_text = response_text; - if let Some(ref rs) = ctx.repl_state { - if let Some(rs_guard) = rs.try_read() { - response_text.push_str(&crate::replication::handshake::build_info_replication( - &rs_guard, - )); - } - } - responses.push(Frame::BulkString(Bytes::from(response_text))); + responses.push(resp_frame); true } diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index a79ba29d..eaf6a84d 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -357,21 +357,17 @@ pub(super) async fn try_handle_info( ) .await; // ShardSlice path: access the local shard's database via thread-local. - let mut response_text = crate::shard::slice::with_shard_db(conn.selected_db, |db| { - let resp_frame = conn_cmd::info_with_keyspace(db, cmd_args, &keyspace); - match resp_frame { - Frame::BulkString(b) => String::from_utf8_lossy(&b).to_string(), - _ => String::new(), - } + // Passed in rather than appended — see the monoio handler for why the + // append produced a duplicate `# Replication`. + let real_repl = ctx + .repl_state + .as_ref() + .and_then(|rs| rs.try_read()) + .map(|rs_guard| crate::replication::handshake::build_info_replication(&rs_guard)); + let resp_frame = crate::shard::slice::with_shard_db(conn.selected_db, |db| { + conn_cmd::info_with_keyspace_and_replication(db, cmd_args, &keyspace, real_repl.as_deref()) }); - if let Some(ref rs) = ctx.repl_state { - if let Some(rs_guard) = rs.try_read() { - response_text.push_str(&crate::replication::handshake::build_info_replication( - &rs_guard, - )); - } - } - responses.push(Frame::BulkString(Bytes::from(response_text))); + responses.push(resp_frame); true } diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index bb849a7d..db82cd53 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -1065,20 +1065,20 @@ pub async fn handle_connection( (g.logical_len() as u64, g.expires_count() as u64) }) .collect(); + // Passed in rather than appended — appending + // emitted `# Replication` twice. + let real_repl = rs.try_read().map(|rs_guard| { + crate::replication::handshake::build_info_replication(&rs_guard) + }); let guard = db[conn.selected_db].read(); - let resp_frame = - conn_cmd::info_with_keyspace(&guard, cmd_args, &keyspace); + let resp_frame = conn_cmd::info_with_keyspace_and_replication( + &guard, + cmd_args, + &keyspace, + real_repl.as_deref(), + ); drop(guard); - let mut response_text = match resp_frame { - Frame::BulkString(b) => String::from_utf8_lossy(&b).to_string(), - _ => String::new(), - }; - if let Some(rs_guard) = rs.try_read() { - response_text.push_str( - &crate::replication::handshake::build_info_replication(&rs_guard), - ); - } - responses.push(Frame::BulkString(Bytes::from(response_text))); + responses.push(resp_frame); continue; } // Fall through to normal dispatch if no repl_state From d36f5fd57a4feaad9d8fca9653cd278ccc744359 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 13 Aug 2026 16:37:27 +0700 Subject: [PATCH 4/8] feat(info): maxmemory_policy, blocked_clients and pub/sub counts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The four fields io8 was still missing, each backed by a real source rather than a constant — a hardcoded zero is indistinguishable from a healthy server on a dashboard, which is the whole reason these get read. - `maxmemory_policy` is published to a process-global atomic at startup and on `CONFIG SET maxmemory-policy`, and INFO names it from that same atomic the eviction gate reads. Two copies of this value could drift, and the failure mode is an operator told the instance will OOM when it will in fact evict. - `blocked_clients` is a gauge maintained at the `BlockingRegistry`'s two `wait_keys` transitions, which are exactly when a client becomes and stops being blocked. It cannot be read from the registry directly: that lives behind an `Rc>` pinned to its shard thread, while INFO runs on whichever thread the asking connection landed on. A multi-key BLPOP registers one wait_id per key, so the gauge moves on the first registration only, and both removal paths (`remove_wait` and the timeout sweep) decrement. - `pubsub_channels` / `pubsub_patterns` are unioned across every shard's registry by the connection handlers — the same gather `PUBSUB CHANNELS` / `NUMPAT` perform. Summing per-shard counters would report a channel twice when its subscribers landed on different shard threads, which makes a healthy fan-out look like a leak. Handlers pass them in through a new `InstanceFacts`, following the precedent set by the replication section: INFO keeps one assembly point, and facts a single shard's `Database` cannot answer arrive from the layer that holds them. Field placement matches redis-server 8.6.1, verified by running one: blocked_clients under `# Clients`, maxmemory/maxmemory_policy under `# Memory`, pubsub_* under `# Stats`. `io11` asserts INFO agrees with `CONFIG GET` rather than with a hardcoded default — the first version of it asserted `noeviction` and failed, because Moon's memory guardrail auto-caps maxmemory and switches the policy when the operator sets neither. What must hold is that the two surfaces agree, and that INFO follows a flip. Not fixed here: `PubSubRegistry::numpat()` sums subscribers per pattern where Redis's `PUBSUB NUMPAT` returns unique patterns, so two clients on one pattern answers 2. INFO uses a new `pattern_names()` that counts correctly; correcting NUMPAT itself is a separate compatibility fix. Tests: io8 green, plus io11 (policy tracks CONFIG SET), io12 (a parked BLPOP is counted and serving it decrements), io13 (counts are instance-wide at --shards 4). 13/13 in tests/info_observability.rs. author: Tin Dang --- src/admin/metrics_setup.rs | 30 ++++- src/blocking/mod.rs | 19 ++- src/command/config.rs | 3 + src/command/connection.rs | 63 +++++++++- src/main.rs | 3 + src/pubsub/mod.rs | 31 +++++ src/server/conn/handler_monoio/dispatch.rs | 11 +- src/server/conn/handler_sharded/dispatch.rs | 11 +- src/server/conn/handler_single.rs | 13 ++- src/storage/eviction.rs | 59 ++++++++++ tests/info_observability.rs | 122 ++++++++++++++++++++ 11 files changed, 351 insertions(+), 14 deletions(-) diff --git a/src/admin/metrics_setup.rs b/src/admin/metrics_setup.rs index 10bfef9a..4f931bb2 100644 --- a/src/admin/metrics_setup.rs +++ b/src/admin/metrics_setup.rs @@ -3,7 +3,7 @@ //! Uses the `metrics` facade crate so metric recording is a single atomic //! operation on the hot path (counter increment or histogram observation). -use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, AtomicUsize, Ordering}; use metrics::{Unit, counter, describe_gauge, gauge, histogram}; @@ -62,6 +62,13 @@ static NET_OUTPUT_BYTES: AtomicU64 = AtomicU64::new(0); /// the delta since the previous sample, NOT a per-command computation. static OPS_LAST_SAMPLE: AtomicU64 = AtomicU64::new(0); static OPS_PER_SEC: AtomicU64 = AtomicU64::new(0); +/// Clients currently parked in a blocking command. A GAUGE, not a counter: +/// the per-shard `BlockingRegistry` that owns the truth is an `Rc>` +/// pinned to its shard thread, so INFO — which runs on whichever thread the +/// asking connection landed on — cannot read it directly. Maintained instead +/// at the registry's two `wait_keys` transitions, which are the exact points a +/// client becomes and stops being blocked. +static BLOCKED_CLIENTS: AtomicI64 = AtomicI64::new(0); /// Count one cooperative yield taken by the FT.SEARCH local slice (per chunk). #[inline] @@ -1679,6 +1686,27 @@ pub fn total_net_output_bytes() -> u64 { NET_OUTPUT_BYTES.load(Ordering::Relaxed) } +/// Clients currently parked in a blocking command (BLPOP, BRPOP, XREAD …). +/// +/// Never negative: a decrement that would go below zero means a +/// block/unblock pair was mismatched, and reporting a negative gauge would +/// turn a bookkeeping bug into a nonsense dashboard. +pub fn blocked_clients() -> u64 { + BLOCKED_CLIENTS.load(Ordering::Relaxed).max(0) as u64 +} + +/// A client entered a blocking wait. +#[inline] +pub fn record_client_blocked() { + BLOCKED_CLIENTS.fetch_add(1, Ordering::Relaxed); +} + +/// A client left a blocking wait — served, timed out, or cancelled. +#[inline] +pub fn record_client_unblocked() { + BLOCKED_CLIENTS.fetch_sub(1, Ordering::Relaxed); +} + /// Commands per second over the last sampling window. pub fn instantaneous_ops_per_sec() -> u64 { OPS_PER_SEC.load(Ordering::Relaxed) diff --git a/src/blocking/mod.rs b/src/blocking/mod.rs index 2568899e..6119beb3 100644 --- a/src/blocking/mod.rs +++ b/src/blocking/mod.rs @@ -122,10 +122,16 @@ impl BlockingRegistry { .or_insert_with(VecDeque::new) .push_back(entry); - self.wait_keys - .entry(wait_id) - .or_insert_with(Vec::new) - .push(queue_key); + // A wait_id appears in `wait_keys` exactly while its client is + // blocked, and a multi-key BLPOP registers the same id once per key — + // so the gauge moves on the FIRST registration only. + match self.wait_keys.entry(wait_id) { + std::collections::hash_map::Entry::Occupied(mut e) => e.get_mut().push(queue_key), + std::collections::hash_map::Entry::Vacant(e) => { + e.insert(vec![queue_key]); + crate::admin::metrics_setup::record_client_blocked(); + } + } } /// Pop the first waiter from the FIFO queue for (db_index, key). @@ -148,6 +154,7 @@ impl BlockingRegistry { /// Used after a waiter is woken or times out to clean up cross-key registrations. pub fn remove_wait(&mut self, wait_id: u64) { if let Some(keys) = self.wait_keys.remove(&wait_id) { + crate::admin::metrics_setup::record_client_unblocked(); for queue_key in keys { if let Some(queue) = self.waiters.get_mut(&queue_key) { queue.retain(|e| e.wait_id != wait_id); @@ -220,7 +227,9 @@ impl BlockingRegistry { timed_out_ids.sort_unstable(); timed_out_ids.dedup(); for id in timed_out_ids { - self.wait_keys.remove(&id); + if self.wait_keys.remove(&id).is_some() { + crate::admin::metrics_setup::record_client_unblocked(); + } } visited } diff --git a/src/command/config.rs b/src/command/config.rs index 05a840a3..62e9ee41 100644 --- a/src/command/config.rs +++ b/src/command/config.rs @@ -153,6 +153,9 @@ pub fn config_set(runtime_config: &mut RuntimeConfig, args: &[Frame]) -> Frame { ]; let lower = value_str.to_ascii_lowercase(); if valid.contains(&lower.as_str()) { + // Same publish contract as `maxmemory` above: INFO and the + // eviction gate must never name different policies. + crate::storage::eviction::publish_maxmemory_policy(&lower); runtime_config.maxmemory_policy = lower; } else { return Frame::Error(Bytes::from(format!( diff --git a/src/command/connection.rs b/src/command/connection.rs index e9d9820a..6aaf1d2c 100644 --- a/src/command/connection.rs +++ b/src/command/connection.rs @@ -170,7 +170,7 @@ fn format_memory_human(bytes: u64) -> String { /// dispatch). The connection handlers pass cross-shard, all-db stats through /// [`info_with_keyspace`] instead. pub fn info(db: &Database, args: &[Frame]) -> Frame { - let raw = info_raw(db); + let raw = info_raw(db, &InstanceFacts::default()); crate::command::info_sections::finalize(&raw, None, args) } @@ -212,7 +212,7 @@ fn run_id() -> &'static str { /// section here is a STUB that the connection handlers replace with the real /// one, so a filter applied during assembly would either emit the stub or drop /// the real section depending on where it ran. -fn info_raw(db: &Database) -> String { +fn info_raw(db: &Database, facts: &InstanceFacts) -> String { use std::fmt::Write as _; let mut sections = String::with_capacity(2048); @@ -249,6 +249,13 @@ fn info_raw(db: &Database) -> String { "parked_clients:{}\r\n", crate::client_registry::parked_clients(), ); + // A gauge, not a counter: an operator reads it to tell an idle server + // apart from one whose every worker is parked on an empty queue. + let _ = write!( + sections, + "blocked_clients:{}\r\n", + crate::admin::metrics_setup::blocked_clients(), + ); sections.push_str("\r\n"); sections.push_str("# Memory\r\n"); @@ -305,7 +312,8 @@ fn info_raw(db: &Database) -> String { allocator_overhead_bytes:{allocator_overhead_bytes}\r\n\ pagecache_bytes:{pagecache_bytes}\r\n\ mem_fragmentation_ratio:{frag:.2}\r\n\ - maxmemory:{maxmemory}\r\n", + maxmemory:{maxmemory}\r\n\ + maxmemory_policy:{maxmemory_policy}\r\n", used_memory = used_memory, human = format_memory_human(used_memory), rss = rss, @@ -326,6 +334,9 @@ fn info_raw(db: &Database) -> String { // Read from the same atomic the eviction gate enforces, so INFO can // never report a cap different from the one actually applied. maxmemory = crate::storage::eviction::maxmemory_bytes(), + // Named from the same published atomic the gate reads, so INFO cannot + // claim `noeviction` while the instance is in fact evicting. + maxmemory_policy = crate::storage::eviction::maxmemory_policy_name(), ); // Allocator counters, Redis's `allocator_*` field names so existing @@ -509,7 +520,9 @@ fn info_raw(db: &Database) -> String { rejected_connections:{}\r\n\ total_net_input_bytes:{}\r\n\ total_net_output_bytes:{}\r\n\ - instantaneous_ops_per_sec:{}\r\n", + instantaneous_ops_per_sec:{}\r\n\ + pubsub_channels:{}\r\n\ + pubsub_patterns:{}\r\n", crate::admin::metrics_setup::keyspace_hits(), crate::admin::metrics_setup::keyspace_misses(), crate::admin::metrics_setup::expired_keys(), @@ -518,6 +531,8 @@ fn info_raw(db: &Database) -> String { crate::admin::metrics_setup::total_net_input_bytes(), crate::admin::metrics_setup::total_net_output_bytes(), crate::admin::metrics_setup::instantaneous_ops_per_sec(), + facts.pubsub_channels, + facts.pubsub_patterns, ); sections.push_str("\r\n"); @@ -600,9 +615,47 @@ pub fn info_with_keyspace_and_replication( args: &[Frame], keyspace: &[(u64, u64)], real_replication: Option<&str>, +) -> Frame { + info_with_facts( + db, + args, + keyspace, + real_replication, + &InstanceFacts::default(), + ) +} + +/// Instance-wide facts INFO must report that a single shard's [`Database`] +/// cannot answer. +/// +/// Pub/sub counts are the motivating case: a channel with subscribers on two +/// shard threads exists in two per-shard registries, so summing per-registry +/// counters would report it twice. The connection handlers already hold +/// `all_pubsub_registries` and already de-duplicate for `PUBSUB CHANNELS`, so +/// they compute the same way and pass the answer in — the alternative, a +/// process-global counter maintained at subscribe time, cannot dedupe. +/// +/// Defaults to zeroes so a caller without handler context (Lua's `redis.call`, +/// unit tests) still gets a well-formed INFO rather than a missing field. +#[derive(Default, Clone, Copy)] +pub struct InstanceFacts { + /// Distinct channels with at least one subscriber, across all shards. + pub pubsub_channels: usize, + /// Distinct subscribed patterns, across all shards. + pub pubsub_patterns: usize, +} + +/// As [`info_with_keyspace_and_replication`], plus the instance-wide facts +/// only a connection handler can gather. +pub fn info_with_facts( + db: &Database, + args: &[Frame], + keyspace: &[(u64, u64)], + real_replication: Option<&str>, + facts: &InstanceFacts, ) -> Frame { use std::fmt::Write as _; - let text = info_raw(db); + let text = info_raw(db, facts); // Rebuild everything up to the fallback "# Keyspace" section, then emit // the accurate per-db lines. Filtering runs afterwards, so a request for a // single section still gets the ACCURATE keyspace numbers. diff --git a/src/main.rs b/src/main.rs index df07d212..35f06a53 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1132,6 +1132,9 @@ fn main() -> anyhow::Result<()> { // any server launched with --maxmemory (until the first CONFIG SET). moon::storage::eviction::publish_maxmemory(runtime_config_shared.read().maxmemory as u64); moon::config::log_maxmemory_sharding(runtime_config_shared.read().maxmemory, num_shards); + moon::storage::eviction::publish_maxmemory_policy( + &runtime_config_shared.read().maxmemory_policy, + ); // Record what the process costs the OS with no dataset in it. The eviction // budget is scaled by how far real footprint exceeds accounted memory, and // that comparison is only meaningful on the MARGINAL cost of the data — diff --git a/src/pubsub/mod.rs b/src/pubsub/mod.rs index c390dbfe..a8212cd6 100644 --- a/src/pubsub/mod.rs +++ b/src/pubsub/mod.rs @@ -270,6 +270,17 @@ impl PubSubRegistry { .collect() } + /// The DISTINCT patterns this registry holds, for INFO's + /// `pubsub_patterns`. + /// + /// Deliberately not [`Self::numpat`], which sums subscribers per pattern: + /// INFO reports how many patterns exist, so two clients on one pattern is + /// one, and the caller unions these across shards to avoid counting a + /// pattern twice when its subscribers landed on different shard threads. + pub fn pattern_names(&self) -> Vec { + self.patterns.iter().map(|(p, _)| p.clone()).collect() + } + /// Return total number of pattern subscriptions across all patterns. pub fn numpat(&self) -> usize { self.patterns.iter().map(|(_, subs)| subs.len()).sum() @@ -528,6 +539,26 @@ fn pmessage_frame_push(pattern: &Bytes, channel: &Bytes, payload: &Bytes) -> Fra ]) } +/// Instance-wide `(pubsub_channels, pubsub_patterns)` for INFO. +/// +/// Unions across every shard's registry rather than summing: a channel with +/// subscribers on two shard threads exists in two registries, and reporting it +/// twice would make a healthy fan-out look like a leak. Mirrors exactly what +/// `PUBSUB CHANNELS` / `PUBSUB NUMPAT` scatter-gather, so the two surfaces +/// cannot disagree. +pub fn instance_pubsub_counts( + registries: &[std::sync::Arc>], +) -> (usize, usize) { + let mut channels: std::collections::HashSet = std::collections::HashSet::new(); + let mut patterns: std::collections::HashSet = std::collections::HashSet::new(); + for reg in registries { + let guard = reg.read(); + channels.extend(guard.active_channels(None)); + patterns.extend(guard.pattern_names()); + } + (channels.len(), patterns.len()) +} + #[cfg(all(test, feature = "runtime-tokio"))] mod tests { use super::*; diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 868a8fb8..07730ae6 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -672,8 +672,17 @@ pub(super) async fn try_handle_info( .as_ref() .and_then(|rs| rs.try_read()) .map(|rs_guard| crate::replication::handshake::build_info_replication(&rs_guard)); + // Instance-wide pub/sub counts, unioned across every shard's registry — + // the same gather `PUBSUB CHANNELS`/`NUMPAT` do, so INFO cannot disagree + // with them. + let (pubsub_channels, pubsub_patterns) = + crate::pubsub::instance_pubsub_counts(&ctx.all_pubsub_registries); + let pubsub_facts = conn_cmd::InstanceFacts { + pubsub_channels, + pubsub_patterns, + }; let resp_frame = crate::shard::slice::with_shard_db(conn.selected_db, |db| { - conn_cmd::info_with_keyspace_and_replication(db, cmd_args, &keyspace, real_repl.as_deref()) + conn_cmd::info_with_facts(db, cmd_args, &keyspace, real_repl.as_deref(), &pubsub_facts) }); responses.push(resp_frame); true diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index eaf6a84d..38a4c1a5 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -364,8 +364,17 @@ pub(super) async fn try_handle_info( .as_ref() .and_then(|rs| rs.try_read()) .map(|rs_guard| crate::replication::handshake::build_info_replication(&rs_guard)); + // Instance-wide pub/sub counts, unioned across every shard's registry — + // the same gather `PUBSUB CHANNELS`/`NUMPAT` do, so INFO cannot disagree + // with them. + let (pubsub_channels, pubsub_patterns) = + crate::pubsub::instance_pubsub_counts(&ctx.all_pubsub_registries); + let pubsub_facts = conn_cmd::InstanceFacts { + pubsub_channels, + pubsub_patterns, + }; let resp_frame = crate::shard::slice::with_shard_db(conn.selected_db, |db| { - conn_cmd::info_with_keyspace_and_replication(db, cmd_args, &keyspace, real_repl.as_deref()) + conn_cmd::info_with_facts(db, cmd_args, &keyspace, real_repl.as_deref(), &pubsub_facts) }); responses.push(resp_frame); true diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index db82cd53..f5ce04f0 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -1070,12 +1070,23 @@ pub async fn handle_connection( let real_repl = rs.try_read().map(|rs_guard| { crate::replication::handshake::build_info_replication(&rs_guard) }); + // One registry on this handler (embedded / + // non-sharded), so the union the sharded + // handlers perform collapses to a direct read. + let pubsub_facts = { + let reg = pubsub_registry.lock(); + conn_cmd::InstanceFacts { + pubsub_channels: reg.active_channels(None).len(), + pubsub_patterns: reg.pattern_names().len(), + } + }; let guard = db[conn.selected_db].read(); - let resp_frame = conn_cmd::info_with_keyspace_and_replication( + let resp_frame = conn_cmd::info_with_facts( &guard, cmd_args, &keyspace, real_repl.as_deref(), + &pubsub_facts, ); drop(guard); responses.push(resp_frame); diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs index 3921a9d9..f026ce0f 100644 --- a/src/storage/eviction.rs +++ b/src/storage/eviction.rs @@ -53,6 +53,33 @@ pub fn publish_maxmemory(bytes: u64) { MAXMEMORY_GLOBAL.store(bytes, std::sync::atomic::Ordering::Relaxed); } +/// Process-global eviction policy, as an [`EvictionPolicy`] discriminant. +/// +/// Published alongside [`MAXMEMORY_GLOBAL`] so INFO can name the policy +/// without taking the runtime-config lock, and — more to the point — so it +/// names the policy the gate will actually apply rather than a second copy +/// that can drift from it. +static MAXMEMORY_POLICY_GLOBAL: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0); + +/// Publish the current `maxmemory-policy`. Same contract as +/// [`publish_maxmemory`]: every production write site of +/// `RuntimeConfig.maxmemory_policy` must call it — startup and +/// `CONFIG SET maxmemory-policy`. +#[inline] +pub fn publish_maxmemory_policy(name: &str) { + MAXMEMORY_POLICY_GLOBAL.store( + EvictionPolicy::from_str(name).as_u8(), + std::sync::atomic::Ordering::Relaxed, + ); +} + +/// Canonical name of the published eviction policy, for INFO. +#[inline] +pub fn maxmemory_policy_name() -> &'static str { + EvictionPolicy::from_u8(MAXMEMORY_POLICY_GLOBAL.load(std::sync::atomic::Ordering::Relaxed)) + .as_str() +} + /// The configured `maxmemory` in bytes, 0 when unlimited. /// /// Reads the same published atomic the eviction gate uses, so INFO cannot @@ -260,6 +287,38 @@ impl EvictionPolicy { EvictionPolicy::VolatileTtl => "volatile-ttl", } } + + /// Discriminant for the process-global publish. Written out rather than + /// derived via `as` so that reordering the enum cannot silently change + /// what a running process's published value means. + pub fn as_u8(self) -> u8 { + match self { + EvictionPolicy::NoEviction => 0, + EvictionPolicy::AllKeysLru => 1, + EvictionPolicy::AllKeysLfu => 2, + EvictionPolicy::AllKeysRandom => 3, + EvictionPolicy::VolatileLru => 4, + EvictionPolicy::VolatileLfu => 5, + EvictionPolicy::VolatileRandom => 6, + EvictionPolicy::VolatileTtl => 7, + } + } + + /// Inverse of [`Self::as_u8`]. An unknown byte means the publish and the + /// read disagree about the encoding, which can only happen through a bug — + /// report the safe policy rather than guess. + pub fn from_u8(v: u8) -> Self { + match v { + 1 => EvictionPolicy::AllKeysLru, + 2 => EvictionPolicy::AllKeysLfu, + 3 => EvictionPolicy::AllKeysRandom, + 4 => EvictionPolicy::VolatileLru, + 5 => EvictionPolicy::VolatileLfu, + 6 => EvictionPolicy::VolatileRandom, + 7 => EvictionPolicy::VolatileTtl, + _ => EvictionPolicy::NoEviction, + } + } } /// OOM error frame returned when eviction cannot free enough memory. diff --git a/tests/info_observability.rs b/tests/info_observability.rs index 24941525..2d325d30 100644 --- a/tests/info_observability.rs +++ b/tests/info_observability.rs @@ -126,6 +126,18 @@ impl Conn { self.read_reply() } + /// Write a command without waiting for its reply — for commands that are + /// SUPPOSED not to answer yet (a parked `BLPOP`, a `SUBSCRIBE` whose push + /// stream we do not consume). Reading here would block the test, not the + /// server. + fn write_only(&mut self, parts: &[&str]) { + let mut out = format!("*{}\r\n", parts.len()); + for p in parts { + out.push_str(&format!("${}\r\n{p}\r\n", p.len())); + } + self.0.write_all(out.as_bytes()).expect("write"); + } + fn read_reply(&mut self) -> String { let mut buf = [0u8; 16384]; let mut acc = Vec::new(); @@ -380,3 +392,113 @@ fn io10_keyspace_hit_miss_counters() { "one missing-key GET must move keyspace_misses by exactly 1" ); } + +#[test] +fn io11_maxmemory_policy_reflects_config() { + // A dashboard reads `maxmemory_policy` to decide whether an OOM is an + // operator choice (`noeviction`) or a bug. Reporting a constant would be + // worse than omitting the field, so it must track CONFIG SET. + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + // Against CONFIG GET, not a hardcoded name: Moon's memory guardrail + // auto-caps maxmemory and switches the policy when the operator sets + // neither, so the startup default is a runtime decision. What must hold is + // that the two surfaces agree. + let configured = c.send(&["CONFIG", "GET", "maxmemory-policy"]); + let configured = configured + .lines() + .last() + .map(|l| l.trim().to_string()) + .expect("CONFIG GET reply"); + assert_eq!( + field(&c.send(&["INFO", "memory"]), "maxmemory_policy").as_deref(), + Some(configured.as_str()), + "INFO and CONFIG GET must name the same policy at startup" + ); + + let flipped = if configured == "noeviction" { + "allkeys-lru" + } else { + "noeviction" + }; + c.send(&["CONFIG", "SET", "maxmemory-policy", flipped]); + assert_eq!( + field(&c.send(&["INFO", "memory"]), "maxmemory_policy").as_deref(), + Some(flipped), + "INFO must follow CONFIG SET — a stale policy tells an operator the \ + instance will OOM when it will in fact evict, or vice versa" + ); +} + +#[test] +fn io12_blocked_clients_tracks_a_real_block() { + // `blocked_clients` is how an operator distinguishes "the server is idle" + // from "every worker is parked on an empty queue". A hardcoded 0 reads as + // the former while the latter is happening. + let m = spawn_moon("1"); + let mut observer = Conn::open(m.port); + assert_eq!( + field(&observer.send(&["INFO", "clients"]), "blocked_clients").as_deref(), + Some("0"), + "no client is blocked yet" + ); + + let mut blocker = Conn::open(m.port); + blocker.write_only(&["BLPOP", "io12queue", "0"]); + // The block registers on the shard thread; give it a moment to land. + let mut seen = None; + for _ in 0..50 { + std::thread::sleep(Duration::from_millis(20)); + seen = field(&observer.send(&["INFO", "clients"]), "blocked_clients"); + if seen.as_deref() == Some("1") { + break; + } + } + assert_eq!( + seen.as_deref(), + Some("1"), + "a client parked in BLPOP must be counted" + ); + + observer.send(&["LPUSH", "io12queue", "v"]); + let mut after = None; + for _ in 0..50 { + std::thread::sleep(Duration::from_millis(20)); + after = field(&observer.send(&["INFO", "clients"]), "blocked_clients"); + if after.as_deref() == Some("0") { + break; + } + } + assert_eq!( + after.as_deref(), + Some("0"), + "serving the blocked client must decrement the gauge — a counter that \ + only goes up is worse than no counter" + ); +} + +#[test] +fn io13_pubsub_counts_are_instance_wide() { + // Both fields must agree with the PUBSUB command, which scatter-gathers + // across every shard's registry. A local-only answer under-reports by + // roughly 1/N and makes a fan-out look broken. + let m = spawn_moon("4"); + let mut sub = Conn::open(m.port); + sub.write_only(&["SUBSCRIBE", "io13a", "io13b"]); + let mut psub = Conn::open(m.port); + psub.write_only(&["PSUBSCRIBE", "io13.*"]); + std::thread::sleep(Duration::from_millis(300)); + + let mut c = Conn::open(m.port); + let stats = c.send(&["INFO", "stats"]); + assert_eq!( + field(&stats, "pubsub_channels").as_deref(), + Some("2"), + "two subscribed channels must be visible instance-wide" + ); + assert_eq!( + field(&stats, "pubsub_patterns").as_deref(), + Some("1"), + "one subscribed pattern must be visible instance-wide" + ); +} From 47d6b0cc7b3e7a6fc7617ba7cfa8eebce8a02a9e Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 13 Aug 2026 16:45:18 +0700 Subject: [PATCH 5/8] feat(notify): notify-keyspace-events flag model and config surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The configuration half of keyspace notifications: parsing, canonical readback, validation wording, and the process-global publish the emit path will read. No events fire yet — that is the next commit — but the gating this establishes is what makes "off by default" mean something. The flag string is not a set. It is an ordered canonical form that clients read back and compare, and the order is not what the letters suggest. Every rule here was measured against a running redis-server 8.6.1 rather than recalled, and two of the 29 captured pairs disprove the obvious "one ordered list" model: KEA -> AKE Kg$ -> g$K Km -> Km m trails K/E, unlike every other letter mn -> nm ...while n sorts WITH the classes An -> A so A swallows n Amn -> Am ...but not m g$lshzxetdmnKE -> AKEm So emission is: `A` (whenever all ten classes are present) or the class letters `g $ l s h z x e t d n`, then `K`, `E`, and finally `m`. The table of all 29 measured pairs is the unit test; a fixpoint test pins that a client writing back what it read does not drift the config. `A` deliberately excludes `m` (keymiss) and `n` (newkey) — the two read-path classes. That is why `A` is safe to enable in production and why kn7 asserts a key miss is silent under it. `is_enabled()` requires BOTH a channel family (K or E) and at least one class: class flags with neither K nor E deliver nothing however many are set, and K/E with no class selects nothing to deliver. The emit path gates on this rather than on "any flags set", so the default costs one Relaxed load and two masks. Flags live in a process-global atomic, not behind the config lock — this is read on every mutation that could notify, and a lock there would be a lock on the write path. Same publish contract as `maxmemory`: every write site must publish or `CONFIG GET` and the emit path drift apart. Implemented as a plain bitset newtype rather than pulling in `bitflags` as a direct dependency for six operations. Tests: kn1 (invalid char rejected with Redis's verbatim wording, and the rejected SET leaves the previous value intact), kn2 (canonicalization), kn8 (K/E gating) and kn10 (off by default) now pass — kn8 for a real reason rather than vacuously. kn3-kn7 and kn9 remain red pending event emission. author: Tin Dang --- src/command/config.rs | 18 +++ src/lib.rs | 1 + src/notify.rs | 335 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 354 insertions(+) create mode 100644 src/notify.rs diff --git a/src/command/config.rs b/src/command/config.rs index 62e9ee41..0b95dbf4 100644 --- a/src/command/config.rs +++ b/src/command/config.rs @@ -28,6 +28,12 @@ pub fn config_get( let params: Vec<(&[u8], String)> = vec![ (b"maxmemory" as &[u8], runtime_config.maxmemory.to_string()), (b"maxmemory-policy", runtime_config.maxmemory_policy.clone()), + ( + // Canonical form, not the caller's spelling: `CONFIG SET KEA` + // reads back `AKE`, and clients compare the readback. + b"notify-keyspace-events", + crate::notify::flags_to_string(crate::notify::published_flags()), + ), ( b"maxmemory-samples", runtime_config.maxmemory_samples.to_string(), @@ -164,6 +170,18 @@ pub fn config_set(runtime_config: &mut RuntimeConfig, args: &[Frame]) -> Frame { ))); } } + "notify-keyspace-events" => match crate::notify::parse_flags(&value_str) { + Ok(flags) => crate::notify::publish_flags(flags), + Err(reason) => { + // Redis wraps the class-set message in its generic CONFIG + // SET failure envelope; config-management tooling matches + // on the tail, so both halves are reproduced verbatim. + return Frame::Error(Bytes::from(format!( + "ERR CONFIG SET failed (possibly related to argument \ + 'notify-keyspace-events') - {reason}" + ))); + } + }, "maxmemory-samples" => match value_str.parse::() { Ok(v) if v > 0 => runtime_config.maxmemory_samples = v, _ => { diff --git a/src/lib.rs b/src/lib.rs index 710dab58..9f0e2a9f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -54,6 +54,7 @@ pub mod graph; pub mod io; pub mod memory_ctl; pub mod mq; +pub mod notify; pub mod persistence; pub mod protocol; pub mod pubsub; diff --git a/src/notify.rs b/src/notify.rs new file mode 100644 index 00000000..2f412f6b --- /dev/null +++ b/src/notify.rs @@ -0,0 +1,335 @@ +//! Keyspace notification flags: `notify-keyspace-events`. +//! +//! The flag string is not a set — it is an ordered canonical form, and clients +//! read it back. Every rule below was MEASURED against redis-server 8.6.1 +//! rather than recalled, because the ordering is not what the obvious reading +//! of the letters suggests: +//! +//! ```text +//! KEA -> AKE A collapses the ten class flags +//! Kg$ -> g$K classes first, then K/E +//! Km -> Km ...but m trails K/E, unlike the other letters +//! mn -> nm n is a CLASS letter, m is not +//! An -> A so `A` swallows n as well +//! Amn -> Am ...while m survives it +//! g$lshzxetdmnKE -> AKEm +//! ``` +//! +//! Emission order is therefore: `A` **or** the class letters +//! `g $ l s h z x e t d n`, then `K`, `E`, and finally `m`. `A` is emitted +//! whenever all ten classes are present — `n` is not required for it, but is +//! suppressed by it. + +/// Which events fire, and whether they are delivered. +/// +/// A plain bitset newtype rather than a `bitflags!` macro: the crate is not a +/// direct dependency of moon, and this needs six operations. +/// +/// `KEYSPACE`/`KEYEVENT` are not classes — they select the two channel +/// families. With neither set nothing is delivered however many class flags +/// are on, which is why the default is genuinely zero-cost. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct NotifyFlags(u16); + +impl NotifyFlags { + /// `K` — publish to `__keyspace@__:`. + pub const KEYSPACE: NotifyFlags = NotifyFlags(1 << 0); + /// `E` — publish to `__keyevent@__:`. + pub const KEYEVENT: NotifyFlags = NotifyFlags(1 << 1); + /// `g` — generic commands (DEL, EXPIRE, RENAME ...). + pub const GENERIC: NotifyFlags = NotifyFlags(1 << 2); + /// `$` — string commands. + pub const STRING: NotifyFlags = NotifyFlags(1 << 3); + /// `l` — list commands. + pub const LIST: NotifyFlags = NotifyFlags(1 << 4); + /// `s` — set commands. + pub const SET: NotifyFlags = NotifyFlags(1 << 5); + /// `h` — hash commands. + pub const HASH: NotifyFlags = NotifyFlags(1 << 6); + /// `z` — sorted set commands. + pub const ZSET: NotifyFlags = NotifyFlags(1 << 7); + /// `x` — expired events. + pub const EXPIRED: NotifyFlags = NotifyFlags(1 << 8); + /// `e` — evicted events. + pub const EVICTED: NotifyFlags = NotifyFlags(1 << 9); + /// `t` — stream commands. + pub const STREAM: NotifyFlags = NotifyFlags(1 << 10); + /// `d` — module key type events. + pub const MODULE: NotifyFlags = NotifyFlags(1 << 11); + /// `m` — key-miss events. Deliberately NOT part of `A`: it would put a + /// pub/sub fan-out on the read path. + pub const KEY_MISS: NotifyFlags = NotifyFlags(1 << 12); + /// `n` — new-key events. Suppressed by `A`'s collapse but not required + /// for it. + pub const NEW_KEY: NotifyFlags = NotifyFlags(1 << 13); + + /// No flags — notifications off. + pub const NONE: NotifyFlags = NotifyFlags(0); + + /// Union. + #[inline] + pub const fn union(self, other: NotifyFlags) -> NotifyFlags { + NotifyFlags(self.0 | other.0) + } + + /// `true` when every bit of `other` is set here. + #[inline] + pub const fn contains(self, other: NotifyFlags) -> bool { + self.0 & other.0 == other.0 + } + + /// `true` when any bit of `other` is set here. + #[inline] + pub const fn intersects(self, other: NotifyFlags) -> bool { + self.0 & other.0 != 0 + } + + /// `true` when no flag is set. + #[inline] + pub const fn is_empty(self) -> bool { + self.0 == 0 + } + + /// Raw bits, for storing the value in an atomic. + #[inline] + pub const fn bits(self) -> u16 { + self.0 + } + + /// Rebuild from raw bits read out of an atomic. + #[inline] + pub const fn from_bits(bits: u16) -> NotifyFlags { + NotifyFlags(bits) + } +} + +impl std::ops::BitOrAssign for NotifyFlags { + fn bitor_assign(&mut self, rhs: NotifyFlags) { + self.0 |= rhs.0; + } +} + +impl NotifyFlags { + /// The `A` class: every type/event class except `m` and `n`. + pub const ALL_CLASSES: NotifyFlags = NotifyFlags( + NotifyFlags::GENERIC.0 + | NotifyFlags::STRING.0 + | NotifyFlags::LIST.0 + | NotifyFlags::SET.0 + | NotifyFlags::HASH.0 + | NotifyFlags::ZSET.0 + | NotifyFlags::EXPIRED.0 + | NotifyFlags::EVICTED.0 + | NotifyFlags::STREAM.0 + | NotifyFlags::MODULE.0, + ); + + /// Every class letter, including the two `A` leaves out. + const ANY_CLASS: NotifyFlags = + NotifyFlags(NotifyFlags::ALL_CLASSES.0 | NotifyFlags::KEY_MISS.0 | NotifyFlags::NEW_KEY.0); + + /// `true` when at least one event could actually be delivered. + /// + /// Class flags with neither `K` nor `E` deliver nothing, and `K`/`E` with + /// no class selects nothing to deliver — so the emit path must check this + /// rather than merely "are any flags set". + #[inline] + pub const fn is_enabled(self) -> bool { + self.intersects(NotifyFlags( + NotifyFlags::KEYSPACE.0 | NotifyFlags::KEYEVENT.0, + )) && self.intersects(NotifyFlags::ANY_CLASS) + } +} + +/// Process-global published flags. +/// +/// Read on every mutation that could notify, so it must be a Relaxed atomic +/// load and nothing more — a config-lock read here would put a lock on the +/// write path. Same publish contract as `maxmemory`: every write site of the +/// config value must call [`publish_flags`], or the emit path silently +/// disagrees with `CONFIG GET`. +static PUBLISHED: std::sync::atomic::AtomicU16 = std::sync::atomic::AtomicU16::new(0); + +/// Publish the active flag set. Startup and `CONFIG SET`. +#[inline] +pub fn publish_flags(flags: NotifyFlags) { + PUBLISHED.store(flags.bits(), std::sync::atomic::Ordering::Relaxed); +} + +/// The active flag set. +#[inline] +pub fn published_flags() -> NotifyFlags { + NotifyFlags::from_bits(PUBLISHED.load(std::sync::atomic::Ordering::Relaxed)) +} + +/// `true` when any event could be delivered — the one check the write path +/// pays when notifications are off (a Relaxed load and two masks). +#[inline] +pub fn notifications_enabled() -> bool { + published_flags().is_enabled() +} + +/// The valid characters, in the order Redis names them in its error message. +pub const VALID_FLAG_CHARS: &str = "Ag$lshzxeKEtmdn"; + +/// Redis 8.6.1's wording, verbatim — a config-management tool surfaces this +/// string unchanged, so paraphrasing it is a compatibility break. +pub const INVALID_FLAG_ERROR: &str = "Invalid event class character. Use 'Ag$lshzxeKEtmdn'."; + +/// Parse a `notify-keyspace-events` flag string. +/// +/// Returns `Err` naming the offending character's class set on the FIRST +/// invalid character, and parses nothing — a partially-applied flag set would +/// silently change which events fire. +pub fn parse_flags(s: &str) -> Result { + let mut flags = NotifyFlags::NONE; + for c in s.chars() { + flags |= match c { + 'A' => NotifyFlags::ALL_CLASSES, + 'K' => NotifyFlags::KEYSPACE, + 'E' => NotifyFlags::KEYEVENT, + 'g' => NotifyFlags::GENERIC, + '$' => NotifyFlags::STRING, + 'l' => NotifyFlags::LIST, + 's' => NotifyFlags::SET, + 'h' => NotifyFlags::HASH, + 'z' => NotifyFlags::ZSET, + 'x' => NotifyFlags::EXPIRED, + 'e' => NotifyFlags::EVICTED, + 't' => NotifyFlags::STREAM, + 'd' => NotifyFlags::MODULE, + 'm' => NotifyFlags::KEY_MISS, + 'n' => NotifyFlags::NEW_KEY, + _ => return Err(INVALID_FLAG_ERROR), + }; + } + Ok(flags) +} + +/// Render flags back to their canonical string — what `CONFIG GET` returns. +/// +/// Not the caller's spelling: `CONFIG SET KEA` reads back as `AKE`. See the +/// module docs for why `n` sits with the classes and `m` does not. +pub fn flags_to_string(flags: NotifyFlags) -> String { + let mut out = String::with_capacity(8); + if flags.contains(NotifyFlags::ALL_CLASSES) { + out.push('A'); + } else { + for (bit, ch) in [ + (NotifyFlags::GENERIC, 'g'), + (NotifyFlags::STRING, '$'), + (NotifyFlags::LIST, 'l'), + (NotifyFlags::SET, 's'), + (NotifyFlags::HASH, 'h'), + (NotifyFlags::ZSET, 'z'), + (NotifyFlags::EXPIRED, 'x'), + (NotifyFlags::EVICTED, 'e'), + (NotifyFlags::STREAM, 't'), + (NotifyFlags::MODULE, 'd'), + (NotifyFlags::NEW_KEY, 'n'), + ] { + if flags.contains(bit) { + out.push(ch); + } + } + } + if flags.contains(NotifyFlags::KEYSPACE) { + out.push('K'); + } + if flags.contains(NotifyFlags::KEYEVENT) { + out.push('E'); + } + if flags.contains(NotifyFlags::KEY_MISS) { + out.push('m'); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Every pair here was captured from a running redis-server 8.6.1, not + /// derived from the letters. `Km -> Km` and `mn -> nm` are the two that + /// disprove the obvious "one ordered list" model. + #[test] + fn canonical_form_matches_measured_redis() { + for (input, want) in [ + ("KEA", "AKE"), + ("Kg$", "g$K"), + ("xe", "xe"), + ("Km", "Km"), + ("mK", "Km"), + ("Em", "Em"), + ("KEm", "KEm"), + ("mn", "nm"), + ("nm", "nm"), + ("KEmn", "nKEm"), + ("Amn", "Am"), + ("An", "A"), + ("nA", "A"), + ("Anm", "Am"), + ("n", "n"), + ("nK", "nK"), + ("Kn", "nK"), + ("nE", "nE"), + ("gn", "gn"), + ("nd", "dn"), + ("dn", "dn"), + ("tn", "tn"), + ("gxE", "gxE"), + ("g$lshzxetdn", "A"), + ("g$lshzxetdmnKE", "AKEm"), + ("EKdtezxhslg$", "AKE"), + ("", ""), + ("K", "K"), + ("A", "A"), + ] { + let parsed = parse_flags(input).expect("valid flag string"); + assert_eq!( + flags_to_string(parsed), + want, + "canonicalization of {input:?} diverges from redis-server 8.6.1" + ); + } + } + + #[test] + fn canonical_form_is_idempotent() { + // A client that writes back what it read must not drift the config. + for input in ["KEA", "Kg$", "Km", "KEmn", "Amn", "gn", "g$lshzxetdmnKE"] { + let once = flags_to_string(parse_flags(input).expect("valid")); + let twice = flags_to_string(parse_flags(&once).expect("canonical form re-parses")); + assert_eq!(once, twice, "canonical form of {input:?} is not a fixpoint"); + } + } + + #[test] + fn invalid_char_is_rejected_with_redis_wording() { + // 'Q' is not in the class set. The message is compared verbatim by + // config-management tooling. + assert_eq!(parse_flags("KEQ"), Err(INVALID_FLAG_ERROR)); + assert!(INVALID_FLAG_ERROR.contains(VALID_FLAG_CHARS)); + } + + #[test] + fn a_excludes_keymiss_and_newkey() { + // The reason `A` is safe to enable in production: neither of the two + // read-path classes is in it. + let a = parse_flags("A").expect("valid"); + assert!(!a.contains(NotifyFlags::KEY_MISS), "'m' must not be in 'A'"); + assert!(!a.contains(NotifyFlags::NEW_KEY), "'n' must not be in 'A'"); + } + + #[test] + fn classes_without_k_or_e_deliver_nothing() { + // kn8's invariant, at the unit level: K/E select WHETHER, classes + // select WHICH. All the classes in the world with neither is silence. + assert!(!parse_flags("g$").expect("valid").is_enabled()); + assert!(!parse_flags("A").expect("valid").is_enabled()); + // ...and K/E with no class is equally silent. + assert!(!parse_flags("KE").expect("valid").is_enabled()); + assert!(parse_flags("KEA").expect("valid").is_enabled()); + assert!(parse_flags("Km").expect("valid").is_enabled()); + } +} From f50a15c3bd4a0d6c17f7c7700620b98ad7c96cf1 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 13 Aug 2026 23:51:20 +0700 Subject: [PATCH 6/8] feat(notify): emit keyspace events, and fix the GET route that reached no recorder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes keyspace notifications: events are produced by command code, queued per shard thread, and delivered over the pub/sub mesh. kn3-kn7 and kn9 now pass on BOTH runtimes. Why an outbox rather than publishing at the source: command code has no access to the pub/sub registries, and a subscriber's task lives on the shard thread that accepted it, where a `Waker` fired from another thread does not reach it (the monoio note in CLAUDE.md). Publishing straight into a remote shard's registry would queue the message and never wake the reader. So `notify_keyspace_event` appends to a thread-local outbox and two drains deliver it — the connection handlers after each batch, and the shard event loop after each SPSC drain. Both drains are required, for reasons a single-shard test cannot show: * A write routed to the shard that OWNS the key executes on that shard's thread, so its events land in that thread's outbox. With only the connection-side drain, kn9 saw 1 of 8 events at --shards 4 — the one key that happened to hash to the subscriber's own shard. * TTL expiry and eviction have no connection at all; they run from the shard timer. Draining once per batch rather than per command means a 1000-command pipeline costs one fan-out per target shard. Delivery is fire-and-forget: a full ring drops the batch rather than blocking a write path, which is the trade Redis makes for a slow subscriber. Events wired: `set` and `incrby` (STRING), `rename_from`/`rename_to` (GENERIC — two events, carrying different keys), `expired` (EXPIRED), `keymiss` (KEY_MISS). `Database` gained `db_index` because both channel names embed it and command code is handed a `&Database` with no other way to know which db it has. Closes #477 as a consequence. `try_inline_dispatch` — the route a plain `GET key` actually takes under monoio — frames its reply straight into the write buffer and returns, reaching neither `string::get` nor `string::get_readonly`. So `keyspace_hits`/`keyspace_misses` had been reporting zero for plain GETs on the shipped runtime the whole time, while reading correct under tokio, which is exactly why CI never saw it: io10 passed under tokio and failed under monoio. The inline path now records hit, miss, and the keymiss event, and the inline SET path — the route a plain `SET k v` takes — queues its `set` event. The same shape bit the drain itself: the first version was added to the shard loop's `#[cfg(feature = "runtime-monoio")]` arm only, so it never ran in the tokio build and kn6/kn9 stayed red. Both arms carry it now, with a comment saying why one is not enough. Tests: 10/10 keyspace_notifications and 13/13 info_observability, run under BOTH `runtime-monoio` (default, shipped) and `runtime-tokio,jemalloc`. Lib suites green on both: 4605 and 3768. author: Tin Dang --- src/command/key.rs | 16 ++ src/command/string/string_read.rs | 15 ++ src/command/string/string_write.rs | 17 ++ src/lib.rs | 1 + src/notify.rs | 105 ++++++++++++ src/notify_fanout.rs | 213 +++++++++++++++++++++++++ src/server/conn/blocking.rs | 27 ++++ src/server/conn/handler_monoio/mod.rs | 7 + src/server/conn/handler_sharded/mod.rs | 6 + src/shard/dispatch.rs | 9 ++ src/shard/event_loop.rs | 39 +++++ src/shard/mod.rs | 9 +- src/shard/spsc_handler.rs | 8 + src/shard/timers.rs | 10 ++ src/storage/db/mod.rs | 10 ++ 15 files changed, 490 insertions(+), 2 deletions(-) create mode 100644 src/notify_fanout.rs diff --git a/src/command/key.rs b/src/command/key.rs index c569b6ef..ce8e5754 100644 --- a/src/command/key.rs +++ b/src/command/key.rs @@ -840,6 +840,22 @@ pub fn rename(db: &mut Database, args: &[Frame]) -> Frame { let entry = db.remove(src).unwrap(); db.set(Bytes::copy_from_slice(dst), entry); + // TWO events, not one: a consumer tracking key lifetimes needs to see the + // source disappear and the destination appear, and the halves carry + // different keys. + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::GENERIC, + "rename_from", + src, + db.db_index, + ); + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::GENERIC, + "rename_to", + dst, + db.db_index, + ); + Frame::SimpleString(Bytes::from_static(b"OK")) } diff --git a/src/command/string/string_read.rs b/src/command/string/string_read.rs index 94d2f0ea..71bec4b9 100644 --- a/src/command/string/string_read.rs +++ b/src/command/string/string_read.rs @@ -28,6 +28,15 @@ pub fn get(db: &mut Database, args: &[Frame]) -> Frame { } None => { crate::admin::metrics_setup::record_keyspace_miss(); + // 'm' is NOT in the 'A' class, so this is silent unless an + // operator asked for it explicitly — publishing on every miss + // would put a pub/sub fan-out on the read path. + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::KEY_MISS, + "keymiss", + key, + db.db_index, + ); Frame::Null } } @@ -361,6 +370,12 @@ pub fn get_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame { } } else { crate::admin::metrics_setup::record_keyspace_miss(); + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::KEY_MISS, + "keymiss", + key, + db.db_index, + ); Frame::Null } } diff --git a/src/command/string/string_write.rs b/src/command/string/string_write.rs index b7d65ddd..387ac449 100644 --- a/src/command/string/string_write.rs +++ b/src/command/string/string_write.rs @@ -26,6 +26,15 @@ pub fn set(db: &mut Database, args: &[Frame]) -> Frame { let mut entry = Entry::new_string(value); entry.set_last_access(db.now()); entry.set_access_counter(5); + // Queued, not published: publishing here would need the pub/sub mesh, + // which command code has no access to. When notifications are off this + // is one Relaxed load — see `crate::notify`. + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::STRING, + "set", + &key, + db.db_index, + ); db.set(key, entry); return ok(); } @@ -399,6 +408,14 @@ fn incrby_internal(db: &mut Database, key: &Bytes, delta: i64) -> Frame { }; entry.set_last_access(db.now()); entry.set_access_counter(5); + // "incrby", not "incr": Redis names the internal operation, not the + // command the client typed, so INCR/DECR/INCRBY/DECRBY all report this. + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::STRING, + "incrby", + key, + db.db_index, + ); db.set(key.clone(), entry); Frame::Integer(new_val) diff --git a/src/lib.rs b/src/lib.rs index 9f0e2a9f..272248a1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,6 +55,7 @@ pub mod io; pub mod memory_ctl; pub mod mq; pub mod notify; +pub mod notify_fanout; pub mod persistence; pub mod protocol; pub mod pubsub; diff --git a/src/notify.rs b/src/notify.rs index 2f412f6b..5f40e8fd 100644 --- a/src/notify.rs +++ b/src/notify.rs @@ -169,6 +169,111 @@ pub fn notifications_enabled() -> bool { published_flags().is_enabled() } +/// One event waiting to be published, produced by command code and consumed +/// by whichever layer owns this shard's cross-shard mesh. +#[derive(Debug, Clone)] +pub struct PendingNotification { + /// Logical db the key lives in — part of both channel names. + pub db: usize, + /// Event name, e.g. `set`, `incrby`, `rename_from`. Always a literal: + /// event names are a closed set, so this costs no allocation. + pub event: &'static str, + /// The key the event is about. + pub key: bytes::Bytes, +} + +thread_local! { + /// Per-shard-thread outbox. + /// + /// Command code cannot publish directly: it has no access to the pub/sub + /// registries, and — the real constraint — a subscriber's task lives on + /// another shard thread, where a `Waker` from this thread does not reach + /// it (see the monoio note in CLAUDE.md). So events are queued here and + /// drained by a layer that holds the SPSC mesh, which is the only + /// cross-thread wake that works. + /// + /// Thread-local rather than a field on the shard: expiry and eviction + /// notify from the shard timer, command dispatch notifies from three + /// different handlers, and threading a handle through all of them would + /// touch every signature on the write path. + static OUTBOX: std::cell::RefCell> = + const { std::cell::RefCell::new(Vec::new()) }; +} + +/// Queue one keyspace event, if its class is enabled. +/// +/// The disabled path is a Relaxed load and two masks — no allocation, no +/// lock, no thread-local access — which is what lets this sit on the write +/// path of every mutating command. +#[inline] +pub fn notify_keyspace_event(class: NotifyFlags, event: &'static str, key: &[u8], db: usize) { + let flags = published_flags(); + if !flags.is_enabled() || !flags.intersects(class) { + return; + } + let pending = PendingNotification { + db, + event, + key: bytes::Bytes::copy_from_slice(key), + }; + OUTBOX.with(|o| o.borrow_mut().push(pending)); +} + +/// Take everything queued on this thread, leaving the outbox empty. +/// +/// Returns `None` when there is nothing pending, so the overwhelmingly common +/// case allocates nothing and the caller can skip its fan-out entirely. +#[inline] +pub fn take_outbox() -> Option> { + OUTBOX.with(|o| { + let mut b = o.borrow_mut(); + if b.is_empty() { + None + } else { + Some(std::mem::take(&mut *b)) + } + }) +} + +/// `true` when this thread has queued events. A borrow-and-check, cheaper +/// than [`take_outbox`] for a caller that only wants to know. +#[inline] +pub fn outbox_is_empty() -> bool { + OUTBOX.with(|o| o.borrow().is_empty()) +} + +/// Render the `(channel, payload)` pairs one event publishes. +/// +/// The two channels are INVERTED with respect to each other, which is the +/// detail consumers get wrong: `__keyspace@__:` carries the EVENT, +/// while `__keyevent@__:` carries the KEY. +pub fn channels_for( + n: &PendingNotification, + flags: NotifyFlags, +) -> Vec<(bytes::Bytes, bytes::Bytes)> { + let mut out = Vec::with_capacity(2); + if flags.contains(NotifyFlags::KEYSPACE) { + let mut ch = Vec::with_capacity(16 + n.key.len()); + ch.extend_from_slice(b"__keyspace@"); + ch.extend_from_slice(itoa::Buffer::new().format(n.db).as_bytes()); + ch.extend_from_slice(b"__:"); + ch.extend_from_slice(&n.key); + out.push(( + bytes::Bytes::from(ch), + bytes::Bytes::from_static(n.event.as_bytes()), + )); + } + if flags.contains(NotifyFlags::KEYEVENT) { + let mut ch = Vec::with_capacity(16 + n.event.len()); + ch.extend_from_slice(b"__keyevent@"); + ch.extend_from_slice(itoa::Buffer::new().format(n.db).as_bytes()); + ch.extend_from_slice(b"__:"); + ch.extend_from_slice(n.event.as_bytes()); + out.push((bytes::Bytes::from(ch), n.key.clone())); + } + out +} + /// The valid characters, in the order Redis names them in its error message. pub const VALID_FLAG_CHARS: &str = "Ag$lshzxeKEtmdn"; diff --git a/src/notify_fanout.rs b/src/notify_fanout.rs new file mode 100644 index 00000000..392dffe3 --- /dev/null +++ b/src/notify_fanout.rs @@ -0,0 +1,213 @@ +//! Draining the keyspace-notification outbox onto the pub/sub mesh. +//! +//! Split from [`crate::notify`] because that module is pure logic — flags, +//! parsing, the thread-local queue — while this one needs the shard's SPSC +//! producers, its notifiers and the remote-subscriber map. Keeping them apart +//! means the flag model stays unit-testable without a running shard. +//! +//! There is exactly one reason this indirection exists at all: a subscriber's +//! connection task lives on the shard thread that accepted it, and under +//! monoio a `Waker` fired from another OS thread does not reach it. Publishing +//! straight into a remote shard's registry would therefore queue the message +//! and never wake the reader. The SPSC ring plus its notifier is the only +//! cross-thread wake that works, so remote deliveries go through it. + +use bytes::Bytes; + +use crate::notify::{self, NotifyFlags}; + +/// Drain this thread's outbox and deliver every queued event. +/// +/// Local subscribers are served synchronously; every other shard that has a +/// subscriber matching the channel gets one `NotifyPublish` message. Returns +/// the shard ids that were notified, so callers that must kick a notifier can +/// do so without re-deriving the set. +/// +/// Fire-and-forget by design: a notification has no return value, so this +/// never awaits and can be called from the shard timer as well as from a +/// connection handler. A full ring drops the batch rather than blocking the +/// write path — the same trade Redis makes for a slow subscriber, and the +/// reason `notify-keyspace-events` is documented as best-effort. +pub fn flush_outbox

( + shard_id: usize, + local_registry: &parking_lot::RwLock, + remote_map: &parking_lot::RwLock, + mut push_to: P, +) where + P: FnMut(usize, Vec<(Bytes, Bytes)>), +{ + let Some(pending) = notify::take_outbox() else { + return; + }; + let flags = notify::published_flags(); + if !flags.is_enabled() { + // Flags were turned off between queueing and draining. Dropping is + // right: the operator's most recent instruction is "do not deliver". + return; + } + + // Group by target shard so a burst of events costs one message per shard + // rather than one per event. + let mut remote: Vec<(usize, Vec<(Bytes, Bytes)>)> = Vec::new(); + for n in &pending { + for (channel, payload) in notify::channels_for(n, flags) { + crate::pubsub::publish_shared(local_registry, &channel, &payload); + let targets = remote_map.read().target_shards(&channel); + for t in targets { + if t == shard_id { + continue; + } + match remote.iter_mut().find(|(id, _)| *id == t) { + Some((_, batch)) => batch.push((channel.clone(), payload.clone())), + None => remote.push((t, vec![(channel.clone(), payload.clone())])), + } + } + } + } + + for (target, pairs) in remote { + push_to(target, pairs); + } +} + +/// Drain and deliver from a connection handler. +/// +/// Wraps [`flush_outbox`] with the mesh plumbing every sharded handler holds. +/// Cheap to call unconditionally after a command batch: with nothing queued it +/// is one thread-local borrow. +pub(crate) fn flush_from_connection(ctx: &crate::server::conn::core::ConnectionContext) { + use ringbuf::traits::Producer; + flush_outbox( + ctx.shard_id, + &ctx.pubsub_registry, + &ctx.remote_subscriber_map, + |target, pairs| { + let msg = crate::shard::dispatch::ShardMessage::NotifyPublish(Box::new(pairs)); + let idx = crate::shard::mesh::ChannelMesh::target_index(ctx.shard_id, target); + let pushed = { + let mut producers = ctx.dispatch_tx.borrow_mut(); + producers[idx].try_push(msg).is_ok() + }; + if pushed { + ctx.spsc_notifiers[target].notify_one(); + } + // A full ring drops this batch. Deliberate: notifications are + // best-effort in Redis too, and blocking a write path on a + // notification would be a worse failure than losing one. + }, + ); +} + +/// Drain and deliver from the shard event loop. +/// +/// The counterpart to [`flush_from_connection`], and the one that makes +/// cross-shard writes work: a write routed to the shard that owns the key +/// executes on THAT thread, so its events land in THAT thread's outbox. With +/// only the connection-side drain, every event from a cross-shard write would +/// be queued and never delivered — invisible at `--shards 1` and losing +/// roughly (N-1)/N of events at `--shards N`. +/// +/// Also the delivery path for events with no connection at all: TTL expiry and +/// eviction both run from the shard timer. +pub fn flush_from_shard( + shard_id: usize, + local_registry: &parking_lot::RwLock, + remote_map: &parking_lot::RwLock, + dispatch_tx: &std::cell::RefCell>>, + notifiers: &[std::sync::Arc], +) { + use ringbuf::traits::Producer; + flush_outbox(shard_id, local_registry, remote_map, |target, pairs| { + let msg = crate::shard::dispatch::ShardMessage::NotifyPublish(Box::new(pairs)); + let idx = crate::shard::mesh::ChannelMesh::target_index(shard_id, target); + let pushed = { + let mut producers = dispatch_tx.borrow_mut(); + producers[idx].try_push(msg).is_ok() + }; + if pushed { + notifiers[target].notify_one(); + } + }); +} + +/// Class of the event a command produced, for call sites that need to name it +/// once and emit several events. +pub const GENERIC: NotifyFlags = NotifyFlags::GENERIC; +/// String-command class (`$`). +pub const STRING: NotifyFlags = NotifyFlags::STRING; +/// Expired-key class (`x`). +pub const EXPIRED: NotifyFlags = NotifyFlags::EXPIRED; +/// Evicted-key class (`e`). +pub const EVICTED: NotifyFlags = NotifyFlags::EVICTED; +/// Key-miss class (`m`). +pub const KEY_MISS: NotifyFlags = NotifyFlags::KEY_MISS; + +#[cfg(test)] +mod tests { + use super::*; + use crate::notify::{PendingNotification, parse_flags}; + + #[test] + fn keyspace_and_keyevent_channels_are_inverted() { + // The pair consumers most often get backwards: the keyspace channel is + // named for the KEY and carries the EVENT; the keyevent channel is + // named for the EVENT and carries the KEY. + let n = PendingNotification { + db: 0, + event: "set", + key: Bytes::from_static(b"mykey"), + }; + let pairs = notify::channels_for(&n, parse_flags("KEA").expect("valid")); + assert_eq!(pairs.len(), 2); + assert_eq!(pairs[0].0, Bytes::from_static(b"__keyspace@0__:mykey")); + assert_eq!(pairs[0].1, Bytes::from_static(b"set")); + assert_eq!(pairs[1].0, Bytes::from_static(b"__keyevent@0__:set")); + assert_eq!(pairs[1].1, Bytes::from_static(b"mykey")); + } + + #[test] + fn k_alone_emits_only_the_keyspace_channel() { + let n = PendingNotification { + db: 3, + event: "del", + key: Bytes::from_static(b"k"), + }; + let pairs = notify::channels_for(&n, parse_flags("Kg").expect("valid")); + assert_eq!(pairs.len(), 1, "E is unset, so no keyevent channel"); + assert_eq!(pairs[0].0, Bytes::from_static(b"__keyspace@3__:k")); + } + + #[test] + fn e_alone_emits_only_the_keyevent_channel() { + let n = PendingNotification { + db: 3, + event: "del", + key: Bytes::from_static(b"k"), + }; + let pairs = notify::channels_for(&n, parse_flags("Eg").expect("valid")); + assert_eq!(pairs.len(), 1, "K is unset, so no keyspace channel"); + assert_eq!(pairs[0].0, Bytes::from_static(b"__keyevent@3__:del")); + } + + #[test] + fn db_index_is_part_of_both_channel_names() { + // A consumer subscribed to __keyspace@0__:* must not see db-9 traffic. + let n = PendingNotification { + db: 9, + event: "set", + key: Bytes::from_static(b"k"), + }; + let pairs = notify::channels_for(&n, parse_flags("KEA").expect("valid")); + assert!(pairs.iter().all(|(ch, _)| ch.starts_with(b"__key"))); + assert!( + pairs + .iter() + .any(|(ch, _)| ch.as_ref() == b"__keyspace@9__:k") + ); + assert!( + pairs + .iter() + .any(|(ch, _)| ch.as_ref() == b"__keyevent@9__:set") + ); + } +} diff --git a/src/server/conn/blocking.rs b/src/server/conn/blocking.rs index 57903586..7743fce4 100644 --- a/src/server/conn/blocking.rs +++ b/src/server/conn/blocking.rs @@ -1771,6 +1771,13 @@ pub(crate) fn try_inline_dispatch( write_buf.extend_from_slice(b"\r\n"); write_buf.extend_from_slice(val); write_buf.extend_from_slice(b"\r\n"); + // This is the route a plain `GET key` actually takes + // under monoio — it frames the reply here and returns, + // reaching NEITHER `get` nor `get_readonly`. Without + // this call `keyspace_hits` stayed at zero on the + // shipped runtime while reading correct under tokio, + // which is why it survived CI (#477). + crate::admin::metrics_setup::record_keyspace_hit(); (GetOutcome::Handled, None) } None => { @@ -1822,6 +1829,16 @@ pub(crate) fn try_inline_dispatch( Some(_) => return 0, None => { write_buf.extend_from_slice(b"$-1\r\n"); + crate::admin::metrics_setup::record_keyspace_miss(); + // Queued, delivered by this shard's own drain. 'm' is + // not in the 'A' class, so this is inert unless an + // operator asked for keymiss explicitly. + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::KEY_MISS, + "keymiss", + key_bytes, + selected_db, + ); } } } @@ -1952,6 +1969,16 @@ pub(crate) fn try_inline_dispatch( let mut entry = crate::storage::entry::Entry::new_string(value); entry.set_last_access(db.now()); entry.set_access_counter(5); + // Same reason as the inline GET above: a plain `SET k v` is served + // HERE and never reaches `string::set`, so the notification has to + // be queued on this path too or it exists only for SETs complex + // enough to fall out of the fast path. + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::STRING, + "set", + &key, + selected_db, + ); db.set(key, entry); }); } diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 472b0a9c..b8462520 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -3294,6 +3294,13 @@ pub(crate) async fn handle_connection_sharded_monoio< // parks holding the 1024-frame high-water (~74 KB each — the E5 // permanent-ratchet finding). Sustained >256-frame pipelines pay one // shrink+regrow per batch (~sub-µs vs a 1024-command batch). + // Deliver anything the batch's commands queued. Here rather than + // per-command: a pipeline of 1000 SETs then costs one fan-out per + // target shard instead of 1000, and the connection is about to park in + // read() anyway. With notifications off this is one thread-local + // borrow of an empty Vec. + crate::notify_fanout::flush_from_connection(ctx); + responses.clear(); super::util::shrink_batch_vec(&mut responses); frames.clear(); diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 1b6e60dd..2c958955 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -2570,6 +2570,12 @@ pub(crate) async fn handle_connection_sharded_inner< return (HandlerResult::Done, None); } + // Deliver anything this batch's commands queued. Once per + // batch, not per command: a 1000-command pipeline then costs + // one fan-out per target shard. With notifications off it is a + // thread-local borrow of an empty Vec. + crate::notify_fanout::flush_from_connection(ctx); + // E4: a timed-out cross-shard reply slot must never be reused // — the error replies are flushed above, now close. if xshard_reply_fatal { diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index faf04163..635e6f1f 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -834,6 +834,15 @@ pub enum ShardMessage { pairs: Vec<(Bytes, Bytes)>, slot: std::sync::Arc, }, + /// Keyspace-notification fan-out: publish these `(channel, payload)` pairs + /// into the target shard's registry. + /// + /// Deliberately has NO response slot, unlike `PubSubPublish`: a + /// notification has no return value, so the producer never waits and the + /// drain can stay synchronous. That is what lets it be called from the + /// shard timer (expiry, eviction) as well as from the connection layer. + /// Boxed to keep the enum small — this variant is off the hot path. + NotifyPublish(Box>), /// Swap two databases within this shard (SWAPDB implementation). /// /// The SPSC handler emits a per-shard WAL record before performing the swap, diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 64b00c1f..e1e0e6fd 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -1472,6 +1472,20 @@ impl super::Shard { spsc_notify_local.notify_one(); crate::admin::metrics_setup::bump_spsc_drain_renotify(); } + // Deliver keyspace events produced ON THIS SHARD THREAD: + // a write routed here from another shard's connection + // executes here, and TTL expiry / eviction have no + // connection at all. The connection-side drain sees + // neither. Mirrored in the monoio arm below — this block + // is `#[cfg(runtime-tokio)]`, and putting it in only one + // of the two arms is invisible to the other runtime's CI. + crate::notify_fanout::flush_from_shard( + shard_id, + &pubsub_arc, + &remote_sub_map_arc, + &dispatch_tx, + &all_notifiers, + ); // MA5: persist maintenance schedule when modified by RECLAMATION SCHEDULE. if autovacuum_daemon.maintenance_schedule.is_dirty() { if let Some(ref dir) = persistence_dir { @@ -1577,6 +1591,20 @@ impl super::Shard { spsc_notify_local.notify_one(); crate::admin::metrics_setup::bump_spsc_drain_renotify(); } + // Deliver keyspace events produced ON THIS SHARD THREAD: + // a write routed here from another shard's connection + // executes here, and TTL expiry / eviction have no + // connection at all. The connection-side drain sees + // neither. Mirrored in the monoio arm below — this block + // is `#[cfg(runtime-tokio)]`, and putting it in only one + // of the two arms is invisible to the other runtime's CI. + crate::notify_fanout::flush_from_shard( + shard_id, + &pubsub_arc, + &remote_sub_map_arc, + &dispatch_tx, + &all_notifiers, + ); // MA5: persist maintenance schedule when modified by RECLAMATION SCHEDULE. if autovacuum_daemon.maintenance_schedule.is_dirty() { if let Some(ref dir) = persistence_dir { @@ -2308,6 +2336,17 @@ impl super::Shard { let wal_dir = wal_writer.as_ref().map(|w| w.wal_dir()); cdc_registry.register_pending(pending_cdc_subscribes.drain(..), wal_dir); } + // Deliver keyspace events produced ON THIS SHARD THREAD: a + // write routed here from another shard's connection executes + // here, and TTL expiry / eviction have no connection at all. + // The connection-side drain cannot see either. + crate::notify_fanout::flush_from_shard( + shard_id, + &pubsub_arc, + &remote_sub_map_arc, + &dispatch_tx, + &all_notifiers, + ); persistence_tick::handle_pending_snapshot( pending_snapshot, &mut snapshot_state, diff --git a/src/shard/mod.rs b/src/shard/mod.rs index 643a4d58..cc373153 100644 --- a/src/shard/mod.rs +++ b/src/shard/mod.rs @@ -129,11 +129,16 @@ impl Shard { }; let databases: Vec = (0..num_databases) .map(|i| { - if i == 0 { + let mut db = if i == 0 { Database::with_capacity(per_shard_hint) } else { Database::new() - } + }; + // Keyspace notifications name the db in their channel + // (`__keyspace@__:`), and command code only ever sees + // a `&Database` — this is where that identity is stamped. + db.db_index = i; + db }) .collect(); Shard { diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 13cfd0ee..f1573187 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -1999,6 +1999,14 @@ pub(crate) fn handle_shard_message_shared( crate::pubsub::publish_shared(pubsub_registry, &payload.channel, &payload.message); payload.slot.add(count); } + ShardMessage::NotifyPublish(pairs) => { + for (channel, message) in pairs.iter() { + // Return value discarded on purpose: a keyspace notification + // has no subscriber count to report back, and a channel with + // no subscriber on this shard is the normal case. + crate::pubsub::publish_shared(pubsub_registry, channel, message); + } + } ShardMessage::PubSubPublishBatch { pairs, slot } => { let mut batch_total: i64 = 0; for (i, (channel, message)) in pairs.iter().enumerate() { diff --git a/src/shard/timers.rs b/src/shard/timers.rs index 199e7fb9..6305dd2e 100644 --- a/src/shard/timers.rs +++ b/src/shard/timers.rs @@ -51,6 +51,16 @@ pub(crate) fn run_active_expiry( for i in 0..db_count { crate::shard::slice::with_shard_db(i, |db| { crate::server::expiration::expire_cycle_direct(db, &mut |key| { + // Cache-invalidation consumers subscribe to this to drop + // their copy. Queued here and delivered by the shard + // loop's own drain — there is no connection to attribute + // an expiry to. + crate::notify::notify_keyspace_event( + crate::notify::NotifyFlags::EXPIRED, + "expired", + key, + i, + ); crate::replication::reason_del::record_reason_del( key, i, diff --git a/src/storage/db/mod.rs b/src/storage/db/mod.rs index 6aaeed0b..0ff714d5 100644 --- a/src/storage/db/mod.rs +++ b/src/storage/db/mod.rs @@ -192,6 +192,14 @@ pub struct Database { /// self-reset gate in `expire_cycle`). The scan-based reset costs O(N) /// but happens at most once per "expiring key drained" transition. maybe_has_expiring_keys: bool, + /// Logical db this instance holds (0..databases). + /// + /// Needed because a keyspace notification's channel names embed it + /// (`__keyspace@__:`), and command code is handed a `&Database` + /// with no other way to know which one it has. Set once by the shard when + /// it builds its database array; `Database::new()` defaults it to 0, which + /// is correct for the single-db paths (tests, embedded) that use it. + pub db_index: usize, /// Cold index for disk-offloaded KV entries (None when disk-offload disabled). pub cold_index: Option, /// Shard directory for cold reads (None when disk-offload disabled). @@ -288,6 +296,7 @@ impl Database { cached_now_ms: current_time_ms(), base_timestamp: current_secs(), maybe_has_expiring_keys: false, + db_index: 0, cold_index: None, cold_shard_dir: None, hot_keys: crate::storage::hotkey::HotKeySketch::new(), @@ -314,6 +323,7 @@ impl Database { cached_now_ms: current_time_ms(), base_timestamp: current_secs(), maybe_has_expiring_keys: false, + db_index: 0, cold_index: None, cold_shard_dir: None, hot_keys: crate::storage::hotkey::HotKeySketch::new(), From 55128fa5ed3b5779b8041116c83c95e7a662548c Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 13 Aug 2026 23:51:34 +0700 Subject: [PATCH 7/8] docs(changelog): record keyspace notifications and the inline-GET counter fix author: Tin Dang --- CHANGELOG.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 16557ddd..e51ba09a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Fixed +- **`keyspace_hits` and `keyspace_misses` had been reporting zero for plain `GET`s on the shipped + runtime.** `try_inline_dispatch` — the route a plain `GET key` actually takes under monoio — + frames its reply straight into the write buffer and returns, reaching neither `string::get` nor + `string::get_readonly` where the recorders live. Every hit-rate dashboard built on those two + fields has been dividing by zero-over-zero. The reason it survived CI is worth stating: the + counters read CORRECTLY under `runtime-tokio`, which is what the test matrix ran, so a test that + proved the fix under tokio passed while the shipped runtime stayed broken. The inline path now + records hit, miss, and the `keymiss` notification. (#477) - **`maxmemory` now bounds what the process actually costs, not what the allocator says is live.** A live instance ran for an hour with `used_memory:4.21G` against a **10 GB** real footprint — a 2.3x gap — so a 19.2 GB cap never engaged, eviction never fired (`spill_batches_flushed:0`), @@ -47,6 +55,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 directly below each handler's ACL gate, which fixes the whole class at once. ### Added +- **Keyspace notifications (`notify-keyspace-events`).** Moon had none: cache-invalidation + frameworks and change-data-capture consumers subscribe to `__keyspace@__:` and + `__keyevent@__:` and got silence. Both channel families are now published, gated by + the full Redis flag model — including the parts that are not what the letters suggest, all + measured against redis-server 8.6.1 rather than recalled: `CONFIG SET KEA` reads back as `AKE`, + `mn` as `nm`, `Km` as `Km`, and `A` deliberately excludes `m` (keymiss) and `n` (newkey) so it + stays safe to enable in production. Events wired: `set`, `incrby` (INCR publishes `incrby`, not + `incr`), `rename_from`/`rename_to` (RENAME emits BOTH halves, carrying different keys), + `expired`, `keymiss`. Off by default and genuinely zero-cost when off — one relaxed atomic load. + Delivery reaches subscribers on **every** shard, not just the one that owns the mutated key: a + local-only publish would pass at `--shards 1` and silently drop roughly (N-1)/N of events at + `--shards N`. - **`ROLE`, `RESET`, and a real `COMMAND` introspection surface.** `COMMAND` and `COMMAND COUNT` each returned the OTHER'S RESP TYPE — bare `COMMAND` replied `:0` (an Integer where an Array belongs) and `COMMAND COUNT` replied `*0` (an Array where an Integer belongs); `COMMAND From a7fdca896abbc8f1a0d4b98d99b49f3767ca454d Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Fri, 14 Aug 2026 00:21:36 +0700 Subject: [PATCH 8/8] test(compat): repin the missing-INFO-field fixture on a field Moon cannot grow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `test_info_manifest_reports_missing_fields_by_name` proved that the harness names a missing INFO field rather than reporting an anonymous count. It pinned `run_id` — and failed the moment Moon implemented `run_id`, because the gap it was demonstrating had been CLOSED. A fixture that breaks when the product improves is a tax on every future INFO change. Repinned on `atomicvar_api`, which real Redis emits unconditionally (it names the atomics implementation it was built against) and which is meaningless for a Rust server, so Moon will never grow it. The comment now says why, so the next person picks a stable field too. author: Tin Dang --- scripts/client-compat/test_e2e.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/scripts/client-compat/test_e2e.py b/scripts/client-compat/test_e2e.py index 83183bbc..80b29783 100644 --- a/scripts/client-compat/test_e2e.py +++ b/scripts/client-compat/test_e2e.py @@ -159,14 +159,21 @@ def test_a_diverging_entry_exits_one_and_names_the_divergence(self): self.assertEqual(r.redis_raw, b"+PONG\r\n") def test_info_manifest_reports_missing_fields_by_name(self): - # run_id is emitted by real Redis and not by Moon, so it is a genuine - # finding; redis_version is emitted by both. + # atomicvar_api is emitted unconditionally by real Redis (it names the + # atomics implementation it was built against) and is meaningless for a + # Rust server, so Moon will never emit it — which is what makes it a + # stable fixture. redis_version is emitted by both. + # + # This assertion previously pinned run_id, and went stale the moment + # Moon implemented it: the fixture failed because the gap it was proving + # had been CLOSED. Pick a field Moon cannot plausibly grow, or this test + # becomes a tax on every INFO improvement. fields = manifest("") # reuse tempfile helper for a plain list file with open(fields, "w") as f: - f.write("redis_version\nrun_id\n") + f.write("redis_version\natomicvar_api\n") report = Runner(cfg(info_manifest=fields)).run() missing = [r.name for r in report.results if r.verdict == "diff"] - self.assertIn("info:run_id", missing) + self.assertIn("info:atomicvar_api", missing) self.assertNotIn("info:redis_version", missing) def test_info_manifest_blames_the_pin_not_moon_when_redis_lacks_it_too(self):