feat(info): INFO parity and keyspace notifications

.add/state.json

@@ -1,7 +1,7 @@
 {
   "project": "moon",
   "stage": "production",
-  "active_task": "protocol-error-lifetime",
+  "active_task": "info-observability",
   "active_milestone": "v0-9-client-compat",
   "tasks": {
     "hotpath-lock-quickwins": {
@@ -476,7 +476,7 @@
     }
   },
   "created": "2026-06-11T03:18:21+00:00",
-  "updated": "2026-08-12T04:55:06+00:00",
+  "updated": "2026-08-13T02:05:58+00:00",
   "setup": {
     "locked": true,
     "locked_at": "2026-06-11T03:28:00+00:00",

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@<db>__:<key>` and
+  `__keyevent@<db>__:<event>` 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

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

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<RefCell<_>>`
+/// 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)

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
     }

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::<usize>() {
                 Ok(v) if v > 0 => runtime_config.maxmemory_samples = v,
                 _ => {