Skip to content

Added changes to propogate FLUSHSLOT instead of UNLINK in replication stream in case of ASM slot migration #4191

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
omanges wants to merge 2 commits into
base: unstable
Choose a base branch
from
Open

Added changes to propogate FLUSHSLOT instead of UNLINK in replication stream in case of ASM slot migration #4191

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ccb2a50
Added changes to propogate FLUSHSLOT instead of UNLINK in replication…
omanges Jul 16, 2026
8aa898e
Added changes to propogate FLUSHSLOT before keys are deleted.
omanges Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/cluster.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1645,7 +1645,7 @@ void clusterCommandFlushslot(client *c) {
return;
}
}
delKeysInSlot(slot, lazy, false, true);
delKeysInSlot(slot, lazy, false, true, false);
addReply(c, shared.ok);
}

Expand Down
2 changes: 1 addition & 1 deletion src/cluster.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
32 changes: 28 additions & 4 deletions src/cluster_legacy.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
}
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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) {

@murphyjacob4 murphyjacob4 Jul 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: naming

Suggested change
void propagateAsmFlushSlot(unsigned int hashslot, int lazy) {
void propagateFlushSlot(unsigned int hashslot, int lazy) {

@omanges omanges Jul 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done updated to propagateFlushSlot

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) {

@murphyjacob4 murphyjacob4 Jul 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: match the naming in cluster.h

Suggested change
unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event, bool is_asm_path) {
unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event, bool propagate_as_flushslot) {

@omanges omanges Jul 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done update to propagate_as_flushslot

if (!countKeysInSlot(hashslot)) return 0;

/* We may lose a slot during the pause. We need to track this
Expand All @@ -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);

@valkey-review-bot valkey-review-bot Bot Jul 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Skipping propagateDeletion() here means the per-key execution units at src/cluster_legacy.c:7560-7571 still flush any replicated module/post-notification writes before the later slot-wide FLUSHSLOT is appended. Notification callbacks are allowed to issue writes (tests/modules/keyspace_events.c:84-98), so an ASM-aware module that writes back into the migrating slot from its del callback will now be replayed on replicas/AOF before the flush and then deleted again by the later FLUSHSLOT, even though it survived on the primary. The old per-key propagation kept the delete and those callback side effects ordered together; tests/unit/moduleapi/cluster.tcl:235-266 covers that invariant for slot-change deletions.

Keep the delete propagation ordered with the notification side effects in this path, or fall back to per-key propagation for ASM cleanup.

@murphyjacob4 murphyjacob4 Jul 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah we need to think this through a bit more:

  1. If we send CLUSTER FLUSHSLOT after, it will flush everything, even items that were recreated by modules -> desync
  2. If we send CLUSTER FLUSHSLOT before, we will only flush the items that existed at the start. But if the module recreates a key, that recreated key may also be iterated a second time (since kvstore iterator does not guarantee we don't iterate over newly added items). If the module doesn't recreate on the second call -> desync

I think the best solution is:

  1. At the beginning, swap in a fresh new hashtable for the keys, expires, etc for that slot in the DB
  2. At the same time, propagate the CLUSTER FLUSHSLOT
  3. Now, iterate over each key in the OLD hashtable and trigger the keyspace notifications
  4. Any recreated keys will land in the NEW hashtable and not be reiterated (don't hit the second issue above)

WDYT?

Another option is to force modules to handle FLUSHSLOT as an event like what we do for FLUSHDB, and not trigger any key-level notifications. But that would require module opt-in, and it may be hard for modules to handle this (e.g. search would need to somehow figure out which keys in the index belong to the slot).

In any case - I think the first option is easier for now.

@omanges omanges Jul 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I agree with the suggested solution. Propagating FLUSHSLOT before deleting the keys ensures that any data or metadata created by callback events during key deletion is preserved on replicas rather than being wiped out. I have updated the PR accordingly.

signalModifiedKey(NULL, db, key);
if (send_del_event) {
/* In the `cluster flushslot` scenario, the keys are actually deleted so notify everyone. */
Expand All @@ -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;
Expand Down
2 changes: 1 addition & 1 deletion src/cluster_migrateslots.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
}
Expand Down
Loading