From 758e95a889debd9c0bbe6a207b9fa5d665e03a45 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sat, 11 Jul 2026 08:22:11 +0700 Subject: [PATCH] fix(server): INFO # Keyspace lists all dbs across all shards; GRAPH.LIST unions all shards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Listing parity (task #26): two commands silently reported a single shard's local view as if it were the whole server. INFO # Keyspace previously printed one hardcoded `db0:` line holding the SELECTED db's LOCAL-shard key count — `SELECT 2; SET k v; INFO` reported the db-2 count as db0, other dbs were invisible, and at --shards N the other shards' keys were uncounted. Now: - New `ShardMessage::KeyspaceStats` typed-reply scatter (same pattern as GetKeysInSlot/AofFold): each shard returns per-db (keys, expires) from O(#dbs) counter reads — no key iteration. - `coordinate_keyspace_info` (shard/coordinator.rs) sums element-wise across shards, degrades to partial results on a dead channel. - `info_with_keyspace` (command/connection.rs) rewrites the # Keyspace section listing every NON-EMPTY logical db as `db{i}:keys=K,expires=E,avg_ttl=0` (Redis semantics). - Wired on all three dispatch paths: monoio dispatch, tokio handler_sharded dispatch, and handler_single (single shard = truth, no scatter needed). GRAPH.LIST previously listed only the connection shard's graphs (~1/N, since a graph lives on the shard that owns its name). Now scatters a GraphCommand to every shard and returns the sorted, deduplicated union via `merge_graph_list_responses` (skips error/non-array frames so one sick shard degrades instead of failing the command). Wired on both monoio and tokio sharded write paths. Verified: lib tests 4109 pass, fmt + clippy clean (default and runtime-tokio,jemalloc), smoke-tested at --shards 1 and --shards 4 (8 graphs created round-robin all listed; SELECT 2/3/5 writes all reported with cross-shard sums). author: Tin Dang --- CHANGELOG.md | 13 +++++ src/command/connection.rs | 55 +++++++++++++++++++++ src/command/graph/graph_read.rs | 26 ++++++++++ src/command/graph/mod.rs | 30 +++++++++++ src/server/conn/handler_monoio/dispatch.rs | 14 +++++- src/server/conn/handler_monoio/mod.rs | 3 +- src/server/conn/handler_monoio/write.rs | 39 ++++++++++++++- src/server/conn/handler_sharded/dispatch.rs | 14 +++++- src/server/conn/handler_sharded/mod.rs | 2 +- src/server/conn/handler_sharded/write.rs | 39 ++++++++++++++- src/server/conn/handler_single.rs | 13 ++++- src/shard/coordinator.rs | 45 +++++++++++++++++ src/shard/dispatch.rs | 7 +++ src/shard/spsc_handler.rs | 11 +++++ 14 files changed, 300 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e0235570..76874679 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/src/command/connection.rs b/src/command/connection.rs index 694fcc7b..072fa0d0 100644 --- a/src/command/connection.rs +++ b/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(&[]); diff --git a/src/command/graph/graph_read.rs b/src/command/graph/graph_read.rs index 84793ba7..6864cf81 100644 --- a/src/command/graph/graph_read.rs +++ b/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 = 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 = names.into_iter().map(Frame::BulkString).collect(); + Frame::Array(frames.into()) +} + // --------------------------------------------------------------------------- // GRAPH.QUERY, GRAPH.RO_QUERY, GRAPH.EXPLAIN // --------------------------------------------------------------------------- diff --git a/src/command/graph/mod.rs b/src/command/graph/mod.rs index d9414986..a2df4036 100644 --- a/src/command/graph/mod.rs +++ b/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 = 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> { let Frame::Array(outer) = resp else { diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 3a007192..b63aa70a 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/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(), diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index c1036f04..864c7b97 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/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; } diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index d33e66c7..4dd04157 100644 --- a/src/server/conn/handler_monoio/write.rs +++ b/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); diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 41c5fbe3..2925adc7 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/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(), diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index bd66a506..24b74717 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/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; } diff --git a/src/server/conn/handler_sharded/write.rs b/src/server/conn/handler_sharded/write.rs index 6fdc0259..8bd8df30 100644 --- a/src/server/conn/handler_sharded/write.rs +++ b/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); diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 649b062b..2f7e4431 100644 --- a/src/server/conn/handler_single.rs +++ b/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(), diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index 82222a1b..8e96b005 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -1779,6 +1779,51 @@ pub async fn coordinate_dbsize( Frame::Integer(total) } +/// Gather per-db `(keys, expires)` across ALL shards for `INFO # Keyspace`. +/// +/// Element-wise sum of each shard's per-db counter vector (a key lives on +/// exactly one shard, so sums never double-count). Returns the local-only +/// vector at `num_shards == 1` without touching the SPSC mesh. A dead remote +/// reply channel degrades to the partial sum (INFO is diagnostics — prefer +/// an under-count over an error frame). +pub async fn coordinate_keyspace_info( + my_shard: usize, + num_shards: usize, + dispatch_tx: &Rc>>>, + spsc_notifiers: &[Arc], +) -> Vec<(u64, u64)> { + let mut totals: Vec<(u64, u64)> = crate::shard::slice::with_shard(|s| { + s.databases + .iter() + .map(|db| (db.len() as u64, db.expires_count() as u64)) + .collect() + }); + if num_shards <= 1 { + return totals; + } + let mut receivers = Vec::with_capacity(num_shards - 1); + for target in 0..num_shards { + if target == my_shard { + continue; + } + let (reply_tx, reply_rx) = channel::oneshot(); + let msg = ShardMessage::KeyspaceStats { reply_tx }; + let _ = spsc_send(dispatch_tx, my_shard, target, msg, spsc_notifiers).await; + receivers.push(reply_rx); + } + for rx in receivers { + if let Ok(stats) = recv_reply_bounded(rx).await { + for (i, (k, e)) in stats.into_iter().enumerate() { + if let Some(t) = totals.get_mut(i) { + t.0 += k; + t.1 += e; + } + } + } + } + totals +} + /// Coordinate HOTKEYS across all shards: merge per-shard top-K sketches. /// /// Each key lives on exactly one shard, so the merge never has to sum diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index 4169c51d..2622a009 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -514,6 +514,13 @@ pub enum ShardMessage { count: usize, reply_tx: channel::OneshotSender>, }, + /// Per-db `(keys, expires)` counters for ALL logical dbs on this shard — + /// the `INFO # Keyspace` scatter (Redis lists every non-empty db; a + /// local-only or db0-only answer under-reports, same class as the + /// FT.INFO `num_docs` scatter). Reply vector is indexed by db number. + KeyspaceStats { + reply_tx: channel::OneshotSender>, + }, /// Notify shard of slot ownership changes (no-op placeholder for future per-shard caching). SlotOwnershipUpdate { add_slots: Vec, diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index c0d8772b..81f0a176 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -1897,6 +1897,17 @@ pub(crate) fn handle_shard_message_shared( }; let _ = reply_tx.send(keys); } + ShardMessage::KeyspaceStats { reply_tx } => { + // Per-db (keys, expires) for INFO # Keyspace. O(#dbs) counter + // reads — no key iteration. + let stats: Vec<(u64, u64)> = crate::shard::slice::with_shard(|s| { + s.databases + .iter() + .map(|db| (db.len() as u64, db.expires_count() as u64)) + .collect() + }); + let _ = reply_tx.send(stats); + } ShardMessage::SlotOwnershipUpdate { add_slots: _, remove_slots: _,