From ccb2a50f5dde444f26c423020dce74af74adfe6d Mon Sep 17 00:00:00 2001 From: Omkar Mestry Date: Thu, 16 Jul 2026 11:14:19 +0000 Subject: [PATCH 1/2] Added changes to propogate FLUSHSLOT instead of UNLINK in replication stream in case of ASM slot migration Signed-off-by: Omkar Mestry --- src/cluster.c | 2 +- src/cluster.h | 2 +- src/cluster_legacy.c | 32 ++++++++++++++++++++++++++++---- src/cluster_migrateslots.c | 2 +- 4 files changed, 31 insertions(+), 7 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 98d765f594c..63dc93e7ea3 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1645,7 +1645,7 @@ void clusterCommandFlushslot(client *c) { return; } } - delKeysInSlot(slot, lazy, false, true); + delKeysInSlot(slot, lazy, false, true, false); addReply(c, shared.ok); } diff --git a/src/cluster.h b/src/cluster.h index 92a254cbb8f..61f7f9c30c2 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -149,7 +149,7 @@ int isNodeAvailable(clusterNode *node); long long getNodeReplicationOffset(clusterNode *node); sds aggregateClientOutputBuffer(client *c); void resetClusterStats(void); -unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event); +unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event, bool propagate_as_flushslot); unsigned int propagateSlotDeletionByKeys(unsigned int hashslot); void clusterUpdateState(void); diff --git a/src/cluster_legacy.c b/src/cluster_legacy.c index 01d3b9fb788..a24749b265f 100644 --- a/src/cluster_legacy.c +++ b/src/cluster_legacy.c @@ -3372,7 +3372,7 @@ void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoc for (int j = 0; j < dirty_slots_count; j++) { serverLog(LL_NOTICE, "Deleting keys in dirty slot %d on node %.40s (%s) in shard %.40s", dirty_slots[j], myself->name, humanNodename(myself), myself->shard_id); - delKeysInSlot(dirty_slots[j], server.lazyfree_lazy_server_del, true, false); + delKeysInSlot(dirty_slots[j], server.lazyfree_lazy_server_del, true, false, false); } } } @@ -6865,7 +6865,7 @@ int verifyClusterConfigWithData(void) { in->name, humanNodename(in), in->shard_id, j, slot_owner->name, humanNodename(slot_owner), slot_owner->shard_id); } - delKeysInSlot(j, server.lazyfree_lazy_server_del, true, false); + delKeysInSlot(j, server.lazyfree_lazy_server_del, true, false, false); } } if (update_config) clusterSaveConfigOrDie(1); @@ -7508,9 +7508,29 @@ void removeChannelsInSlot(unsigned int slot) { pubsubShardUnsubscribeAllChannelsInSlot(slot); } +/* Helper function to propagate the CLUSTER FLUSHSLOT command to replicas and AOF during ASM. */ +void propagateAsmFlushSlot(unsigned int hashslot, int lazy) { + robj *argv[4]; + int argc = 3; + argv[0] = shared.cluster; + argv[1] = createStringObject("FLUSHSLOT", 9); + argv[2] = createStringObjectFromLongLong(hashslot); + if (lazy) { + argv[3] = createStringObject("ASYNC", 5); + argc = 4; + } + enterExecutionUnit(1, 0); + alsoPropagate(0, argv, argc, PROPAGATE_REPL | PROPAGATE_AOF, hashslot); + exitExecutionUnit(); + postExecutionUnitOperations(); + decrRefCount(argv[1]); + decrRefCount(argv[2]); + if (lazy) decrRefCount(argv[3]); +} + /* Remove all the keys in the specified hash slot. * The number of removed items is returned. */ -unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event) { +unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event, bool is_asm_path) { if (!countKeysInSlot(hashslot)) return 0; /* We may lose a slot during the pause. We need to track this @@ -7536,7 +7556,7 @@ unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, dbSyncDelete(db, key); } // if is command, skip del propagate - if (propagate_del) propagateDeletion(db, key, lazy, hashslot); + if (propagate_del && !is_asm_path) propagateDeletion(db, key, lazy, hashslot); signalModifiedKey(NULL, db, key); if (send_del_event) { /* In the `cluster flushslot` scenario, the keys are actually deleted so notify everyone. */ @@ -7556,6 +7576,10 @@ unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, kvstoreReleaseHashtableIterator(kvs_di); } + if (propagate_del && is_asm_path) { + propagateAsmFlushSlot(hashslot, lazy); + } + server.server_del_keys_in_slot = 0; serverAssert(server.execution_nesting == before_execution_nesting); return j; diff --git a/src/cluster_migrateslots.c b/src/cluster_migrateslots.c index 7de3845ab88..36e85e76a2a 100644 --- a/src/cluster_migrateslots.c +++ b/src/cluster_migrateslots.c @@ -219,7 +219,7 @@ void delKeysNotOwnedByMyself(list *slot_ranges) { slotRange *range = ln->value; for (int i = range->start_slot; i <= range->end_slot; i++) { if (server.cluster->slots[i] != server.cluster->myself) { - delKeysInSlot(i, 1, true, false); + delKeysInSlot(i, 1, true, false, true); } } } From 8aa898ef15ad6a9e7f7e0be469f23d85f28689e5 Mon Sep 17 00:00:00 2001 From: Omkar Mestry Date: Tue, 21 Jul 2026 12:10:09 +0000 Subject: [PATCH 2/2] Added changes to propogate FLUSHSLOT before keys are deleted. Signed-off-by: Omkar Mestry --- src/cluster_legacy.c | 50 +++++++++++------ tests/unit/cluster/cluster-migrateslots.tcl | 59 +++++++++++++++++++++ tests/unit/moduleapi/cluster.tcl | 2 +- 3 files changed, 94 insertions(+), 17 deletions(-) diff --git a/src/cluster_legacy.c b/src/cluster_legacy.c index a24749b265f..eb33d79cfa0 100644 --- a/src/cluster_legacy.c +++ b/src/cluster_legacy.c @@ -3372,7 +3372,7 @@ void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoc for (int j = 0; j < dirty_slots_count; j++) { serverLog(LL_NOTICE, "Deleting keys in dirty slot %d on node %.40s (%s) in shard %.40s", dirty_slots[j], myself->name, humanNodename(myself), myself->shard_id); - delKeysInSlot(dirty_slots[j], server.lazyfree_lazy_server_del, true, false, false); + delKeysInSlot(dirty_slots[j], server.lazyfree_lazy_server_del, true, false, true); } } } @@ -6865,7 +6865,7 @@ int verifyClusterConfigWithData(void) { in->name, humanNodename(in), in->shard_id, j, slot_owner->name, humanNodename(slot_owner), slot_owner->shard_id); } - delKeysInSlot(j, server.lazyfree_lazy_server_del, true, false, false); + delKeysInSlot(j, server.lazyfree_lazy_server_del, true, false, true); } } if (update_config) clusterSaveConfigOrDie(1); @@ -7509,7 +7509,7 @@ void removeChannelsInSlot(unsigned int slot) { } /* Helper function to propagate the CLUSTER FLUSHSLOT command to replicas and AOF during ASM. */ -void propagateAsmFlushSlot(unsigned int hashslot, int lazy) { +static void propagateFlushSlot(unsigned int hashslot, int lazy) { robj *argv[4]; int argc = 3; argv[0] = shared.cluster; @@ -7519,10 +7519,12 @@ void propagateAsmFlushSlot(unsigned int hashslot, int lazy) { argv[3] = createStringObject("ASYNC", 5); argc = 4; } + enterExecutionUnit(1, 0); alsoPropagate(0, argv, argc, PROPAGATE_REPL | PROPAGATE_AOF, hashslot); exitExecutionUnit(); postExecutionUnitOperations(); + decrRefCount(argv[1]); decrRefCount(argv[2]); if (lazy) decrRefCount(argv[3]); @@ -7530,8 +7532,9 @@ void propagateAsmFlushSlot(unsigned int hashslot, int lazy) { /* Remove all the keys in the specified hash slot. * The number of removed items is returned. */ -unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event, bool is_asm_path) { - if (!countKeysInSlot(hashslot)) return 0; +unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event, bool propagate_as_flushslot) { + unsigned int num_keys = countKeysInSlot(hashslot); + if (!num_keys) return 0; /* We may lose a slot during the pause. We need to track this * state so that we don't assert in propagateNow(). */ @@ -7539,24 +7542,42 @@ unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, unsigned int j = 0; int before_execution_nesting = server.execution_nesting; + /* 1. Propagate CLUSTER FLUSHSLOT upfront before deleting keys on the primary. + * Any keys recreated by module callbacks during notification handling will be + * propagated after FLUSHSLOT and safely survive on both primary and replicas. */ + if (propagate_del && propagate_as_flushslot) { + propagateFlushSlot(hashslot, lazy); + } + for (int i = 0; i < server.dbnum; i++) { - kvstoreHashtableIterator *kvs_di = NULL; - void *next; serverDb *db = server.db[i]; if (db == NULL) continue; - kvs_di = kvstoreGetHashtableIterator(db->keys, hashslot, HASHTABLE_ITER_SAFE); + + /* 2. Collect existing keys in the slot into a temporary snapshot array. + * This avoids iterator invalidation and prevents re-deleting keys created + * during module callback notifications. */ + robj **keys_to_del = zmalloc(sizeof(robj *) * num_keys); + unsigned int keys_count = 0; + + kvstoreHashtableIterator *kvs_di = kvstoreGetHashtableIterator(db->keys, hashslot, HASHTABLE_ITER_SAFE); + void *next; while (kvstoreHashtableIteratorNext(kvs_di, &next)) { - robj *valkey = next; + sds sdskey = objectGetKey((robj *)next); + keys_to_del[keys_count++] = createStringObject(sdskey, sdslen(sdskey)); + } + kvstoreReleaseHashtableIterator(kvs_di); + + /* 3. Delete the snapshot of collected keys. */ + for (unsigned int k = 0; k < keys_count; k++) { + robj *key = keys_to_del[k]; enterExecutionUnit(1, 0); - sds sdskey = objectGetKey(valkey); - robj *key = createStringObject(sdskey, sdslen(sdskey)); if (lazy) { dbAsyncDelete(db, key); } else { dbSyncDelete(db, key); } // if is command, skip del propagate - if (propagate_del && !is_asm_path) propagateDeletion(db, key, lazy, hashslot); + if (propagate_del && !propagate_as_flushslot) propagateDeletion(db, key, lazy, hashslot); signalModifiedKey(NULL, db, key); if (send_del_event) { /* In the `cluster flushslot` scenario, the keys are actually deleted so notify everyone. */ @@ -7573,11 +7594,8 @@ unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, j++; server.dirty++; } - kvstoreReleaseHashtableIterator(kvs_di); - } - if (propagate_del && is_asm_path) { - propagateAsmFlushSlot(hashslot, lazy); + zfree(keys_to_del); } server.server_del_keys_in_slot = 0; diff --git a/tests/unit/cluster/cluster-migrateslots.tcl b/tests/unit/cluster/cluster-migrateslots.tcl index 21916576460..c314adcff18 100644 --- a/tests/unit/cluster/cluster-migrateslots.tcl +++ b/tests/unit/cluster/cluster-migrateslots.tcl @@ -600,6 +600,65 @@ start_cluster 3 3 {tags {logreqres:skip external:skip cluster} overrides {cluste } } + test "Slot Migration cleanup propagates CLUSTER FLUSHSLOT" { + # Populate data before migration + populate 1000 "$16383_slot_tag:" 1000 -2 + + # Attach stream listener to Node 2 + set repl [attach_to_replication_stream_on_connection -2] + + # Perform slot migration + assert_match "OK" [R 2 CLUSTER MIGRATESLOTS SLOTSRANGE 16383 16383 NODE $node0_id] + set jobname [get_job_name 2 16383] + wait_for_migration 0 16383 + + # Keys successfully migrated + assert_match "1000" [R 0 CLUSTER COUNTKEYSINSLOT 16383] + assert_match "0" [R 2 CLUSTER COUNTKEYSINSLOT 16383] + + # Also eventually reflected in replicas + wait_for_countkeysinslot 3 16383 1000 + wait_for_countkeysinslot 5 16383 0 + + # Migration log shows success on both ends + assert {[dict get [get_migration_by_name 0 $jobname] state] eq "success"} + assert {[dict get [get_migration_by_name 2 $jobname] state] eq "success"} + + # Read stream with 5s timeout for CLUSTER FLUSHSLOT + set found_flushslot 0 + set found_individual_dels 0 + set empty_count 0 + while {$empty_count < 50} { + set cmd [read_from_replication_stream $repl] + if {$cmd eq ""} { + incr empty_count + after 100 + continue + } + set empty_count 0 + if {[string match -nocase "*flushslot*" $cmd]} { + set found_flushslot 1 + break + } + if {[string match -nocase "*unlink*" $cmd] || [string match -nocase "*del*" $cmd]} { + set found_individual_dels 1 + } + } + + assert_equal 1 $found_flushslot + assert_equal 0 $found_individual_dels + + close_replication_stream $repl + + # Cleanup for next test + assert_match "OK" [R 0 FLUSHDB SYNC] + assert_match "OK" [R 2 FLUSHDB SYNC] + assert_match "OK" [R 0 SAVE] + assert_match "OK" [R 2 SAVE] + assert_match "OK" [R 0 CLUSTER MIGRATESLOTS SLOTSRANGE 16383 16383 NODE $node2_id] + wait_for_migration 2 16383 + } + proc verify_client_flag {idx flag expected_count} { set clients [split [string trim [R $idx client list]] "\r\n"] set found 0 diff --git a/tests/unit/moduleapi/cluster.tcl b/tests/unit/moduleapi/cluster.tcl index 2342b1cc0ce..e887c08b062 100644 --- a/tests/unit/moduleapi/cluster.tcl +++ b/tests/unit/moduleapi/cluster.tcl @@ -258,8 +258,8 @@ start_cluster 2 2 [list config_lines $modules] { # the {lpush before_deleted count_dels_{4oi}} is a post notification job registered when 'count_dels_{4oi}' was removed assert_replication_stream $repl { + {cluster FLUSHSLOT 16382 ASYNC} {multi} - {unlink count_dels_{4oi}} {keyspace.incr_dels} {lpush before_deleted count_dels_{4oi}} {exec}