" | "" }
-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`
+ - `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.
+
+### 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.
---
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 16557ddd6..e51ba09a3 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
diff --git a/scripts/client-compat/test_e2e.py b/scripts/client-compat/test_e2e.py
index 83183bbc0..80b29783b 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):
diff --git a/src/admin/metrics_setup.rs b/src/admin/metrics_setup.rs
index 10bfef9a3..4f931bb2a 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 2568899e9..6119beb36 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 05a840a37..0b95dbf46 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(),
@@ -153,6 +159,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!(
@@ -161,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/command/connection.rs b/src/command/connection.rs
index 2118425b3..6aaf1d2c6 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, &InstanceFacts::default());
+ 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, facts: &InstanceFacts) -> 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");
@@ -194,6 +249,13 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
"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");
@@ -250,7 +312,8 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
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,
@@ -271,6 +334,9 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
// 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
@@ -441,6 +507,33 @@ 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\
+ 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(),
+ 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(),
+ facts.pubsub_channels,
+ facts.pubsub_patterns,
+ );
sections.push_str("\r\n");
// # CPU
@@ -497,7 +590,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 +600,67 @@ 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 {
+ 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 base = match info(db, args) {
- Frame::BulkString(b) => b,
- other => return other,
- };
- let text = String::from_utf8_lossy(&base);
+ let text = info_raw(db, facts);
// 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 +673,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 000000000..e8d88f3cf
--- /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/key.rs b/src/command/key.rs
index c569b6efc..ce8e5754d 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/mod.rs b/src/command/mod.rs
index d05647562..8743f8e88 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/command/string/string_read.rs b/src/command/string/string_read.rs
index 94d2f0eac..71bec4b96 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 b7d65dddf..387ac4496 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 710dab589..272248a12 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -54,6 +54,8 @@ pub mod graph;
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/main.rs b/src/main.rs
index df07d212f..35f06a539 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/notify.rs b/src/notify.rs
new file mode 100644
index 000000000..5f40e8fda
--- /dev/null
+++ b/src/notify.rs
@@ -0,0 +1,440 @@
+//! 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()
+}
+
+/// 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";
+
+/// 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());
+ }
+}
diff --git a/src/notify_fanout.rs b/src/notify_fanout.rs
new file mode 100644
index 000000000..392dffe32
--- /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/pubsub/mod.rs b/src/pubsub/mod.rs
index c390dbfe8..a8212cd6a 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/blocking.rs b/src/server/conn/blocking.rs
index 579035866..7743fce40 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/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs
index 882960611..07730ae61 100644
--- a/src/server/conn/handler_monoio/dispatch.rs
+++ b/src/server/conn/handler_monoio/dispatch.rs
@@ -664,22 +664,27 @@ 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));
+ // 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_facts(db, cmd_args, &keyspace, real_repl.as_deref(), &pubsub_facts)
});
- 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_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs
index 472b0a9c1..b8462520d 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/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs
index a79ba29d1..38a4c1a50 100644
--- a/src/server/conn/handler_sharded/dispatch.rs
+++ b/src/server/conn/handler_sharded/dispatch.rs
@@ -357,21 +357,26 @@ 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));
+ // 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_facts(db, cmd_args, &keyspace, real_repl.as_deref(), &pubsub_facts)
});
- 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/mod.rs b/src/server/conn/handler_sharded/mod.rs
index 1b6e60dd8..2c9589556 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/server/conn/handler_single.rs b/src/server/conn/handler_single.rs
index bb849a7de..f5ce04f02 100644
--- a/src/server/conn/handler_single.rs
+++ b/src/server/conn/handler_single.rs
@@ -1065,20 +1065,31 @@ 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)
+ });
+ // 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(&guard, cmd_args, &keyspace);
+ let resp_frame = conn_cmd::info_with_facts(
+ &guard,
+ cmd_args,
+ &keyspace,
+ real_repl.as_deref(),
+ &pubsub_facts,
+ );
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
diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs
index faf04163f..635e6f1f8 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 64b00c1f5..e1e0e6fde 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 643a4d584..cc3731538 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 13cfd0ee3..f15731873 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 199e7fb91..6305dd2eb 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 6aaeed0b7..0ff714d5a 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(),
diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs
index 3921a9d9d..f026ce0fe 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
new file mode 100644
index 000000000..2d325d30f
--- /dev/null
+++ b/tests/info_observability.rs
@@ -0,0 +1,504 @@
+//! 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()
+ }
+
+ /// 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();
+ 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"
+ );
+}
+
+#[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"
+ );
+}
diff --git a/tests/keyspace_notifications.rs b/tests/keyspace_notifications.rs
new file mode 100644
index 000000000..a11e39bb1
--- /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:?})"
+ );
+}