fix(server): INFO # Keyspace lists all dbs across all shards; GRAPH.LIST unions all shards

CHANGELOG.md

@@ -42,6 +42,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   discovered at failover. Full support (id-pinned record forms + replica
   apply arms + snapshot coverage) is tracked as follow-up work, alongside
   the pre-existing Lua-EVAL and expiry/eviction propagation gaps.
+### Fixed — listing parity: `INFO # Keyspace` all dbs × all shards, `GRAPH.LIST` all shards
+
+- `INFO`'s `# Keyspace` section always printed a single `db0:` line holding
+  the SELECTED db's LOCAL-shard key count — `SELECT 2; SET k v; INFO`
+  reported the db-2 count as db0, every other db was invisible, and at
+  `--shards N` the other shards' keys were uncounted. It now lists every
+  NON-EMPTY logical db (Redis semantics) with `(keys, expires)` summed
+  across all shards via a new `KeyspaceStats` scatter (O(#dbs) counter
+  reads per shard, no key iteration). All three dispatch paths
+  (monoio / tokio sharded / single).
+- `GRAPH.LIST` at `--shards N` listed only the connection shard's graphs
+  (~1/N of them, since a graph lives on the shard that owns its name). It
+  now scatters to every shard and returns the sorted, deduplicated union.
 
 ### Fixed — CLIENT TRACKING dead on the monoio runtime (H-3 reorder regression)
 

src/command/connection.rs

@@ -164,6 +164,11 @@ fn format_memory_human(bytes: u64) -> String {
 /// INFO command handler.
 ///
 /// Returns a BulkString with server info sections matching Redis INFO format.
+///
+/// The `# Keyspace` section reports the CALLING shard's `db` as `db0` only —
+/// 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 {
     use std::fmt::Write as _;
     let mut sections = String::with_capacity(2048);
@@ -329,6 +334,38 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
     Frame::BulkString(Bytes::from(sections))
 }
 
+/// INFO with an externally-gathered `# Keyspace` section: one `(keys,
+/// expires)` entry per logical db, already summed across shards
+/// (`coordinate_keyspace_info`). Lists every NON-EMPTY db like Redis —
+/// previously the section always read `db0:` with the SELECTED db's local
+/// 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 {
+    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);
+    // Rebuild everything up to the fallback "# Keyspace" section, then emit
+    // the accurate per-db lines.
+    let Some(cut) = text.find("# Keyspace\r\n") else {
+        return Frame::BulkString(base);
+    };
+    let mut sections = String::with_capacity(text.len() + keyspace.len() * 32);
+    sections.push_str(&text[..cut]);
+    sections.push_str("# Keyspace\r\n");
+    for (db_idx, (keys, expires)) in keyspace.iter().enumerate() {
+        if *keys > 0 {
+            let _ = write!(
+                sections,
+                "db{db_idx}:keys={keys},expires={expires},avg_ttl=0\r\n"
+            );
+        }
+    }
+    Frame::BulkString(Bytes::from(sections))
+}
+
 /// INFO command handler (read-only variant for RwLock read path).
 ///
 /// Identical to info() -- Database methods used (len, expires_count) are already &self.
