fix(redis): delete whole subtree by prefix on group delete
This commit is contained in:
27 files changed
+278
-7
No files matched your search
@@ -1075,6 +1075,66 @@ pub async fn redis_delete_keys_in_db_core(
|
||||
}
|
||||
}
|
||||
|
||||
/// Deletes every key under a key-browser group.
|
||||
///
|
||||
/// The browser only holds the part of a group's subtree it has scanned, so a
|
||||
/// delete driven by the loaded rows silently leaves the rest behind. The group
|
||||
/// is identified by its `SCAN MATCH` pattern instead, and both the direct and
|
||||
/// the cluster path walk the whole keyspace for that pattern.
|
||||
pub async fn redis_delete_keys_by_pattern_in_db_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
db: u32,
|
||||
pattern: &str,
|
||||
) -> Result<u64, String> {
|
||||
ensure_redis_pool(state, connection_id).await?;
|
||||
let pool = state.pool_handle(connection_id).await.ok_or("Not found")?;
|
||||
match &pool {
|
||||
PoolKind::Redis(redis) => match redis.as_ref() {
|
||||
RedisConnection::Direct(con) => {
|
||||
let mut con = con.lock().await;
|
||||
redis_driver::select_db(&mut *con, db).await?;
|
||||
redis_driver::delete_keys_by_pattern(
|
||||
&mut *con,
|
||||
pattern,
|
||||
redis_driver::DELETE_KEYS_BY_PATTERN_SCAN_COUNT,
|
||||
)
|
||||
.await
|
||||
}
|
||||
RedisConnection::Cluster(cluster) => {
|
||||
redis_driver::ensure_cluster_db(db)?;
|
||||
// One node's SCAN only covers its own slots, so the subtree is
|
||||
// scanned node-by-node through the shared cluster cursor and
|
||||
// every key keeps its slot-routed connection, like
|
||||
// `redis_delete_keys_in_db_core`.
|
||||
let mut cursor = 0u64;
|
||||
let mut deleted = 0u64;
|
||||
loop {
|
||||
let page = redis_driver::scan_cluster_keys_batch(
|
||||
cluster,
|
||||
cursor,
|
||||
pattern,
|
||||
redis_driver::DELETE_KEYS_BY_PATTERN_SCAN_COUNT,
|
||||
1,
|
||||
false,
|
||||
)
|
||||
.await?;
|
||||
for key in &page.keys {
|
||||
let raw = redis_driver::redis_key_raw_to_bytes(&key.key_raw)?;
|
||||
let mut con = redis_driver::cluster_key_connection(cluster, &raw).await?;
|
||||
deleted += redis_driver::delete_keys(&mut con, std::slice::from_ref(&raw)).await?;
|
||||
}
|
||||
cursor = page.cursor;
|
||||
if cursor == 0 {
|
||||
return Ok(deleted);
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
_ => Err("Not a Redis connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn redis_flush_db_core(state: &AppState, connection_id: &str, db: u32) -> Result<(), String> {
|
||||
ensure_redis_pool(state, connection_id).await?;
|
||||
let pool = state.pool_handle(connection_id).await.ok_or("Not found")?;
|
||||
|
||||
@@ -3867,6 +3867,40 @@ where
|
||||
cmd.query_async(con).await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// `SCAN COUNT` hint used by [`delete_keys_by_pattern`]; large enough to keep
|
||||
/// the cycle count low, small enough that one cycle stays cheap on the server.
|
||||
pub const DELETE_KEYS_BY_PATTERN_SCAN_COUNT: usize = 500;
|
||||
/// Upper bound on the keys carried by a single `DEL` command.
|
||||
const DELETE_KEYS_BY_PATTERN_DELETE_BATCH: usize = 512;
|
||||
|
||||
/// Deletes every key matching `pattern` with repeated `SCAN` + `DEL` cycles.
|
||||
///
|
||||
/// A group in the key browser is a key *prefix*: the loaded rows are only the
|
||||
/// part of that subtree the browser has scanned so far, so deleting just those
|
||||
/// rows leaves the rest of the subtree behind. `KEYS` or one oversized `DEL`
|
||||
/// would block the server for a large subtree, so keys are discovered with one
|
||||
/// `SCAN` cycle at a time and deleted in bounded batches.
|
||||
pub async fn delete_keys_by_pattern<C>(con: &mut C, pattern: &str, scan_count: usize) -> Result<u64, String>
|
||||
where
|
||||
C: ConnectionLike + Send + Sync + Unpin,
|
||||
{
|
||||
let count = scan_count.max(1);
|
||||
let mut cursor = 0u64;
|
||||
let mut deleted = 0u64;
|
||||
loop {
|
||||
let page = scan_keys_batch_inner(con, cursor, pattern, count, 1, false, false).await?;
|
||||
let keys: Result<Vec<Vec<u8>>, String> =
|
||||
page.keys.iter().map(|key| redis_key_raw_to_bytes(&key.key_raw)).collect();
|
||||
for batch in keys?.chunks(DELETE_KEYS_BY_PATTERN_DELETE_BATCH) {
|
||||
deleted += delete_keys(con, batch).await?;
|
||||
}
|
||||
cursor = page.cursor;
|
||||
if cursor == 0 {
|
||||
return Ok(deleted);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn load_more_collection<C>(
|
||||
con: &mut C,
|
||||
key: &[u8],
|
||||
|
||||
@@ -868,6 +868,7 @@ async fn serve() {
|
||||
.route("/redis/set-keys-ttl", post(routes::redis::set_keys_ttl))
|
||||
.route("/redis/set-keys-expire-at", post(routes::redis::set_keys_expire_at))
|
||||
.route("/redis/delete-keys", post(routes::redis::delete_keys))
|
||||
.route("/redis/delete-keys-by-pattern", post(routes::redis::delete_keys_by_pattern))
|
||||
.route("/redis/flush-db", post(routes::redis::flush_db))
|
||||
.route("/redis/execute-command", post(routes::redis::execute_command))
|
||||
.route("/redis/pubsub/publish", post(routes::redis::publish_message))
|
||||
|
||||
@@ -289,6 +289,14 @@ pub struct RedisKeysRequest {
|
||||
pub key_raws: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RedisKeysByPatternRequest {
|
||||
pub connection_id: String,
|
||||
pub db: u32,
|
||||
pub pattern: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RedisDbRequest {
|
||||
@@ -856,6 +864,22 @@ pub async fn delete_keys(
|
||||
Ok(Json(result))
|
||||
}
|
||||
|
||||
pub async fn delete_keys_by_pattern(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<RedisKeysByPatternRequest>,
|
||||
) -> Result<Json<u64>, AppError> {
|
||||
ensure_writable(&state.app, &req.connection_id, "Delete keys").await?;
|
||||
let result = dbx_core::redis_ops::redis_delete_keys_by_pattern_in_db_core(
|
||||
&state.app,
|
||||
&req.connection_id,
|
||||
req.db,
|
||||
&req.pattern,
|
||||
)
|
||||
.await
|
||||
.map_err(AppError::from)?;
|
||||
Ok(Json(result))
|
||||
}
|
||||
|
||||
pub async fn flush_db(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<RedisDbRequest>,
|
||||
|
||||
Reference in new issue
Block a user