fix(redis): batch SCAN server-side to reduce fetch-all latency
This commit is contained in:
1 parent
08fb5bbf4d
commit
ab5bd1dd25
10 files changed
+204
-49
No files matched your search
@@ -1000,41 +1000,72 @@ pub async fn scan_keys_page<C>(con: &mut C, cursor: u64, pattern: &str, count: u
|
||||
where
|
||||
C: ConnectionLike + Send + Sync + Unpin,
|
||||
{
|
||||
let raw: RedisRawValue = redis::cmd("SCAN")
|
||||
.arg(cursor)
|
||||
.arg("MATCH")
|
||||
.arg(pattern)
|
||||
.arg("COUNT")
|
||||
.arg(count)
|
||||
.query_async(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
scan_keys_batch(con, cursor, pattern, count, 1).await
|
||||
}
|
||||
|
||||
let (next_cursor, keys) = parse_scan_keys(raw)?;
|
||||
let total_keys: u64 = redis::cmd("DBSIZE").query_async(con).await.unwrap_or(0);
|
||||
if keys.is_empty() {
|
||||
return Ok(RedisScanResult { cursor: next_cursor, keys: Vec::new(), total_keys });
|
||||
/// Batch-scan keys with server-side multi-SCAN support.
|
||||
///
|
||||
/// Performs up to `max_iterations` SCAN→TYPE cycles in a single call,
|
||||
/// dramatically reducing frontend↔backend roundtrips when fetching many keys.
|
||||
/// DBSIZE is only called on the first iteration (cursor == 0).
|
||||
pub async fn scan_keys_batch<C>(
|
||||
con: &mut C,
|
||||
cursor: u64,
|
||||
pattern: &str,
|
||||
count: usize,
|
||||
max_iterations: usize,
|
||||
) -> Result<RedisScanResult, String>
|
||||
where
|
||||
C: ConnectionLike + Send + Sync + Unpin,
|
||||
{
|
||||
let iterations = max_iterations.max(1);
|
||||
let total_keys: u64 = if cursor == 0 { redis::cmd("DBSIZE").query_async(con).await.unwrap_or(0) } else { 0 };
|
||||
|
||||
let mut all_keys: Vec<RedisKeyInfo> = Vec::new();
|
||||
let mut current_cursor = cursor;
|
||||
|
||||
for _ in 0..iterations {
|
||||
let raw: RedisRawValue = redis::cmd("SCAN")
|
||||
.arg(current_cursor)
|
||||
.arg("MATCH")
|
||||
.arg(pattern)
|
||||
.arg("COUNT")
|
||||
.arg(count)
|
||||
.query_async(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let (next_cursor, keys) = parse_scan_keys(raw)?;
|
||||
|
||||
if !keys.is_empty() {
|
||||
let mut pipe = redis::pipe();
|
||||
for key in &keys {
|
||||
pipe.cmd("TYPE").arg(key);
|
||||
}
|
||||
let key_types: Vec<String> = pipe.query_async(con).await.unwrap_or_default();
|
||||
|
||||
for (index, key) in keys.iter().enumerate() {
|
||||
let key_type = key_types.get(index).cloned().unwrap_or_else(|| "unknown".to_string());
|
||||
all_keys.push(RedisKeyInfo {
|
||||
key_display: redis_key_bytes_to_display(key),
|
||||
key_raw: redis_key_bytes_to_raw(key),
|
||||
key_type,
|
||||
ttl: -2,
|
||||
size: 0,
|
||||
value_preview: redis_key_value_preview(
|
||||
key_types.get(index).map(String::as_str).unwrap_or("unknown"),
|
||||
),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if next_cursor == 0 {
|
||||
return Ok(RedisScanResult { cursor: 0, keys: all_keys, total_keys });
|
||||
}
|
||||
current_cursor = next_cursor;
|
||||
}
|
||||
|
||||
let mut pipe = redis::pipe();
|
||||
for key in &keys {
|
||||
pipe.cmd("TYPE").arg(key);
|
||||
}
|
||||
let key_types: Vec<String> = pipe.query_async(con).await.unwrap_or_default();
|
||||
|
||||
let mut result = Vec::with_capacity(keys.len());
|
||||
for (index, key) in keys.iter().enumerate() {
|
||||
let key_type = key_types.get(index).cloned().unwrap_or_else(|| "unknown".to_string());
|
||||
result.push(RedisKeyInfo {
|
||||
key_display: redis_key_bytes_to_display(key),
|
||||
key_raw: redis_key_bytes_to_raw(key),
|
||||
key_type,
|
||||
ttl: -2,
|
||||
size: 0,
|
||||
value_preview: redis_key_value_preview(key_types.get(index).map(String::as_str).unwrap_or("unknown")),
|
||||
});
|
||||
}
|
||||
Ok(RedisScanResult { cursor: next_cursor, keys: result, total_keys })
|
||||
Ok(RedisScanResult { cursor: current_cursor, keys: all_keys, total_keys })
|
||||
}
|
||||
|
||||
pub async fn scan_values_page<C>(
|
||||
@@ -1048,7 +1079,8 @@ pub async fn scan_values_page<C>(
|
||||
where
|
||||
C: ConnectionLike + Send + Sync + Unpin,
|
||||
{
|
||||
let total_keys: u64 = redis::cmd("DBSIZE").query_async(con).await.unwrap_or(0);
|
||||
// Only call DBSIZE on the first page (cursor == 0) to avoid redundant work.
|
||||
let total_keys: u64 = if cursor == 0 { redis::cmd("DBSIZE").query_async(con).await.unwrap_or(0) } else { 0 };
|
||||
if query.trim().is_empty() {
|
||||
return Ok(RedisScanResult { cursor, keys: Vec::new(), total_keys });
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use crate::connection::{AppState, PoolKind};
|
||||
use crate::db::redis_driver::{
|
||||
self, RedisCommandResult, RedisConnection, RedisDatabaseInfo, RedisScanResult, RedisValue,
|
||||
self, RedisCommandResult, RedisConnection, RedisDatabaseInfo, RedisKeyInfo, RedisScanResult, RedisValue,
|
||||
};
|
||||
|
||||
async fn ensure_redis_pool(state: &AppState, connection_id: &str) -> Result<(), String> {
|
||||
@@ -33,6 +33,23 @@ pub async fn redis_scan_keys_core(
|
||||
cursor: u64,
|
||||
pattern: &str,
|
||||
count: usize,
|
||||
) -> Result<RedisScanResult, String> {
|
||||
redis_scan_keys_batch_core(state, connection_id, db, cursor, pattern, count, 1).await
|
||||
}
|
||||
|
||||
/// Batch-scan keys with server-side multi-SCAN support.
|
||||
///
|
||||
/// Performs up to `max_iterations` SCAN→TYPE cycles server-side in a single
|
||||
/// API call, dramatically reducing frontend↔backend roundtrips when fetching
|
||||
/// many keys (e.g. "fetch all" in the key browser).
|
||||
pub async fn redis_scan_keys_batch_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
db: u32,
|
||||
cursor: u64,
|
||||
pattern: &str,
|
||||
count: usize,
|
||||
max_iterations: usize,
|
||||
) -> Result<RedisScanResult, String> {
|
||||
ensure_redis_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
@@ -42,11 +59,30 @@ pub async fn redis_scan_keys_core(
|
||||
RedisConnection::Direct(con) => {
|
||||
let mut con = con.lock().await;
|
||||
redis_driver::select_db(&mut *con, db).await?;
|
||||
redis_driver::scan_keys_page(&mut *con, cursor, pattern, count).await
|
||||
redis_driver::scan_keys_batch(&mut *con, cursor, pattern, count, max_iterations).await
|
||||
}
|
||||
RedisConnection::Cluster(cluster) => {
|
||||
redis_driver::ensure_cluster_db(db)?;
|
||||
redis_driver::scan_cluster_keys_page(cluster, cursor, pattern, count).await
|
||||
// Cluster scan already iterates across nodes; for batch mode we
|
||||
// loop the cluster-level scan to accumulate keys server-side.
|
||||
if max_iterations <= 1 {
|
||||
return redis_driver::scan_cluster_keys_page(cluster, cursor, pattern, count).await;
|
||||
}
|
||||
let mut all_keys: Vec<RedisKeyInfo> = Vec::new();
|
||||
let mut current_cursor = cursor;
|
||||
let mut total_keys: u64 = 0;
|
||||
for i in 0..max_iterations {
|
||||
let page = redis_driver::scan_cluster_keys_page(cluster, current_cursor, pattern, count).await?;
|
||||
if i == 0 {
|
||||
total_keys = page.total_keys;
|
||||
}
|
||||
all_keys.extend(page.keys);
|
||||
if page.cursor == 0 {
|
||||
return Ok(RedisScanResult { cursor: 0, keys: all_keys, total_keys });
|
||||
}
|
||||
current_cursor = page.cursor;
|
||||
}
|
||||
Ok(RedisScanResult { cursor: current_cursor, keys: all_keys, total_keys })
|
||||
}
|
||||
},
|
||||
_ => Err("Not a Redis connection".to_string()),
|
||||
|
||||
Reference in new issue
Block a user