@@ -868,6 +905,24 @@ pub fn hello(
 mod tests {
     use super::*;
 
+    #[test]
+    fn info_with_keyspace_lists_every_nonempty_db() {
+        let db = Database::new();
+        let stats = vec![(0u64, 0u64), (0, 0), (7, 2), (0, 0), (1, 0)];
+        let Frame::BulkString(b) = info_with_keyspace(&db, &[], &stats) else {
+            panic!("expected bulk string");
+        };
+        let text = String::from_utf8_lossy(&b);
+        assert!(text.contains("# Keyspace\r\n"));
+        assert!(text.contains("db2:keys=7,expires=2,avg_ttl=0\r\n"));
+        assert!(text.contains("db4:keys=1,expires=0,avg_ttl=0\r\n"));
+        assert!(!text.contains("db0:"), "empty dbs must not be listed");
+        assert!(!text.contains("db1:"));
+        assert!(!text.contains("db3:"));
+        // The fallback single-db section must have been replaced, not doubled.
+        assert_eq!(text.matches("# Keyspace").count(), 1);
+    }
+
     #[test]
     fn test_ping_no_args() {
         let result = ping(&[]);

src/command/graph/graph_read.rs

@@ -520,6 +520,32 @@ pub fn graph_list(store: &GraphStore) -> Frame {
     Frame::Array(frames.into())
 }
 
+/// Merge the local shard's GRAPH.LIST reply with the other shards' replies
+/// (multi-shard scatter — a graph lives on the shard that owns its name, so
+/// a local-only answer listed roughly 1/N of the graphs). Union of names,
+/// sorted for a stable client-visible order. Error frames from remote shards
+/// are skipped (prefer a partial listing over failing the whole command).
+pub fn merge_graph_list_responses(local: Frame, remotes: &[Frame]) -> Frame {
+    let mut names: Vec<bytes::Bytes> = Vec::new();
+    let mut collect = |frame: &Frame| {
+        if let Frame::Array(items) = frame {
+            for item in items.iter() {
+                if let Frame::BulkString(name) = item {
+                    names.push(name.clone());
+                }
+            }
+        }
+    };
+    collect(&local);
+    for r in remotes {
+        collect(r);
+    }
+    names.sort_unstable();
+    names.dedup();
+    let frames: Vec<Frame> = names.into_iter().map(Frame::BulkString).collect();
+    Frame::Array(frames.into())
+}
+
 // ---------------------------------------------------------------------------
 // GRAPH.QUERY, GRAPH.RO_QUERY, GRAPH.EXPLAIN
 // ---------------------------------------------------------------------------

src/command/graph/mod.rs

@@ -11,6 +11,7 @@ pub mod graph_write;
 pub use graph_read::{
     graph_explain, graph_hybrid, graph_info, graph_list, graph_neighbors, graph_profile,
     graph_query, graph_query_or_write, graph_query_write, graph_ro_query, graph_vsearch,
+    merge_graph_list_responses,
 };
 pub use graph_write::{graph_addedge, graph_addnode, graph_create, graph_delete};
 
@@ -225,6 +226,35 @@ mod tests {
         Frame::Array(FrameVec::from_vec(frames))
     }
 
+    #[test]
+    fn merge_graph_list_unions_sorts_dedups_and_skips_errors() {
+        let mk = |names: &[&[u8]]| {
+            let frames: Vec<Frame> = names
+                .iter()
+                .map(|n| Frame::BulkString(Bytes::from(n.to_vec())))
+                .collect();
+            Frame::Array(FrameVec::from_vec(frames))
+        };
+        let local = mk(&[b"g2", b"g1"]);
+        let remotes = vec![
+            mk(&[b"g3", b"g1"]),                           // g1 duplicated across shards
+            Frame::Error(Bytes::from_static(b"ERR boom")), // skipped
+            mk(&[]),
+        ];
+        let merged = merge_graph_list_responses(local, &remotes);
+        let Frame::Array(items) = merged else {
+            panic!("expected array");
+        };
+        let names: Vec<&[u8]> = items
+            .iter()
+            .map(|f| match f {
+                Frame::BulkString(b) => b.as_ref(),
+                _ => panic!("expected bulk string"),
+            })
+            .collect();
+        assert_eq!(names, vec![&b"g1"[..], &b"g2"[..], &b"g3"[..]]);
+    }
+
     /// Decode `[headers, rows, stats]` from a GRAPH.QUERY response into rows of cells.
     fn result_rows(resp: &Frame) -> Vec<Vec<Frame>> {
         let Frame::Array(outer) = resp else {

src/server/conn/handler_monoio/dispatch.rs

@@ -628,7 +628,7 @@ pub(super) fn try_handle_psync(
 
 /// Handle INFO command. Returns `true` if consumed.
 #[inline]
-pub(super) fn try_handle_info(
+pub(super) async fn try_handle_info(
     cmd: &[u8],
     cmd_args: &[Frame],
     conn: &ConnectionState,
@@ -638,8 +638,18 @@ pub(super) fn try_handle_info(
     if !cmd.eq_ignore_ascii_case(b"INFO") {
         return false;
     }
+    // # Keyspace parity: per-db (keys, expires) summed across ALL shards —
+    // previously the section reported the selected db's LOCAL count as db0
+    // (other dbs invisible, other shards uncounted).
+    let keyspace = crate::shard::coordinator::coordinate_keyspace_info(
+        ctx.shard_id,
+        ctx.num_shards,
+        &ctx.dispatch_tx,
+        &ctx.spsc_notifiers,
+    )
+    .await;
     let response_text = crate::shard::slice::with_shard_db(conn.selected_db, |db| {
-        let resp_frame = conn_cmd::info_readonly(db, cmd_args);
+        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(),

src/server/conn/handler_monoio/mod.rs

@@ -902,7 +902,8 @@ pub(crate) async fn handle_connection_sharded_monoio<
                     continue;
                 }
             }
-            if cmd_len == 4 && dispatch::try_handle_info(cmd, cmd_args, &conn, ctx, &mut responses)
+            if cmd_len == 4
+                && dispatch::try_handle_info(cmd, cmd_args, &conn, ctx, &mut responses).await
             {
                 continue;
             }

src/server/conn/handler_monoio/write.rs

@@ -852,8 +852,43 @@ pub(super) async fn try_handle_graph_command(
     // thread-local once ShardSlice is initialized, so non-owner commands MUST
     // hop via ShardMessage::GraphCommand — the shard-side handler dispatches
     // on its own store and drains graph WAL records locally.
-    // GRAPH.LIST has no name argument and stays connection-local (it reports
-    // this shard's graphs only — recorded as a v3 observe delta).
+    // GRAPH.LIST has no name argument: scatter to EVERY shard and union the
+    // names (a local-only answer listed roughly 1/N of the graphs).
+    if ctx.num_shards > 1 && cmd.eq_ignore_ascii_case(b"GRAPH.LIST") {
+        let mut receivers = Vec::with_capacity(ctx.num_shards - 1);
+        for target in 0..ctx.num_shards {
+            if target == ctx.shard_id {
+                continue;
+            }
+            let (reply_tx, reply_rx) = crate::runtime::channel::oneshot();
+            let msg = crate::shard::dispatch::ShardMessage::GraphCommand {
+                command: std::sync::Arc::new(frame.clone()),
+                reply_tx,
+            };
+            let _ = crate::shard::coordinator::spsc_send(
+                &ctx.dispatch_tx,
+                ctx.shard_id,
+                target,
+                msg,
+                &ctx.spsc_notifiers,
+            )
+            .await;
+            receivers.push(reply_rx);
+        }
+        let mut remotes = Vec::with_capacity(receivers.len());
+        for rx in receivers {
+            if let Ok(f) = crate::shard::coordinator::recv_reply_bounded(rx).await {
+                remotes.push(f);
+            }
+        }
+        let local = crate::shard::slice::with_shard(|s| {
+            crate::command::graph::dispatch_graph_read(&s.graph_store, cmd, cmd_args, None)
+        });
+        responses.push(crate::command::graph::merge_graph_list_responses(
+            local, &remotes,
+        ));
+        return true;
+    }
     if ctx.num_shards > 1 && !cmd.eq_ignore_ascii_case(b"GRAPH.LIST") {
         if let Some(name) = cmd_args.first().and_then(extract_bytes) {
             let owner = crate::shard::dispatch::graph_to_shard(&name, ctx.num_shards);

src/server/conn/handler_sharded/dispatch.rs

@@ -308,7 +308,7 @@ pub(super) fn try_handle_replicaof(
 }
 
 /// Handle INFO command. Returns `true` if consumed.
-pub(super) fn try_handle_info(
+pub(super) async fn try_handle_info(
     cmd: &[u8],
     cmd_args: &[Frame],
     conn: &ConnectionState,
@@ -318,9 +318,19 @@ pub(super) fn try_handle_info(
     if !cmd.eq_ignore_ascii_case(b"INFO") {
         return false;
     }
+    // # Keyspace parity: per-db (keys, expires) summed across ALL shards —
+    // previously the section reported the selected db's LOCAL count as db0
+    // (other dbs invisible, other shards uncounted).
+    let keyspace = crate::shard::coordinator::coordinate_keyspace_info(
+        ctx.shard_id,
+        ctx.num_shards,
+        &ctx.dispatch_tx,
+        &ctx.spsc_notifiers,
+    )
+    .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_readonly(db, cmd_args);
+        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(),

src/server/conn/handler_sharded/mod.rs

@@ -867,7 +867,7 @@ pub(crate) async fn handle_connection_sharded_inner<
                     }
 
                     // --- INFO ---
-                    if dispatch::try_handle_info(cmd, cmd_args, &conn, ctx, &mut responses) {
+                    if dispatch::try_handle_info(cmd, cmd_args, &conn, ctx, &mut responses).await {
                         continue;
                     }
 

src/server/conn/handler_sharded/write.rs

@@ -759,8 +759,43 @@ pub(super) async fn try_handle_graph_command(
     // thread-local once ShardSlice is initialized, so non-owner commands MUST
     // hop via ShardMessage::GraphCommand — the shard-side handler dispatches
     // on its own store and drains graph WAL records locally.
-    // GRAPH.LIST has no name argument and stays connection-local (it reports
-    // this shard's graphs only — recorded as a v3 observe delta).
+    // GRAPH.LIST has no name argument: scatter to EVERY shard and union the
+    // names (a local-only answer listed roughly 1/N of the graphs).
+    if ctx.num_shards > 1 && cmd.eq_ignore_ascii_case(b"GRAPH.LIST") {
+        let mut receivers = Vec::with_capacity(ctx.num_shards - 1);
+        for target in 0..ctx.num_shards {
+            if target == ctx.shard_id {
+                continue;
+            }
+            let (reply_tx, reply_rx) = crate::runtime::channel::oneshot();
+            let msg = crate::shard::dispatch::ShardMessage::GraphCommand {
+                command: std::sync::Arc::new(frame.clone()),
+                reply_tx,
+            };
+            let _ = crate::shard::coordinator::spsc_send(
+                &ctx.dispatch_tx,
+                ctx.shard_id,
+                target,
+                msg,
+                &ctx.spsc_notifiers,
+            )
+            .await;
+            receivers.push(reply_rx);
+        }
+        let mut remotes = Vec::with_capacity(receivers.len());
+        for rx in receivers {
+            if let Ok(f) = crate::shard::coordinator::recv_reply_bounded(rx).await {
+                remotes.push(f);
+            }
+        }
+        let local = crate::shard::slice::with_shard(|s| {
+            crate::command::graph::dispatch_graph_read(&s.graph_store, cmd, cmd_args, None)
+        });
+        responses.push(crate::command::graph::merge_graph_list_responses(
+            local, &remotes,
+        ));
+        return true;
+    }
     if ctx.num_shards > 1 && !cmd.eq_ignore_ascii_case(b"GRAPH.LIST") {
         if let Some(name) = cmd_args.first().and_then(extract_bytes) {
             let owner = crate::shard::dispatch::graph_to_shard(&name, ctx.num_shards);

src/server/conn/handler_single.rs

@@ -874,8 +874,19 @@ pub async fn handle_connection(
                         // --- INFO (append replication section) ---
                         if cmd.eq_ignore_ascii_case(b"INFO") {
                             if let Some(ref rs) = repl_state {
+                                // # Keyspace parity: every non-empty db, not
+                                // just the selected one mislabeled as db0.
+                                // Single shard — the local dbs are the truth.
+                                let keyspace: Vec<(u64, u64)> = db
+                                    .iter()
+                                    .map(|d| {
+                                        let g = d.read();
+                                        (g.len() as u64, g.expires_count() as u64)
+                                    })
+                                    .collect();
                                 let guard = db[conn.selected_db].read();
-                                let resp_frame = conn_cmd::info_readonly(&guard, cmd_args);
+                                let resp_frame =
+                                    conn_cmd::info_with_keyspace(&guard, cmd_args, &keyspace);
                                 drop(guard);
                                 let mut response_text = match resp_frame {
                                     Frame::BulkString(b) => String::from_utf8_lossy(&b).to_string(),