From 74d5753a349398522b62375f75b4357379fdf87b Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Thu, 18 Jun 2026 09:52:29 +0300 Subject: [PATCH 01/19] Add MR_ClusterRefreshTopology for event-driven OSS topology refresh Expose MR_ClusterRefreshTopology(): refreshes the LibMR cluster view in response to a Redis server event instead of requiring a manual REFRESHCLUSTER on every primary. Gated by clusterCtx.isOss so it is a no-op outside OSS cluster mode. A reshard issues CLUSTER SETSLOT per slot, so the topology-change event can fire thousands of times in a burst; refreshing per event would tear down and rebuild every inter-shard connection thousands of times (MR_RefreshClusterData frees the whole cluster). So the refresh is trailing-debounced via an event-loop task: each event only bumps a counter, and a single refresh runs once the counter has been stable for one debounce window. Verified on a live 4-shard reshard: ~8000 events collapse to 1-2 refreshes while cross-shard queries stay complete. Also vendors the RedisModuleEvent_ClusterTopologyChange definitions (event id 20, subevents, info struct) into the local redismodule.h so consumers can subscribe; this mirrors redis/redis#15350 and will be reconciled by the normal redismodule.h sync. Relates-to: MOD-9152, RED-148990 Co-Authored-By: Claude Opus 4.8 --- src/cluster.c | 59 +++++++++++++++++++++++++++++++++++++++++++++++ src/cluster.h | 4 ++++ src/redismodule.h | 12 +++++++++- 3 files changed, 74 insertions(+), 1 deletion(-) diff --git a/src/cluster.c b/src/cluster.c index 373f00e8..02392605 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1323,6 +1323,65 @@ static int MR_ClusterRefresh(RedisModuleCtx *ctx, RedisModuleString **argv, int return REDISMODULE_OK; } +/* Trailing debounce for the automatic topology refresh. A reshard issues + * CLUSTER SETSLOT per slot, so the topology-change event can fire thousands of + * times in quick succession; a naive refresh-per-event would tear down and + * rebuild every inter-shard connection thousands of times (MR_RefreshClusterData + * frees the whole cluster), which is both wasteful and disruptive to in-flight + * cross-shard queries. Instead every event just bumps a counter; an event-loop + * task reschedules itself until the counter stops advancing for one debounce + * window, then does a single refresh. Net effect: one refresh per burst. */ +#define MR_TOPOLOGY_DEBOUNCE_MS 100 + +static unsigned long long clusterTopoEventSeq = 0; /* bumped by the event callback */ +static char clusterTopoDebounceArmed = 0; /* at most one debounce chain */ +static unsigned long long clusterTopoSeenSeq = 0; /* event-loop thread only */ + +/* Runs on the event loop. Refreshes once the event counter has been stable for + * a full debounce window, otherwise waits another window. */ +static void MR_TopoDebounceCheck(void* ctx){ + REDISMODULE_NOT_USED(ctx); + unsigned long long cur = __atomic_load_n(&clusterTopoEventSeq, __ATOMIC_SEQ_CST); + if (cur != clusterTopoSeenSeq) { + /* More events arrived during the window; keep waiting. */ + clusterTopoSeenSeq = cur; + MR_EventLoopAddTaskWithDelay(MR_TopoDebounceCheck, NULL, MR_TOPOLOGY_DEBOUNCE_MS); + return; + } + /* Quiesced. Disarm, then re-check so an event that raced in just now is not + * lost (it would otherwise see the armed flag set and not start a chain). */ + __atomic_clear(&clusterTopoDebounceArmed, __ATOMIC_SEQ_CST); + if (__atomic_load_n(&clusterTopoEventSeq, __ATOMIC_SEQ_CST) != cur) { + if (!__atomic_test_and_set(&clusterTopoDebounceArmed, __ATOMIC_SEQ_CST)) { + clusterTopoSeenSeq = __atomic_load_n(&clusterTopoEventSeq, __ATOMIC_SEQ_CST); + MR_EventLoopAddTaskWithDelay(MR_TopoDebounceCheck, NULL, MR_TOPOLOGY_DEBOUNCE_MS); + } + return; + } + MR_RefreshClusterData(); +} + +/* Runs on the event loop: start a debounce chain (called via the thread-safe + * MR_EventLoopAddTask so the delayed timer is armed from the loop thread). */ +static void MR_TopoDebounceStart(void* ctx){ + REDISMODULE_NOT_USED(ctx); + clusterTopoSeenSeq = __atomic_load_n(&clusterTopoEventSeq, __ATOMIC_SEQ_CST); + MR_EventLoopAddTaskWithDelay(MR_TopoDebounceCheck, NULL, MR_TOPOLOGY_DEBOUNCE_MS); +} + +/* Request an OSS cluster topology refresh, debounced. No-op when not running as + * an OSS cluster (Enterprise/standalone), mirroring the OSS-only registration of + * the REFRESHCLUSTER command, so it is safe to call unconditionally from a Redis + * server-event callback. */ +void MR_ClusterRefreshTopology(void){ + if (!clusterCtx.isOss) return; + __atomic_add_fetch(&clusterTopoEventSeq, 1, __ATOMIC_SEQ_CST); + if (__atomic_test_and_set(&clusterTopoDebounceArmed, __ATOMIC_SEQ_CST)) { + return; /* a debounce chain is already running; it will see the bump */ + } + MR_EventLoopAddTask(MR_TopoDebounceStart, NULL); +} + static int MR_ClusterSet(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){ if (!(IsShortFormClusterSet(argc) || IsLongFormClusterSet(argc))) { RedisModule_ReplyWithError(ctx, "Could not parse cluster set arguments"); diff --git a/src/cluster.h b/src/cluster.h index e2c3995c..9ae4dd40 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -37,6 +37,10 @@ size_t MR_ClusterGetSize(); int MR_ClusterInit(RedisModuleCtx* rctx, char *password); +/* Schedule an OSS cluster topology refresh on the event loop. No-op outside of + * OSS cluster mode. Safe to call from a Redis server-event callback. */ +void MR_ClusterRefreshTopology(void); + size_t MR_ClusterGetSlotByKey(const char* key, size_t len); int MR_ClusterIsMySlot(size_t slot); diff --git a/src/redismodule.h b/src/redismodule.h index 84ae63c0..5417f73a 100644 --- a/src/redismodule.h +++ b/src/redismodule.h @@ -520,7 +520,8 @@ typedef void (*RedisModuleEventLoopOneShotFunc)(void *user_data); #define REDISMODULE_EVENT_KEY 17 #define REDISMODULE_EVENT_CLUSTER_SLOT_MIGRATION 18 #define REDISMODULE_EVENT_CLUSTER_SLOT_MIGRATION_TRIM 19 -#define _REDISMODULE_EVENT_NEXT 20 /* Next event flag, should be updated if a new event added. */ +#define REDISMODULE_EVENT_CLUSTER_TOPOLOGY_CHANGE 20 +#define _REDISMODULE_EVENT_NEXT 21 /* Next event flag, should be updated if a new event added. */ typedef struct RedisModuleEvent { uint64_t id; /* REDISMODULE_EVENT_... defines. */ @@ -639,6 +640,10 @@ static const RedisModuleEvent RedisModuleEvent_ClusterSlotMigrationTrim = { REDISMODULE_EVENT_CLUSTER_SLOT_MIGRATION_TRIM, 1 + }, + RedisModuleEvent_ClusterTopologyChange = { + REDISMODULE_EVENT_CLUSTER_TOPOLOGY_CHANGE, + 1 }; /* Those are values that are used for the 'subevent' callback argument. */ @@ -731,6 +736,11 @@ static const RedisModuleEvent #define REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_TRIM_BACKGROUND 2 #define _REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_TRIM_NEXT 3 +#define REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_STARTUP 0 +#define REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_TOPOLOGY_CHANGED 1 +#define REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_ROLE_CHANGED 2 +#define _REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_NEXT 3 + /* RedisModuleClientInfo flags. */ #define REDISMODULE_CLIENTINFO_FLAG_SSL (1<<0) #define REDISMODULE_CLIENTINFO_FLAG_PUBSUB (1<<1) From f4814605d7ddb51b14849b5faac6d9e2cc9b9e1a Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Wed, 24 Jun 2026 21:32:55 +0300 Subject: [PATCH 02/19] MOD-16382 Consume cluster topology-change reason flags; map-only refresh on in-place reshard MR_ClusterRefreshTopology takes the change_flags reason bitmask from RedisModuleEvent_ClusterTopologyChange and picks the cheapest refresh that stays correct: - FLAG_NODE / FLAG_ROLE / FLAG_STATE (a node joined/left, a role flip, or an OK/FAIL transition -- the set of primaries may have changed) -> full MR_RefreshClusterData (reconnect to the new set of primaries). - FLAG_SLOT only (an in-place reshard: slots moved between primaries we are already connected to) -> new MR_UpdateClusterSlots, which repoints slot->node routing while reusing the existing connections, so in-flight fan-out / cross-shard queries are not aborted and slot-routed queries stay correct mid-reshard. The debounce accumulates the reason flags across the coalesced window and is race-safe: whichever refresh observes a rebuild-worthy flag does the full rebuild, so a membership/role/state change is never masked by the slot-map fast path. MR_UpdateClusterSlots self-upgrades to a full rebuild if an unknown shard appears. Co-Authored-By: Claude Opus 4.8 --- src/cluster.c | 120 +++++++++++++++++++++++++++++++++++++++++++++- src/cluster.h | 12 +++-- src/redismodule.h | 26 ++++++++-- 3 files changed, 149 insertions(+), 9 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 02392605..4545646f 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -955,6 +955,91 @@ static void MR_RefreshClusterData(){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } +/* Update only the slot->node routing from a fresh CLUSTER SLOTS, reusing the + * existing Node structs (and their live connections) instead of tearing the whole + * cluster down and reconnecting. Used for an in-place reshard (only the + * REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_SLOT reason set): the set of primaries is unchanged, + * so only the slot map moved. A full MR_RefreshClusterData would abort in-flight + * cross-shard executions (MR_AbortRunningExecutions) and rebuild every connection + * for nothing, which is exactly what disrupts fan-out queries running during a + * reshard. + * + * In OSS mode CurrCluster->slots is the routing source of truth (the SendMsgType_BySlot + * path), while fan-out (SendMsgType_All) iterates the nodes dict -- both stay correct + * here since the node set is unchanged and only slots[] is repointed. Per-node + * slotRanges are only consumed by the Enterprise CLUSTERSET build and the debug + * RG.INFOCLUSTER reply, so they are intentionally left untouched. + * + * If a primary we have never seen appears, the change was not purely slot-level + * after all (a node entered the serving set); fall back to a full rebuild so the + * new shard is set up and connected through the normal path. */ +static void MR_UpdateClusterSlots(){ + if (!clusterCtx.CurrCluster) { + /* Nothing to update incrementally yet; build from scratch. */ + MR_RefreshClusterData(); + return; + } + + if(!(RedisModule_GetContextFlags(mr_staticCtx) & REDISMODULE_CTX_FLAGS_CLUSTER)){ + return; + } + + RedisModule_Log(mr_staticCtx, "notice", "Got cluster slot-map update (in-place reshard)"); + + RedisModule_ThreadSafeContextLock(mr_staticCtx); + RedisModuleCallReply *allSlotsReply = RedisModule_Call(mr_staticCtx, "cluster", "c", "slots"); + RedisModule_ThreadSafeContextUnlock(mr_staticCtx); + RedisModule_Assert(RedisModule_CallReplyType(allSlotsReply) == REDISMODULE_REPLY_ARRAY); + + /* Repoint the whole slot map from the fresh reply; reused nodes keep their + * connections. Runs on the event-loop thread, same as the dispatch path that + * reads slots[], so no reader can observe a half-updated map. */ + memset(clusterCtx.CurrCluster->slots, 0, sizeof(clusterCtx.CurrCluster->slots)); + + for(size_t i = 0 ; i < RedisModule_CallReplyLength(allSlotsReply) ; ++i){ + RedisModuleCallReply *slotRangeReply = RedisModule_CallReplyArrayElement(allSlotsReply, i); + + RedisModuleCallReply *minSlotReply = RedisModule_CallReplyArrayElement(slotRangeReply, 0); + RedisModule_Assert(RedisModule_CallReplyType(minSlotReply) == REDISMODULE_REPLY_INTEGER); + long long minSlot = RedisModule_CallReplyInteger(minSlotReply); + + RedisModuleCallReply *maxSlotReply = RedisModule_CallReplyArrayElement(slotRangeReply, 1); + RedisModule_Assert(RedisModule_CallReplyType(maxSlotReply) == REDISMODULE_REPLY_INTEGER); + long long maxSlot = RedisModule_CallReplyInteger(maxSlotReply); + + RedisModuleCallReply *nodeDetailsReply = RedisModule_CallReplyArrayElement(slotRangeReply, 2); + RedisModule_Assert(RedisModule_CallReplyType(nodeDetailsReply) == REDISMODULE_REPLY_ARRAY); + RedisModule_Assert(RedisModule_CallReplyLength(nodeDetailsReply) >= 3); + RedisModuleCallReply *nodeidReply = RedisModule_CallReplyArrayElement(nodeDetailsReply, 2); + size_t idLen; + const char* id = RedisModule_CallReplyStringPtr(nodeidReply,&idLen); + + char nodeId[REDISMODULE_NODE_ID_LEN + 1]; + memcpy(nodeId, id, REDISMODULE_NODE_ID_LEN); + nodeId[REDISMODULE_NODE_ID_LEN] = '\0'; + + Node* n = MR_GetNode(nodeId); + if(!n){ + /* A primary we did not know about -> not a pure slot move after all. */ + RedisModule_Log(mr_staticCtx, "notice", + "Slot-map update saw an unknown shard %s; doing a full topology refresh", nodeId); + RedisModule_FreeCallReply(allSlotsReply); + MR_RefreshClusterData(); + return; + } + + if (n->isMe) { + clusterCtx.minSlot = minSlot; + clusterCtx.maxSlot = maxSlot; + } + + for(int k = minSlot ; k <= maxSlot ; ++k){ + clusterCtx.CurrCluster->slots[k] = n; + } + } + RedisModule_FreeCallReply(allSlotsReply); +} + static void GenerateRunId(Cluster* cluster){ RedisModule_GetRandomHexChars(cluster->runId, RUN_ID_SIZE); cluster->runId[RUN_ID_SIZE] = '\0'; @@ -1336,6 +1421,20 @@ static int MR_ClusterRefresh(RedisModuleCtx *ctx, RedisModuleString **argv, int static unsigned long long clusterTopoEventSeq = 0; /* bumped by the event callback */ static char clusterTopoDebounceArmed = 0; /* at most one debounce chain */ static unsigned long long clusterTopoSeenSeq = 0; /* event-loop thread only */ +/* OR of REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_* reasons accumulated across the + * events coalesced into the current debounce window, so we can pick the cheapest + * refresh that still covers every change. */ +static unsigned int clusterTopoPendingChanges = 0; + +/* Reasons that may change the set of primaries we hold connections to -- a node + * joined/left (NODE), a primary/replica role flip (ROLE), or an OK/FAIL transition + * that can bring shards in or out of service (STATE) -- and therefore require a + * full rebuild (reconnect). A SLOT-only change is an in-place reshard between + * primaries we are already connected to, so a connection-preserving slot-map + * update suffices and avoids aborting in-flight cross-shard queries. */ +#define MR_TOPO_REBUILD_FLAGS (REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_NODE | \ + REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_ROLE | \ + REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_STATE) /* Runs on the event loop. Refreshes once the event counter has been stable for * a full debounce window, otherwise waits another window. */ @@ -1358,7 +1457,20 @@ static void MR_TopoDebounceCheck(void* ctx){ } return; } - MR_RefreshClusterData(); + /* Quiesced for real. Consume the accumulated reason flags and pick the cheapest + * refresh that covers them: a full rebuild if the set of primaries may have + * changed (NODE/ROLE/STATE), otherwise a connection-preserving slot-map update + * for an in-place reshard (SLOT only). If a rebuild-worthy event raced in after + * the re-check above, it re-armed its own chain and re-OR'd its bit; whichever + * exchange observes the bit does the full rebuild, and the other chain's refresh + * is a harmless idempotent slot-map update -- so a membership/role/state change + * is never masked by this fast path. */ + unsigned int pending = __atomic_exchange_n(&clusterTopoPendingChanges, 0u, __ATOMIC_SEQ_CST); + if (pending & MR_TOPO_REBUILD_FLAGS) { + MR_RefreshClusterData(); + } else { + MR_UpdateClusterSlots(); + } } /* Runs on the event loop: start a debounce chain (called via the thread-safe @@ -1373,8 +1485,12 @@ static void MR_TopoDebounceStart(void* ctx){ * an OSS cluster (Enterprise/standalone), mirroring the OSS-only registration of * the REFRESHCLUSTER command, so it is safe to call unconditionally from a Redis * server-event callback. */ -void MR_ClusterRefreshTopology(void){ +void MR_ClusterRefreshTopology(int change_flags){ if (!clusterCtx.isOss) return; + /* Accumulate the reason flags before bumping the sequence, so the debounce task + * that eventually fires always sees at least the bits for every event it is + * coalescing. */ + __atomic_or_fetch(&clusterTopoPendingChanges, (unsigned int)change_flags, __ATOMIC_SEQ_CST); __atomic_add_fetch(&clusterTopoEventSeq, 1, __ATOMIC_SEQ_CST); if (__atomic_test_and_set(&clusterTopoDebounceArmed, __ATOMIC_SEQ_CST)) { return; /* a debounce chain is already running; it will see the bump */ diff --git a/src/cluster.h b/src/cluster.h index 9ae4dd40..5d974414 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -37,9 +37,15 @@ size_t MR_ClusterGetSize(); int MR_ClusterInit(RedisModuleCtx* rctx, char *password); -/* Schedule an OSS cluster topology refresh on the event loop. No-op outside of - * OSS cluster mode. Safe to call from a Redis server-event callback. */ -void MR_ClusterRefreshTopology(void); +/* Schedule an OSS cluster topology refresh on the event loop, debounced. + * 'change_flags' is a bitmask of REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_* + * describing the reasons for the change: a NODE/ROLE/STATE reason triggers a full + * rebuild (reconnecting to the possibly-changed set of primaries), while a + * SLOT-only change updates just the slot->node routing and preserves the existing + * connections (so in-flight cross-shard queries are not disrupted by an in-place + * reshard). No-op outside of OSS cluster mode. Safe to call from a Redis + * server-event callback. */ +void MR_ClusterRefreshTopology(int change_flags); size_t MR_ClusterGetSlotByKey(const char* key, size_t len); diff --git a/src/redismodule.h b/src/redismodule.h index 5417f73a..8105e699 100644 --- a/src/redismodule.h +++ b/src/redismodule.h @@ -736,10 +736,8 @@ static const RedisModuleEvent #define REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_TRIM_BACKGROUND 2 #define _REDISMODULE_SUBEVENT_CLUSTER_SLOT_MIGRATION_TRIM_NEXT 3 -#define REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_STARTUP 0 -#define REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_TOPOLOGY_CHANGED 1 -#define REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_ROLE_CHANGED 2 -#define _REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_NEXT 3 +/* RedisModuleEvent_ClusterTopologyChange has no meaningful subevent. */ +#define _REDISMODULE_SUBEVENT_CLUSTER_TOPOLOGY_CHANGE_NEXT 0 /* RedisModuleClientInfo flags. */ #define REDISMODULE_CLIENTINFO_FLAG_SSL (1<<0) @@ -905,6 +903,26 @@ typedef struct RedisModuleClusterSlotMigrationTrimInfo { #define RedisModuleClusterSlotMigrationTrimInfo RedisModuleClusterSlotMigrationTrimInfoV1 +/* Reason flags reported in RedisModuleClusterTopologyChangeInfo.change_flags. + * More than one bit may be set when several changes were collapsed into a + * single (debounced) RedisModuleEvent_ClusterTopologyChange notification. */ +#define REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_SLOT (1<<0) /* Slot ownership changed. */ +#define REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_ROLE (1<<1) /* A node changed its primary/replica role. */ +#define REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_STATE (1<<2) /* The cluster OK/FAIL state changed. */ +#define REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_NODE (1<<3) /* A node joined or left the cluster. */ + +#define REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_INFO_VERSION 1 + +typedef struct RedisModuleClusterTopologyChangeInfo { + uint64_t version; /* Not used since this structure is never passed + from the module to the core right now. Here + for future compatibility. */ + uint64_t change_flags; /* Bitmask of REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_* + reasons that contributed to this notification. */ +} RedisModuleClusterTopologyChangeInfoV1; + +#define RedisModuleClusterTopologyChangeInfo RedisModuleClusterTopologyChangeInfoV1 + typedef enum { REDISMODULE_ACL_LOG_AUTH = 0, /* Authentication failure */ REDISMODULE_ACL_LOG_CMD, /* Command authorization failure */ From 08698d7e693e5a889eac8d9363c090dadf63bf3b Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Sun, 5 Jul 2026 13:45:34 +0300 Subject: [PATCH 03/19] MOD-16382 Decide rebuild-vs-in-place by diffing the primary set, not the reason flags MR_UpdateClusterSlots now reconciles against a fresh CLUSTER SLOTS and rebuilds (which aborts in-flight cross-shard executions) only when the set of slot-serving primaries actually changed -- a primary entered (unknown shard) or left (distinct-count shrank). Otherwise it repoints the slot map in place, preserving connections and in-flight commands. So an over-broad or spurious topology event -- a replica re-pointing, an OK<->FAIL flip, a slotless node joining -- no longer drops in-flight multi-key commands. The debounced handler always calls this single reconcile path; the event's reason flags become advisory (change_flags is no longer consulted). Co-Authored-By: Claude Opus 4.8 --- src/cluster.c | 111 ++++++++++++++++++++++++++------------------------ 1 file changed, 58 insertions(+), 53 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 4545646f..d4dd801f 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -955,27 +955,33 @@ static void MR_RefreshClusterData(){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } -/* Update only the slot->node routing from a fresh CLUSTER SLOTS, reusing the - * existing Node structs (and their live connections) instead of tearing the whole - * cluster down and reconnecting. Used for an in-place reshard (only the - * REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_SLOT reason set): the set of primaries is unchanged, - * so only the slot map moved. A full MR_RefreshClusterData would abort in-flight - * cross-shard executions (MR_AbortRunningExecutions) and rebuild every connection - * for nothing, which is exactly what disrupts fan-out queries running during a - * reshard. +/* Reconcile the LibMR cluster view with a fresh CLUSTER SLOTS, reusing the existing + * Node structs (and their live connections) wherever possible. This is the single + * entry point the debounced topology-change handler calls, regardless of which + * reason flags the event carried, and it decides what to do from the actual + * primary-set delta: * - * In OSS mode CurrCluster->slots is the routing source of truth (the SendMsgType_BySlot - * path), while fan-out (SendMsgType_All) iterates the nodes dict -- both stay correct - * here since the node set is unchanged and only slots[] is repointed. Per-node - * slotRanges are only consumed by the Enterprise CLUSTERSET build and the debug - * RG.INFOCLUSTER reply, so they are intentionally left untouched. + * - If the set of slot-serving primaries is unchanged (only slot ownership moved + * between primaries we are already connected to -- an in-place reshard), we just + * repoint CurrCluster->slots in place. Connections and in-flight cross-shard + * executions are left untouched, so a reshard never drops multi-key commands. + * - If the set of primaries changed (a primary entered or left the serving set -- + * scale-out, scale-in, failover), we escalate to MR_RefreshClusterData, which + * reconnects to the new set and aborts in-flight executions (the client retries). + * + * Deciding by the primary-set delta rather than by the event's reason flags means an + * over-broad or spurious notification (a replica re-pointing, an OK<->FAIL flip, a + * slotless node joining) that does not change the primary set costs only a cheap + * in-place repoint -- never a connection rebuild or a dropped in-flight multi-key + * command. * - * If a primary we have never seen appears, the change was not purely slot-level - * after all (a node entered the serving set); fall back to a full rebuild so the - * new shard is set up and connected through the normal path. */ + * In OSS mode CurrCluster->slots is the routing source of truth (the SendMsgType_BySlot + * path) and fan-out (SendMsgType_All) iterates the nodes dict; both stay correct here. + * Per-node slotRanges are only consumed by the Enterprise CLUSTERSET build and the + * debug RG.INFOCLUSTER reply, so they are intentionally left untouched. */ static void MR_UpdateClusterSlots(){ if (!clusterCtx.CurrCluster) { - /* Nothing to update incrementally yet; build from scratch. */ + /* Nothing to reconcile against yet; build from scratch. */ MR_RefreshClusterData(); return; } @@ -984,8 +990,6 @@ static void MR_UpdateClusterSlots(){ return; } - RedisModule_Log(mr_staticCtx, "notice", "Got cluster slot-map update (in-place reshard)"); - RedisModule_ThreadSafeContextLock(mr_staticCtx); RedisModuleCallReply *allSlotsReply = RedisModule_Call(mr_staticCtx, "cluster", "c", "slots"); RedisModule_ThreadSafeContextUnlock(mr_staticCtx); @@ -996,6 +1000,11 @@ static void MR_UpdateClusterSlots(){ * reads slots[], so no reader can observe a half-updated map. */ memset(clusterCtx.CurrCluster->slots, 0, sizeof(clusterCtx.CurrCluster->slots)); + /* Distinct primaries seen in the fresh reply, so we can detect a primary that + * *left* the serving set after the loop (a primary that *entered* is caught + * inline below as an unknown shard id). */ + mr_dict* seenPrimaries = mr_dictCreate(&mr_dictTypeHeapStrings, NULL); + for(size_t i = 0 ; i < RedisModule_CallReplyLength(allSlotsReply) ; ++i){ RedisModuleCallReply *slotRangeReply = RedisModule_CallReplyArrayElement(allSlotsReply, i); @@ -1020,13 +1029,15 @@ static void MR_UpdateClusterSlots(){ Node* n = MR_GetNode(nodeId); if(!n){ - /* A primary we did not know about -> not a pure slot move after all. */ + /* A primary entered the serving set -> rebuild so it gets connected. */ RedisModule_Log(mr_staticCtx, "notice", - "Slot-map update saw an unknown shard %s; doing a full topology refresh", nodeId); + "Topology reconcile saw a new shard %s; doing a full topology refresh", nodeId); + mr_dictRelease(seenPrimaries); RedisModule_FreeCallReply(allSlotsReply); MR_RefreshClusterData(); return; } + mr_dictAdd(seenPrimaries, nodeId, NULL); /* duplicate ranges for a node are ignored */ if (n->isMe) { clusterCtx.minSlot = minSlot; @@ -1038,6 +1049,19 @@ static void MR_UpdateClusterSlots(){ } } RedisModule_FreeCallReply(allSlotsReply); + + /* If a primary we knew about is no longer serving any slots (scale-in / failover), + * the set of primaries shrank -- rebuild so we drop it and abort in-flight + * executions that assumed it. Additions were handled above, so a distinct count + * smaller than the primaries we hold means exactly a removal. */ + size_t primariesNow = mr_dictSize(seenPrimaries); + mr_dictRelease(seenPrimaries); + if (primariesNow != mr_dictSize(clusterCtx.CurrCluster->nodes)) { + RedisModule_Log(mr_staticCtx, "notice", + "Topology reconcile: the set of primaries changed; doing a full topology refresh"); + MR_RefreshClusterData(); + return; + } } static void GenerateRunId(Cluster* cluster){ @@ -1421,20 +1445,6 @@ static int MR_ClusterRefresh(RedisModuleCtx *ctx, RedisModuleString **argv, int static unsigned long long clusterTopoEventSeq = 0; /* bumped by the event callback */ static char clusterTopoDebounceArmed = 0; /* at most one debounce chain */ static unsigned long long clusterTopoSeenSeq = 0; /* event-loop thread only */ -/* OR of REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_* reasons accumulated across the - * events coalesced into the current debounce window, so we can pick the cheapest - * refresh that still covers every change. */ -static unsigned int clusterTopoPendingChanges = 0; - -/* Reasons that may change the set of primaries we hold connections to -- a node - * joined/left (NODE), a primary/replica role flip (ROLE), or an OK/FAIL transition - * that can bring shards in or out of service (STATE) -- and therefore require a - * full rebuild (reconnect). A SLOT-only change is an in-place reshard between - * primaries we are already connected to, so a connection-preserving slot-map - * update suffices and avoids aborting in-flight cross-shard queries. */ -#define MR_TOPO_REBUILD_FLAGS (REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_NODE | \ - REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_ROLE | \ - REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_STATE) /* Runs on the event loop. Refreshes once the event counter has been stable for * a full debounce window, otherwise waits another window. */ @@ -1457,20 +1467,14 @@ static void MR_TopoDebounceCheck(void* ctx){ } return; } - /* Quiesced for real. Consume the accumulated reason flags and pick the cheapest - * refresh that covers them: a full rebuild if the set of primaries may have - * changed (NODE/ROLE/STATE), otherwise a connection-preserving slot-map update - * for an in-place reshard (SLOT only). If a rebuild-worthy event raced in after - * the re-check above, it re-armed its own chain and re-OR'd its bit; whichever - * exchange observes the bit does the full rebuild, and the other chain's refresh - * is a harmless idempotent slot-map update -- so a membership/role/state change - * is never masked by this fast path. */ - unsigned int pending = __atomic_exchange_n(&clusterTopoPendingChanges, 0u, __ATOMIC_SEQ_CST); - if (pending & MR_TOPO_REBUILD_FLAGS) { - MR_RefreshClusterData(); - } else { - MR_UpdateClusterSlots(); - } + /* Quiesced for real. Reconcile against CLUSTER SLOTS: MR_UpdateClusterSlots + * decides from the actual primary-set delta whether a full rebuild (which aborts + * in-flight executions) is needed or a connection-preserving in-place slot-map + * update suffices. The event's reason flags therefore don't need to be exact, + * and a spurious/over-broad event never drops in-flight multi-key commands. If an + * event raced in after the re-check above it re-armed its own chain; a redundant + * reconcile is a harmless no-op. */ + MR_UpdateClusterSlots(); } /* Runs on the event loop: start a debounce chain (called via the thread-safe @@ -1487,10 +1491,11 @@ static void MR_TopoDebounceStart(void* ctx){ * server-event callback. */ void MR_ClusterRefreshTopology(int change_flags){ if (!clusterCtx.isOss) return; - /* Accumulate the reason flags before bumping the sequence, so the debounce task - * that eventually fires always sees at least the bits for every event it is - * coalescing. */ - __atomic_or_fetch(&clusterTopoPendingChanges, (unsigned int)change_flags, __ATOMIC_SEQ_CST); + /* change_flags is advisory: the debounced refresh reconciles against CLUSTER + * SLOTS and decides rebuild-vs-in-place from the actual primary-set delta, so we + * don't act on the reason bits here. Kept in the signature for callers / future + * use. */ + REDISMODULE_NOT_USED(change_flags); __atomic_add_fetch(&clusterTopoEventSeq, 1, __ATOMIC_SEQ_CST); if (__atomic_test_and_set(&clusterTopoDebounceArmed, __ATOMIC_SEQ_CST)) { return; /* a debounce chain is already running; it will see the bump */ From d7af940a9b261853e4386c9d944ad184cb294881 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Sun, 5 Jul 2026 22:30:17 +0300 Subject: [PATCH 04/19] MOD-16399 Skip the rebuild when CLUSTERSET carries an unchanged topology A cluster-set rebuild drops all inter-shard connections and aborts every in-flight initiator execution with 'cluster topology changed'. The topology is re-broadcast on many events that do not change it (node events, shard reconnects, delivery retries), so multi-shard commands racing such a resend failed spuriously. Compare the incoming long-form command against the stored one (MYID excluded) and keep the current cluster when they match. Short-form commands still rebuild: their topology derives from the server's cluster state, not from the arguments. Co-Authored-By: Claude Fable 5 --- src/cluster.c | 28 ++++++++++++ tests/mr_test_module/pytests/test_network.py | 45 ++++++++++++++++---- 2 files changed, 65 insertions(+), 8 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 373f00e8..0a9ba3bb 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1252,7 +1252,35 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } +/* Returns true when the incoming long-form cluster-set command carries the exact + * topology the current cluster was built from. The MYID slot is excluded: it names + * the receiving shard, not the topology, and is stored as NULL (see CopyClusterSetArgs). */ +static bool ClusterSetCommandIsUnchanged(RedisModuleString** argv, int argc){ + Cluster* cur = clusterCtx.CurrCluster; + if (!cur || !cur->clusterSetCommand || cur->clusterSetCommandSize != argc) + return false; + if (!IsLongFormClusterSet(argc)) + return false; + for (int i = 1 ; i < argc ; ++i) { + if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) + continue; + const char* arg = RedisModule_StringPtrLen(argv[i], NULL); + if (strcmp(arg, cur->clusterSetCommand[i]) != 0) + return false; + } + return true; +} + static int MR_SetClusterData(RedisModuleString** argv, int argc){ + /* The topology is re-broadcast on many events that do not change it (node events, + * shard reconnects, delivery retries). Rebuilding for an identical topology would + * drop all inter-shard connections and abort in-flight executions for nothing. */ + if (ClusterSetCommandIsUnchanged(argv, argc)) { + RedisModule_Log(mr_staticCtx, "notice", + "Got cluster set command with an unchanged topology, skipping the rebuild"); + return REDISMODULE_OK; + } + if(clusterCtx.CurrCluster) MR_ClusterFree(); diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index 09ca7594..73e9ab84 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -210,15 +210,13 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _send_cluster_set(self): - # try to promote to internal connection - promote_internal_client_if_supported(env=self.env) + def _cluster_set_args(self, mock_shard_slot_start=8193): # IPv6 endpoints must be bracketed in host:port strings endpoint_host = '[%s]' % self.host if ':' in self.host else self.host # Build arguments according to MR_SetClusterData parser: # argv[6] => myId, argv[7] => "RANGES", argv[8] => numOfRanges, then repeating: # "SHARD" "SLOTRANGE" "ADDR" ["MASTER"] - args = [ + return [ 'NO-USED', # [1] 'NO-USED', # [2] 'NO-USED', # [3] @@ -229,16 +227,20 @@ def _send_cluster_set(self): '2', # [8] two ranges # Shard 1 (current Redis) - HARDCODED PORT 6379 'SHARD', '1', - 'SLOTRANGE', '0', '8192', + 'SLOTRANGE', '0', str(mock_shard_slot_start - 1), 'ADDR', 'password@%s:6379' % endpoint_host, 'MASTER', # Shard 2 (mock shard) 'SHARD', '2', - 'SLOTRANGE', '8193', '16383', + 'SLOTRANGE', str(mock_shard_slot_start), '16383', 'ADDR', 'password@%s:%d' % (endpoint_host, self.port), 'MASTER' ] - self.env.cmd('MRTESTS.CLUSTERSET', *args) + + def _send_cluster_set(self): + # try to promote to internal connection + promote_internal_client_if_supported(env=self.env) + self.env.cmd('MRTESTS.CLUSTERSET', *self._cluster_set_args()) self.env.cmd('MRTESTS.FORCESHARDSCONNECTION') def __enter__(self): @@ -681,7 +683,11 @@ def testMassiveClusterSet(env, conn): with ShardMock(env, host) as shardMock: for i in range(1000): conn = shardMock.GetConnection(sendHelloResponse=False) - shardMock._send_cluster_set() + # Alternate the slot boundary so every command carries a changed + # topology and forces a rebuild — identical re-sends are a no-op. + promote_internal_client_if_supported(env=env) + env.cmd('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_slot_start=8194 - (i % 2))) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') @MRTestDecorator(skipOnCluster=True) def testMassiveClusterSetFromShard(env, conn): @@ -744,3 +750,26 @@ def testSendMultiRangePerNodeTopology(env, conn): res = env.cmd(*cmd) assert res == 'OK' + + +@MRTestDecorator(skipOnCluster=True) +def testIdenticalClusterSetIsNoOp(env, conn): + for host in _get_hosts(): + with ShardMock(env, host) as shardMock: + conn = shardMock.GetConnection() + + run_id = env.cmd('MRTESTS.INFOCLUSTER')[3] + + # Re-sending the exact same topology must not rebuild the cluster: + # the run id is kept, and so is the live connection to the mock shard. + env.expect('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args()).equal('OK') + env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') + time.sleep(0.5) + env.assertTrue(shardMock.new_conns.empty()) + + # A genuinely different topology (moved slot boundary) must still rebuild. + env.expect('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_slot_start=4096)).equal('OK') + env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') + conn = shardMock.GetConnection() From 6f7b604cc5c0669f59b1369893b8a1256de8048b Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Tue, 7 Jul 2026 10:57:27 +0300 Subject: [PATCH 05/19] Review round: positive naming, length-aware compare, short-form no-op skip - ClusterSetIsNewTopology replaces the double-negative ClusterSetCommandIsUnchanged - Arguments are compared with their lengths (memcmp) instead of strcmp - The short form now skips too: the derived master set is snapshotted into a canonical string (sorted by node id), stored on the cluster, and compared against the snapshot derived for the next short-form command Co-Authored-By: Claude Fable 5 --- src/cluster.c | 119 ++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 100 insertions(+), 19 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 0a9ba3bb..b8180040 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -149,6 +149,7 @@ typedef struct Cluster { Node* slots[NUMBER_OF_SLOTS]; size_t clusterSetCommandSize; char** clusterSetCommand; + char* shortFormTopology; /* canonical master-set snapshot; set only by the short form */ char runId[RUN_ID_SIZE + 1]; }Cluster; @@ -811,6 +812,10 @@ static void MR_ClusterFree(){ MR_FREE(clusterCtx.CurrCluster->clusterSetCommand); } + if(clusterCtx.CurrCluster->shortFormTopology){ + MR_FREE(clusterCtx.CurrCluster->shortFormTopology); + } + MR_FREE(clusterCtx.CurrCluster); clusterCtx.CurrCluster = NULL; clusterCtx.minSlot = 0; @@ -1252,32 +1257,105 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } -/* Returns true when the incoming long-form cluster-set command carries the exact - * topology the current cluster was built from. The MYID slot is excluded: it names - * the receiving shard, not the topology, and is stored as NULL (see CopyClusterSetArgs). */ -static bool ClusterSetCommandIsUnchanged(RedisModuleString** argv, int argc){ +static int compare_snapshot_entries(const void* a, const void* b){ + return strcmp(*(const char**)a, *(const char**)b); +} + +/* Build a canonical snapshot of the master set currently visible through the + * cluster API: one " : " line per master, sorted by + * node id so the nodes-list iteration order does not matter. The password is + * deliberately absent: it comes from the command arguments, which are compared + * separately. Returns NULL when the cluster API is unavailable; the caller owns + * the result. */ +static char* ShortFormTopologySnapshot(void){ + if (RedisModule_GetClusterNodeSlotRanges == NULL) + return NULL; + + size_t numNodes; + char **nodeList = RedisModule_GetClusterNodesList(mr_staticCtx, &numNodes); + if (!nodeList) + return NULL; + + ARR(char*) entries = array_new(char*, numNodes); + for (size_t i = 0; i < numNodes; i++) { + char nodeId[REDISMODULE_NODE_ID_LEN + 1]; // nodeList[i] is not null-terminated + memcpy(nodeId, nodeList[i], REDISMODULE_NODE_ID_LEN); + nodeId[REDISMODULE_NODE_ID_LEN] = '\0'; + + char ip[INET6_ADDRSTRLEN]; // INET6_ADDRSTRLEN includes the closing '\0' + int port, flags; + if (RedisModule_GetClusterNodeInfo(mr_staticCtx, nodeId, ip, NULL, &port, &flags) != REDISMODULE_OK) + continue; + if (!(flags & REDISMODULE_NODE_MASTER)) continue; // Skip replica nodes + + RedisModuleSlotRangeArray *slots = RedisModule_GetClusterNodeSlotRanges(mr_staticCtx, nodeId); + RedisModule_Assert(slots != NULL); + size_t cap = REDISMODULE_NODE_ID_LEN + INET6_ADDRSTRLEN + slots->num_ranges * 12 + 32; + char* entry = MR_ALLOC(cap); + size_t len = snprintf(entry, cap, "%s %s:%d", nodeId, ip, port); + for (size_t j = 0; j < slots->num_ranges; j++) + len += snprintf(entry + len, cap - len, " %d-%d", (int)slots->ranges[j].start, (int)slots->ranges[j].end); + RedisModule_ClusterFreeSlotRanges(mr_staticCtx, slots); + entries = array_append(entries, entry); + } + RedisModule_FreeClusterNodesList(nodeList); + + qsort(entries, array_len(entries), sizeof(*entries), compare_snapshot_entries); + + size_t total = 1; + for (size_t i = 0; i < array_len(entries); i++) + total += strlen(entries[i]) + 1; + char* snapshot = MR_ALLOC(total); + size_t pos = 0; + for (size_t i = 0; i < array_len(entries); i++) { + pos += snprintf(snapshot + pos, total - pos, "%s\n", entries[i]); + MR_FREE(entries[i]); + } + snapshot[pos] = '\0'; + array_free(entries); + return snapshot; +} + +/* Returns true when the incoming cluster-set command carries a topology different + * from the one the current cluster was built from. The long form is compared by + * its arguments; the MYID slot is excluded, as it names the receiving shard rather + * than the topology, and is stored as NULL (see CopyClusterSetArgs). The short form + * derives the topology from the server's cluster state, so on top of its arguments + * (just the AUTH password) it compares `snapshot`, the freshly derived master set. */ +static bool ClusterSetIsNewTopology(RedisModuleString** argv, int argc, const char* snapshot){ Cluster* cur = clusterCtx.CurrCluster; if (!cur || !cur->clusterSetCommand || cur->clusterSetCommandSize != argc) - return false; - if (!IsLongFormClusterSet(argc)) - return false; + return true; for (int i = 1 ; i < argc ; ++i) { - if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) + if (IsLongFormClusterSet(argc) && i == CLUSTERSET_MYID_LONG_FORM_INDEX) continue; - const char* arg = RedisModule_StringPtrLen(argv[i], NULL); - if (strcmp(arg, cur->clusterSetCommand[i]) != 0) - return false; - } - return true; + size_t argLen; + const char* arg = RedisModule_StringPtrLen(argv[i], &argLen); + if (argLen != strlen(cur->clusterSetCommand[i]) || + memcmp(arg, cur->clusterSetCommand[i], argLen) != 0) + return true; + } + if (IsShortFormClusterSet(argc)) + return !snapshot || !cur->shortFormTopology || strcmp(snapshot, cur->shortFormTopology) != 0; + return false; } static int MR_SetClusterData(RedisModuleString** argv, int argc){ + if (!(IsLongFormClusterSet(argc) || IsShortFormClusterSet(argc))) { + RedisModule_Log(mr_staticCtx, "warning", "Could not parse cluster set arguments"); + return REDISMODULE_ERR; + } + + char* snapshot = IsShortFormClusterSet(argc) ? ShortFormTopologySnapshot() : NULL; + /* The topology is re-broadcast on many events that do not change it (node events, * shard reconnects, delivery retries). Rebuilding for an identical topology would * drop all inter-shard connections and abort in-flight executions for nothing. */ - if (ClusterSetCommandIsUnchanged(argv, argc)) { + if (!ClusterSetIsNewTopology(argv, argc, snapshot)) { RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command with an unchanged topology, skipping the rebuild"); + if (snapshot) + MR_FREE(snapshot); return REDISMODULE_OK; } @@ -1287,12 +1365,15 @@ static int MR_SetClusterData(RedisModuleString** argv, int argc){ if (IsLongFormClusterSet(argc)) { SetClusterDataLongForm(argv, argc); return REDISMODULE_OK; - } else if (IsShortFormClusterSet(argc)) { - return SetClusterDataShortForm(argv, argc); - } else { - RedisModule_Log(mr_staticCtx, "warning", "Could not parse cluster set arguments"); - return REDISMODULE_ERR; } + + int res = SetClusterDataShortForm(argv, argc); + if (res == REDISMODULE_OK && clusterCtx.CurrCluster) { + clusterCtx.CurrCluster->shortFormTopology = snapshot; // now owned by the cluster + } else if (snapshot) { + MR_FREE(snapshot); + } + return res; } /* runs in the event loop so its safe to update cluster From b35bab108258f83a61af2bab6d2e0b4bbf15c4ed Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 13 Jul 2026 12:00:02 +0300 Subject: [PATCH 06/19] MOD-16382 Drop the topology-refresh debounce; reconcile per event Review feedback (keep it as simple as possible): every topology-change event now schedules one reconcile task directly on the event loop. MR_UpdateClusterSlots already makes a redundant reconcile harmless (one CLUSTER SLOTS read, connection-preserving unless the primary set changed), so the debounce only saved reconcile churn during legacy per-slot resharding bursts, at the cost of delaying convergence by the quiet window. Co-Authored-By: Claude Opus 4.8 --- src/cluster.c | 78 +++++++++++---------------------------------------- src/cluster.h | 10 +++---- 2 files changed, 21 insertions(+), 67 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index d4dd801f..1191666b 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -957,7 +957,7 @@ static void MR_RefreshClusterData(){ /* Reconcile the LibMR cluster view with a fresh CLUSTER SLOTS, reusing the existing * Node structs (and their live connections) wherever possible. This is the single - * entry point the debounced topology-change handler calls, regardless of which + * entry point the topology-change handler calls, regardless of which * reason flags the event carried, and it decides what to do from the actual * primary-set delta: * @@ -1432,75 +1432,29 @@ static int MR_ClusterRefresh(RedisModuleCtx *ctx, RedisModuleString **argv, int return REDISMODULE_OK; } -/* Trailing debounce for the automatic topology refresh. A reshard issues - * CLUSTER SETSLOT per slot, so the topology-change event can fire thousands of - * times in quick succession; a naive refresh-per-event would tear down and - * rebuild every inter-shard connection thousands of times (MR_RefreshClusterData - * frees the whole cluster), which is both wasteful and disruptive to in-flight - * cross-shard queries. Instead every event just bumps a counter; an event-loop - * task reschedules itself until the counter stops advancing for one debounce - * window, then does a single refresh. Net effect: one refresh per burst. */ -#define MR_TOPOLOGY_DEBOUNCE_MS 100 - -static unsigned long long clusterTopoEventSeq = 0; /* bumped by the event callback */ -static char clusterTopoDebounceArmed = 0; /* at most one debounce chain */ -static unsigned long long clusterTopoSeenSeq = 0; /* event-loop thread only */ - -/* Runs on the event loop. Refreshes once the event counter has been stable for - * a full debounce window, otherwise waits another window. */ -static void MR_TopoDebounceCheck(void* ctx){ +/* Runs on the event loop. MR_UpdateClusterSlots decides from the actual + * primary-set delta whether a full rebuild (which aborts in-flight executions) + * is needed or a connection-preserving in-place slot-map update suffices, so a + * redundant event costs one CLUSTER SLOTS read and never drops in-flight + * multi-key commands. */ +static void MR_TopoRefreshTask(void* ctx){ REDISMODULE_NOT_USED(ctx); - unsigned long long cur = __atomic_load_n(&clusterTopoEventSeq, __ATOMIC_SEQ_CST); - if (cur != clusterTopoSeenSeq) { - /* More events arrived during the window; keep waiting. */ - clusterTopoSeenSeq = cur; - MR_EventLoopAddTaskWithDelay(MR_TopoDebounceCheck, NULL, MR_TOPOLOGY_DEBOUNCE_MS); - return; - } - /* Quiesced. Disarm, then re-check so an event that raced in just now is not - * lost (it would otherwise see the armed flag set and not start a chain). */ - __atomic_clear(&clusterTopoDebounceArmed, __ATOMIC_SEQ_CST); - if (__atomic_load_n(&clusterTopoEventSeq, __ATOMIC_SEQ_CST) != cur) { - if (!__atomic_test_and_set(&clusterTopoDebounceArmed, __ATOMIC_SEQ_CST)) { - clusterTopoSeenSeq = __atomic_load_n(&clusterTopoEventSeq, __ATOMIC_SEQ_CST); - MR_EventLoopAddTaskWithDelay(MR_TopoDebounceCheck, NULL, MR_TOPOLOGY_DEBOUNCE_MS); - } - return; - } - /* Quiesced for real. Reconcile against CLUSTER SLOTS: MR_UpdateClusterSlots - * decides from the actual primary-set delta whether a full rebuild (which aborts - * in-flight executions) is needed or a connection-preserving in-place slot-map - * update suffices. The event's reason flags therefore don't need to be exact, - * and a spurious/over-broad event never drops in-flight multi-key commands. If an - * event raced in after the re-check above it re-armed its own chain; a redundant - * reconcile is a harmless no-op. */ MR_UpdateClusterSlots(); } -/* Runs on the event loop: start a debounce chain (called via the thread-safe - * MR_EventLoopAddTask so the delayed timer is armed from the loop thread). */ -static void MR_TopoDebounceStart(void* ctx){ - REDISMODULE_NOT_USED(ctx); - clusterTopoSeenSeq = __atomic_load_n(&clusterTopoEventSeq, __ATOMIC_SEQ_CST); - MR_EventLoopAddTaskWithDelay(MR_TopoDebounceCheck, NULL, MR_TOPOLOGY_DEBOUNCE_MS); -} - -/* Request an OSS cluster topology refresh, debounced. No-op when not running as - * an OSS cluster (Enterprise/standalone), mirroring the OSS-only registration of - * the REFRESHCLUSTER command, so it is safe to call unconditionally from a Redis +/* Request an OSS cluster topology refresh; every event schedules one reconcile + * on the event loop. No-op when not running as an OSS cluster + * (Enterprise/standalone), mirroring the OSS-only registration of the + * REFRESHCLUSTER command, so it is safe to call unconditionally from a Redis * server-event callback. */ void MR_ClusterRefreshTopology(int change_flags){ if (!clusterCtx.isOss) return; - /* change_flags is advisory: the debounced refresh reconciles against CLUSTER - * SLOTS and decides rebuild-vs-in-place from the actual primary-set delta, so we - * don't act on the reason bits here. Kept in the signature for callers / future - * use. */ + /* change_flags is advisory: the refresh reconciles against CLUSTER SLOTS + * and decides rebuild-vs-in-place from the actual primary-set delta, so we + * don't act on the reason bits here. Kept in the signature for callers / + * future use. */ REDISMODULE_NOT_USED(change_flags); - __atomic_add_fetch(&clusterTopoEventSeq, 1, __ATOMIC_SEQ_CST); - if (__atomic_test_and_set(&clusterTopoDebounceArmed, __ATOMIC_SEQ_CST)) { - return; /* a debounce chain is already running; it will see the bump */ - } - MR_EventLoopAddTask(MR_TopoDebounceStart, NULL); + MR_EventLoopAddTask(MR_TopoRefreshTask, NULL); } static int MR_ClusterSet(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){ diff --git a/src/cluster.h b/src/cluster.h index 5d974414..31594018 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -37,11 +37,11 @@ size_t MR_ClusterGetSize(); int MR_ClusterInit(RedisModuleCtx* rctx, char *password); -/* Schedule an OSS cluster topology refresh on the event loop, debounced. - * 'change_flags' is a bitmask of REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_* - * describing the reasons for the change: a NODE/ROLE/STATE reason triggers a full - * rebuild (reconnecting to the possibly-changed set of primaries), while a - * SLOT-only change updates just the slot->node routing and preserves the existing +/* Schedule an OSS cluster topology refresh on the event loop. 'change_flags' + * is a bitmask of REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_* reasons and is + * advisory only: the refresh reconciles against CLUSTER SLOTS and rebuilds the + * connections only when the set of slot-serving primaries actually changed, + * otherwise it updates just the slot->node routing and preserves the existing * connections (so in-flight cross-shard queries are not disrupted by an in-place * reshard). No-op outside of OSS cluster mode. Safe to call from a Redis * server-event callback. */ From d3830c16c33a8116219ce35344f4514069a79b86 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 13 Jul 2026 12:09:20 +0300 Subject: [PATCH 07/19] MOD-16382 Rebuild when a known primary changes its address The in-place reconcile matched primaries by node id only, so a primary restarting at a new address with the same id (a pod reschedule that keeps nodes.conf) kept its stale ip/port and the reconnect loop redialed the dead address forever. Compare the address from the fresh CLUSTER SLOTS reply (port via RedisModule_GetClusterNodeInfo, matching MR_RefreshClusterData) and escalate to a full rebuild on mismatch. Co-Authored-By: Claude Opus 4.8 --- src/cluster.c | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/src/cluster.c b/src/cluster.c index 1191666b..55207e2a 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1037,6 +1037,29 @@ static void MR_UpdateClusterSlots(){ MR_RefreshClusterData(); return; } + + /* Same id but a new address (the node restarted elsewhere, e.g. a pod + * reschedule that kept nodes.conf): the cached connection would redial + * the old address forever, so rebuild to reconnect. The port is taken + * from RedisModule_GetClusterNodeInfo for the same reason as in + * MR_RefreshClusterData (CLUSTER SLOTS reports the non-TLS port, see + * redis/redis#12233). */ + RedisModuleCallReply *nodeipReply = RedisModule_CallReplyArrayElement(nodeDetailsReply, 0); + size_t ipLen; + const char* ip = RedisModule_CallReplyStringPtr(nodeipReply, &ipLen); + int port = 0; + RedisModule_ThreadSafeContextLock(mr_staticCtx); + RedisModule_GetClusterNodeInfo(mr_staticCtx, nodeId, NULL, NULL, &port, NULL); + RedisModule_ThreadSafeContextUnlock(mr_staticCtx); + if (strlen(n->ip) != ipLen || memcmp(n->ip, ip, ipLen) != 0 || + (port != 0 && n->port != (unsigned short)port)) { + RedisModule_Log(mr_staticCtx, "notice", + "Topology reconcile: shard %s changed its address; doing a full topology refresh", nodeId); + mr_dictRelease(seenPrimaries); + RedisModule_FreeCallReply(allSlotsReply); + MR_RefreshClusterData(); + return; + } mr_dictAdd(seenPrimaries, nodeId, NULL); /* duplicate ranges for a node are ignored */ if (n->isMe) { From 6118f92b366cfc79803da961de79eaa70a97697c Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 13 Jul 2026 15:04:10 +0300 Subject: [PATCH 08/19] MOD-16382 Review: build the reconcile view from the cluster module API Per review: drop the CLUSTER SLOTS RM_Call+parse in favor of the same cluster module API the short-form CLUSTERSET builds from (GetClusterNodesList/GetClusterNodeInfo/GetClusterNodeSlotRanges), expressed as a comparable topology-view struct; the decision is a plain master-set compare (ids + addresses). Falls back to a full refresh when the API is unavailable. Trims the over-long comments. Co-Authored-By: Claude Opus 4.8 --- src/cluster.c | 242 ++++++++++++++++++++++++-------------------------- src/cluster.h | 10 +-- 2 files changed, 120 insertions(+), 132 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 55207e2a..931fa486 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -955,136 +955,134 @@ static void MR_RefreshClusterData(){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } -/* Reconcile the LibMR cluster view with a fresh CLUSTER SLOTS, reusing the existing - * Node structs (and their live connections) wherever possible. This is the single - * entry point the topology-change handler calls, regardless of which - * reason flags the event carried, and it decides what to do from the actual - * primary-set delta: - * - * - If the set of slot-serving primaries is unchanged (only slot ownership moved - * between primaries we are already connected to -- an in-place reshard), we just - * repoint CurrCluster->slots in place. Connections and in-flight cross-shard - * executions are left untouched, so a reshard never drops multi-key commands. - * - If the set of primaries changed (a primary entered or left the serving set -- - * scale-out, scale-in, failover), we escalate to MR_RefreshClusterData, which - * reconnects to the new set and aborts in-flight executions (the client retries). - * - * Deciding by the primary-set delta rather than by the event's reason flags means an - * over-broad or spurious notification (a replica re-pointing, an OK<->FAIL flip, a - * slotless node joining) that does not change the primary set costs only a cheap - * in-place repoint -- never a connection rebuild or a dropped in-flight multi-key - * command. - * - * In OSS mode CurrCluster->slots is the routing source of truth (the SendMsgType_BySlot - * path) and fan-out (SendMsgType_All) iterates the nodes dict; both stay correct here. - * Per-node slotRanges are only consumed by the Enterprise CLUSTERSET build and the - * debug RG.INFOCLUSTER reply, so they are intentionally left untouched. */ -static void MR_UpdateClusterSlots(){ - if (!clusterCtx.CurrCluster) { - /* Nothing to reconcile against yet; build from scratch. */ - MR_RefreshClusterData(); - return; - } - - if(!(RedisModule_GetContextFlags(mr_staticCtx) & REDISMODULE_CTX_FLAGS_CLUSTER)){ - return; - } - - RedisModule_ThreadSafeContextLock(mr_staticCtx); - RedisModuleCallReply *allSlotsReply = RedisModule_Call(mr_staticCtx, "cluster", "c", "slots"); - RedisModule_ThreadSafeContextUnlock(mr_staticCtx); - RedisModule_Assert(RedisModule_CallReplyType(allSlotsReply) == REDISMODULE_REPLY_ARRAY); - - /* Repoint the whole slot map from the fresh reply; reused nodes keep their - * connections. Runs on the event-loop thread, same as the dispatch path that - * reads slots[], so no reader can observe a half-updated map. */ - memset(clusterCtx.CurrCluster->slots, 0, sizeof(clusterCtx.CurrCluster->slots)); - - /* Distinct primaries seen in the fresh reply, so we can detect a primary that - * *left* the serving set after the loop (a primary that *entered* is caught - * inline below as an unknown shard id). */ - mr_dict* seenPrimaries = mr_dictCreate(&mr_dictTypeHeapStrings, NULL); +/* One entry of a topology view: a master node and one of its slot ranges + * (minSlot > maxSlot when it serves none). A master with several ranges + * appears once per range. */ +typedef struct TopologyViewEntry { + char id[REDISMODULE_NODE_ID_LEN + 1]; + char ip[INET6_ADDRSTRLEN]; + unsigned short port; + long long minSlot; + long long maxSlot; +} TopologyViewEntry; + +/* Build a topology view of the masters currently visible through the cluster + * module API (the same source the short-form CLUSTERSET builds from). Masters + * serving no slots are left out unless includeSlotless is set. Returns NULL + * when the cluster API is unavailable; the caller owns the returned array. */ +static ARR(TopologyViewEntry) BuildClusterApiView(bool includeSlotless){ + if (RedisModule_GetClusterNodeSlotRanges == NULL) + return NULL; - for(size_t i = 0 ; i < RedisModule_CallReplyLength(allSlotsReply) ; ++i){ - RedisModuleCallReply *slotRangeReply = RedisModule_CallReplyArrayElement(allSlotsReply, i); + size_t numNodes; + char **nodeList = RedisModule_GetClusterNodesList(mr_staticCtx, &numNodes); + if (!nodeList) + return NULL; - RedisModuleCallReply *minSlotReply = RedisModule_CallReplyArrayElement(slotRangeReply, 0); - RedisModule_Assert(RedisModule_CallReplyType(minSlotReply) == REDISMODULE_REPLY_INTEGER); - long long minSlot = RedisModule_CallReplyInteger(minSlotReply); + ARR(TopologyViewEntry) view = array_new(TopologyViewEntry, numNodes); + for (size_t i = 0; i < numNodes; i++) { + TopologyViewEntry e = { .minSlot = 0, .maxSlot = -1 }; + memcpy(e.id, nodeList[i], REDISMODULE_NODE_ID_LEN); // nodeList[i] is not null-terminated + e.id[REDISMODULE_NODE_ID_LEN] = '\0'; - RedisModuleCallReply *maxSlotReply = RedisModule_CallReplyArrayElement(slotRangeReply, 1); - RedisModule_Assert(RedisModule_CallReplyType(maxSlotReply) == REDISMODULE_REPLY_INTEGER); - long long maxSlot = RedisModule_CallReplyInteger(maxSlotReply); + int port, flags; + if (RedisModule_GetClusterNodeInfo(mr_staticCtx, e.id, e.ip, NULL, &port, &flags) != REDISMODULE_OK) + continue; + if (!(flags & REDISMODULE_NODE_MASTER)) continue; // Skip replica nodes + e.port = (unsigned short)port; - RedisModuleCallReply *nodeDetailsReply = RedisModule_CallReplyArrayElement(slotRangeReply, 2); - RedisModule_Assert(RedisModule_CallReplyType(nodeDetailsReply) == REDISMODULE_REPLY_ARRAY); - RedisModule_Assert(RedisModule_CallReplyLength(nodeDetailsReply) >= 3); - RedisModuleCallReply *nodeidReply = RedisModule_CallReplyArrayElement(nodeDetailsReply, 2); - size_t idLen; - const char* id = RedisModule_CallReplyStringPtr(nodeidReply,&idLen); + RedisModuleSlotRangeArray *slots = RedisModule_GetClusterNodeSlotRanges(mr_staticCtx, e.id); + RedisModule_Assert(slots != NULL); + if (slots->num_ranges == 0) { + if (includeSlotless) + view = array_append(view, e); + } else { + for (size_t j = 0; j < slots->num_ranges; j++) { + e.minSlot = slots->ranges[j].start; + e.maxSlot = slots->ranges[j].end; + view = array_append(view, e); + } + } + RedisModule_ClusterFreeSlotRanges(mr_staticCtx, slots); + } + RedisModule_FreeClusterNodesList(nodeList); + return view; +} - char nodeId[REDISMODULE_NODE_ID_LEN + 1]; - memcpy(nodeId, id, REDISMODULE_NODE_ID_LEN); - nodeId[REDISMODULE_NODE_ID_LEN] = '\0'; +/* Compare the view against the current cluster and, when the master set is + * unchanged -- same ids at the same addresses -- repoint the slot map in + * place, keeping the nodes and their live connections. Returns false when a + * master entered/left the set or changed its address; the caller must then do + * a full rebuild. Runs on the event-loop thread, same as the dispatch path + * that reads slots[], so no reader can observe a half-updated map. */ +static bool MR_TryApplyTopologyInPlace(const TopologyViewEntry* view, size_t n){ + if (!clusterCtx.CurrCluster) { + return false; + } - Node* n = MR_GetNode(nodeId); - if(!n){ - /* A primary entered the serving set -> rebuild so it gets connected. */ + /* Pass 1: validate. Every master in the view must already be known, at the + * same address; and none we hold may have left (every view id is in the + * nodes dict, so an equal distinct count means equal sets). */ + mr_dict* seen = mr_dictCreate(&mr_dictTypeHeapStrings, NULL); + for (size_t i = 0 ; i < n ; ++i) { + Node* node = MR_GetNode(view[i].id); + if (!node) { RedisModule_Log(mr_staticCtx, "notice", - "Topology reconcile saw a new shard %s; doing a full topology refresh", nodeId); - mr_dictRelease(seenPrimaries); - RedisModule_FreeCallReply(allSlotsReply); - MR_RefreshClusterData(); - return; + "Topology reconcile saw a new shard %s; doing a full topology refresh", view[i].id); + mr_dictRelease(seen); + return false; } - - /* Same id but a new address (the node restarted elsewhere, e.g. a pod - * reschedule that kept nodes.conf): the cached connection would redial - * the old address forever, so rebuild to reconnect. The port is taken - * from RedisModule_GetClusterNodeInfo for the same reason as in - * MR_RefreshClusterData (CLUSTER SLOTS reports the non-TLS port, see - * redis/redis#12233). */ - RedisModuleCallReply *nodeipReply = RedisModule_CallReplyArrayElement(nodeDetailsReply, 0); - size_t ipLen; - const char* ip = RedisModule_CallReplyStringPtr(nodeipReply, &ipLen); - int port = 0; - RedisModule_ThreadSafeContextLock(mr_staticCtx); - RedisModule_GetClusterNodeInfo(mr_staticCtx, nodeId, NULL, NULL, &port, NULL); - RedisModule_ThreadSafeContextUnlock(mr_staticCtx); - if (strlen(n->ip) != ipLen || memcmp(n->ip, ip, ipLen) != 0 || - (port != 0 && n->port != (unsigned short)port)) { + if (strcmp(node->ip, view[i].ip) != 0 || + (view[i].port != 0 && node->port != view[i].port)) { RedisModule_Log(mr_staticCtx, "notice", - "Topology reconcile: shard %s changed its address; doing a full topology refresh", nodeId); - mr_dictRelease(seenPrimaries); - RedisModule_FreeCallReply(allSlotsReply); - MR_RefreshClusterData(); - return; + "Topology reconcile: shard %s changed its address; doing a full topology refresh", view[i].id); + mr_dictRelease(seen); + return false; } - mr_dictAdd(seenPrimaries, nodeId, NULL); /* duplicate ranges for a node are ignored */ + mr_dictAdd(seen, (void*)view[i].id, NULL); /* duplicate ranges for a shard are ignored */ + } + size_t shardsNow = mr_dictSize(seen); + mr_dictRelease(seen); + if (shardsNow != mr_dictSize(clusterCtx.CurrCluster->nodes)) { + RedisModule_Log(mr_staticCtx, "notice", + "Topology reconcile: the set of shards changed; doing a full topology refresh"); + return false; + } - if (n->isMe) { - clusterCtx.minSlot = minSlot; - clusterCtx.maxSlot = maxSlot; + /* Pass 2: apply -- repoint the whole slot map; the reused nodes keep their + * connections. */ + memset(clusterCtx.CurrCluster->slots, 0, sizeof(clusterCtx.CurrCluster->slots)); + bool mySlotsSet = false; + for (size_t i = 0 ; i < n ; ++i) { + Node* node = MR_GetNode(view[i].id); + if (node->isMe && !mySlotsSet) { + /* fill the fallback single-range from my first range, like the + * full builders do */ + clusterCtx.minSlot = view[i].minSlot; + clusterCtx.maxSlot = view[i].maxSlot; + mySlotsSet = true; } - - for(int k = minSlot ; k <= maxSlot ; ++k){ - clusterCtx.CurrCluster->slots[k] = n; + for (long long k = view[i].minSlot ; k <= view[i].maxSlot ; ++k) { + clusterCtx.CurrCluster->slots[k] = node; } } - RedisModule_FreeCallReply(allSlotsReply); + return true; +} - /* If a primary we knew about is no longer serving any slots (scale-in / failover), - * the set of primaries shrank -- rebuild so we drop it and abort in-flight - * executions that assumed it. Additions were handled above, so a distinct count - * smaller than the primaries we hold means exactly a removal. */ - size_t primariesNow = mr_dictSize(seenPrimaries); - mr_dictRelease(seenPrimaries); - if (primariesNow != mr_dictSize(clusterCtx.CurrCluster->nodes)) { - RedisModule_Log(mr_staticCtx, "notice", - "Topology reconcile: the set of primaries changed; doing a full topology refresh"); - MR_RefreshClusterData(); +/* Reconcile against the current cluster state: rebuild the connections only + * when the set of masters (or one of their addresses) changed; otherwise just + * repoint the slot map in place and keep the existing connections. */ +static void MR_UpdateClusterSlots(){ + if(!(RedisModule_GetContextFlags(mr_staticCtx) & REDISMODULE_CTX_FLAGS_CLUSTER)){ return; } + + ARR(TopologyViewEntry) view = BuildClusterApiView(false); + bool applied = view != NULL && MR_TryApplyTopologyInPlace(view, array_len(view)); + if (view) + array_free(view); + if (!applied) + MR_RefreshClusterData(); } static void GenerateRunId(Cluster* cluster){ @@ -1455,27 +1453,19 @@ static int MR_ClusterRefresh(RedisModuleCtx *ctx, RedisModuleString **argv, int return REDISMODULE_OK; } -/* Runs on the event loop. MR_UpdateClusterSlots decides from the actual - * primary-set delta whether a full rebuild (which aborts in-flight executions) - * is needed or a connection-preserving in-place slot-map update suffices, so a - * redundant event costs one CLUSTER SLOTS read and never drops in-flight - * multi-key commands. */ +/* Runs on the event loop. */ static void MR_TopoRefreshTask(void* ctx){ REDISMODULE_NOT_USED(ctx); MR_UpdateClusterSlots(); } /* Request an OSS cluster topology refresh; every event schedules one reconcile - * on the event loop. No-op when not running as an OSS cluster - * (Enterprise/standalone), mirroring the OSS-only registration of the - * REFRESHCLUSTER command, so it is safe to call unconditionally from a Redis - * server-event callback. */ + * on the event loop. No-op when not in OSS cluster mode, so it is safe to call + * unconditionally from a Redis server-event callback. */ void MR_ClusterRefreshTopology(int change_flags){ if (!clusterCtx.isOss) return; - /* change_flags is advisory: the refresh reconciles against CLUSTER SLOTS - * and decides rebuild-vs-in-place from the actual primary-set delta, so we - * don't act on the reason bits here. Kept in the signature for callers / - * future use. */ + /* change_flags is advisory: the refresh compares the actual master set, so + * the reason bits are not acted on. */ REDISMODULE_NOT_USED(change_flags); MR_EventLoopAddTask(MR_TopoRefreshTask, NULL); } diff --git a/src/cluster.h b/src/cluster.h index 31594018..61946bc6 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -39,12 +39,10 @@ int MR_ClusterInit(RedisModuleCtx* rctx, char *password); /* Schedule an OSS cluster topology refresh on the event loop. 'change_flags' * is a bitmask of REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_* reasons and is - * advisory only: the refresh reconciles against CLUSTER SLOTS and rebuilds the - * connections only when the set of slot-serving primaries actually changed, - * otherwise it updates just the slot->node routing and preserves the existing - * connections (so in-flight cross-shard queries are not disrupted by an in-place - * reshard). No-op outside of OSS cluster mode. Safe to call from a Redis - * server-event callback. */ + * advisory only: the refresh rebuilds the connections when the set of master + * nodes changed and otherwise just updates the slot->node routing in place. + * No-op outside of OSS cluster mode. Safe to call from a Redis server-event + * callback. */ void MR_ClusterRefreshTopology(int change_flags); size_t MR_ClusterGetSlotByKey(const char* key, size_t len); From 57a2dd15050b7a6bb536f63c791c30b970ad9b9c Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Sun, 5 Jul 2026 22:30:17 +0300 Subject: [PATCH 09/19] MOD-16399 Skip the rebuild when CLUSTERSET carries an unchanged topology A cluster-set rebuild drops all inter-shard connections and aborts every in-flight initiator execution with 'cluster topology changed'. The topology is re-broadcast on many events that do not change it (node events, shard reconnects, delivery retries), so multi-shard commands racing such a resend failed spuriously. Compare the incoming long-form command against the stored one (MYID excluded) and keep the current cluster when they match. Short-form commands still rebuild: their topology derives from the server's cluster state, not from the arguments. Co-Authored-By: Claude Fable 5 --- src/cluster.c | 28 ++++++++++++ tests/mr_test_module/pytests/test_network.py | 45 ++++++++++++++++---- 2 files changed, 65 insertions(+), 8 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 931fa486..81075e2a 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1382,7 +1382,35 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } +/* Returns true when the incoming long-form cluster-set command carries the exact + * topology the current cluster was built from. The MYID slot is excluded: it names + * the receiving shard, not the topology, and is stored as NULL (see CopyClusterSetArgs). */ +static bool ClusterSetCommandIsUnchanged(RedisModuleString** argv, int argc){ + Cluster* cur = clusterCtx.CurrCluster; + if (!cur || !cur->clusterSetCommand || cur->clusterSetCommandSize != argc) + return false; + if (!IsLongFormClusterSet(argc)) + return false; + for (int i = 1 ; i < argc ; ++i) { + if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) + continue; + const char* arg = RedisModule_StringPtrLen(argv[i], NULL); + if (strcmp(arg, cur->clusterSetCommand[i]) != 0) + return false; + } + return true; +} + static int MR_SetClusterData(RedisModuleString** argv, int argc){ + /* The topology is re-broadcast on many events that do not change it (node events, + * shard reconnects, delivery retries). Rebuilding for an identical topology would + * drop all inter-shard connections and abort in-flight executions for nothing. */ + if (ClusterSetCommandIsUnchanged(argv, argc)) { + RedisModule_Log(mr_staticCtx, "notice", + "Got cluster set command with an unchanged topology, skipping the rebuild"); + return REDISMODULE_OK; + } + if(clusterCtx.CurrCluster) MR_ClusterFree(); diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index 09ca7594..73e9ab84 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -210,15 +210,13 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _send_cluster_set(self): - # try to promote to internal connection - promote_internal_client_if_supported(env=self.env) + def _cluster_set_args(self, mock_shard_slot_start=8193): # IPv6 endpoints must be bracketed in host:port strings endpoint_host = '[%s]' % self.host if ':' in self.host else self.host # Build arguments according to MR_SetClusterData parser: # argv[6] => myId, argv[7] => "RANGES", argv[8] => numOfRanges, then repeating: # "SHARD" "SLOTRANGE" "ADDR" ["MASTER"] - args = [ + return [ 'NO-USED', # [1] 'NO-USED', # [2] 'NO-USED', # [3] @@ -229,16 +227,20 @@ def _send_cluster_set(self): '2', # [8] two ranges # Shard 1 (current Redis) - HARDCODED PORT 6379 'SHARD', '1', - 'SLOTRANGE', '0', '8192', + 'SLOTRANGE', '0', str(mock_shard_slot_start - 1), 'ADDR', 'password@%s:6379' % endpoint_host, 'MASTER', # Shard 2 (mock shard) 'SHARD', '2', - 'SLOTRANGE', '8193', '16383', + 'SLOTRANGE', str(mock_shard_slot_start), '16383', 'ADDR', 'password@%s:%d' % (endpoint_host, self.port), 'MASTER' ] - self.env.cmd('MRTESTS.CLUSTERSET', *args) + + def _send_cluster_set(self): + # try to promote to internal connection + promote_internal_client_if_supported(env=self.env) + self.env.cmd('MRTESTS.CLUSTERSET', *self._cluster_set_args()) self.env.cmd('MRTESTS.FORCESHARDSCONNECTION') def __enter__(self): @@ -681,7 +683,11 @@ def testMassiveClusterSet(env, conn): with ShardMock(env, host) as shardMock: for i in range(1000): conn = shardMock.GetConnection(sendHelloResponse=False) - shardMock._send_cluster_set() + # Alternate the slot boundary so every command carries a changed + # topology and forces a rebuild — identical re-sends are a no-op. + promote_internal_client_if_supported(env=env) + env.cmd('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_slot_start=8194 - (i % 2))) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') @MRTestDecorator(skipOnCluster=True) def testMassiveClusterSetFromShard(env, conn): @@ -744,3 +750,26 @@ def testSendMultiRangePerNodeTopology(env, conn): res = env.cmd(*cmd) assert res == 'OK' + + +@MRTestDecorator(skipOnCluster=True) +def testIdenticalClusterSetIsNoOp(env, conn): + for host in _get_hosts(): + with ShardMock(env, host) as shardMock: + conn = shardMock.GetConnection() + + run_id = env.cmd('MRTESTS.INFOCLUSTER')[3] + + # Re-sending the exact same topology must not rebuild the cluster: + # the run id is kept, and so is the live connection to the mock shard. + env.expect('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args()).equal('OK') + env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') + time.sleep(0.5) + env.assertTrue(shardMock.new_conns.empty()) + + # A genuinely different topology (moved slot boundary) must still rebuild. + env.expect('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_slot_start=4096)).equal('OK') + env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') + conn = shardMock.GetConnection() From e8a35786ce59e90e34d0c784ef92f140e387e271 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Tue, 7 Jul 2026 10:57:27 +0300 Subject: [PATCH 10/19] Review round: positive naming, length-aware compare, short-form no-op skip - ClusterSetIsNewTopology replaces the double-negative ClusterSetCommandIsUnchanged - Arguments are compared with their lengths (memcmp) instead of strcmp - The short form now skips too: the derived master set is snapshotted into a canonical string (sorted by node id), stored on the cluster, and compared against the snapshot derived for the next short-form command Co-Authored-By: Claude Fable 5 --- src/cluster.c | 119 ++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 100 insertions(+), 19 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 81075e2a..60af6c6a 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -149,6 +149,7 @@ typedef struct Cluster { Node* slots[NUMBER_OF_SLOTS]; size_t clusterSetCommandSize; char** clusterSetCommand; + char* shortFormTopology; /* canonical master-set snapshot; set only by the short form */ char runId[RUN_ID_SIZE + 1]; }Cluster; @@ -811,6 +812,10 @@ static void MR_ClusterFree(){ MR_FREE(clusterCtx.CurrCluster->clusterSetCommand); } + if(clusterCtx.CurrCluster->shortFormTopology){ + MR_FREE(clusterCtx.CurrCluster->shortFormTopology); + } + MR_FREE(clusterCtx.CurrCluster); clusterCtx.CurrCluster = NULL; clusterCtx.minSlot = 0; @@ -1382,32 +1387,105 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } -/* Returns true when the incoming long-form cluster-set command carries the exact - * topology the current cluster was built from. The MYID slot is excluded: it names - * the receiving shard, not the topology, and is stored as NULL (see CopyClusterSetArgs). */ -static bool ClusterSetCommandIsUnchanged(RedisModuleString** argv, int argc){ +static int compare_snapshot_entries(const void* a, const void* b){ + return strcmp(*(const char**)a, *(const char**)b); +} + +/* Build a canonical snapshot of the master set currently visible through the + * cluster API: one " : " line per master, sorted by + * node id so the nodes-list iteration order does not matter. The password is + * deliberately absent: it comes from the command arguments, which are compared + * separately. Returns NULL when the cluster API is unavailable; the caller owns + * the result. */ +static char* ShortFormTopologySnapshot(void){ + if (RedisModule_GetClusterNodeSlotRanges == NULL) + return NULL; + + size_t numNodes; + char **nodeList = RedisModule_GetClusterNodesList(mr_staticCtx, &numNodes); + if (!nodeList) + return NULL; + + ARR(char*) entries = array_new(char*, numNodes); + for (size_t i = 0; i < numNodes; i++) { + char nodeId[REDISMODULE_NODE_ID_LEN + 1]; // nodeList[i] is not null-terminated + memcpy(nodeId, nodeList[i], REDISMODULE_NODE_ID_LEN); + nodeId[REDISMODULE_NODE_ID_LEN] = '\0'; + + char ip[INET6_ADDRSTRLEN]; // INET6_ADDRSTRLEN includes the closing '\0' + int port, flags; + if (RedisModule_GetClusterNodeInfo(mr_staticCtx, nodeId, ip, NULL, &port, &flags) != REDISMODULE_OK) + continue; + if (!(flags & REDISMODULE_NODE_MASTER)) continue; // Skip replica nodes + + RedisModuleSlotRangeArray *slots = RedisModule_GetClusterNodeSlotRanges(mr_staticCtx, nodeId); + RedisModule_Assert(slots != NULL); + size_t cap = REDISMODULE_NODE_ID_LEN + INET6_ADDRSTRLEN + slots->num_ranges * 12 + 32; + char* entry = MR_ALLOC(cap); + size_t len = snprintf(entry, cap, "%s %s:%d", nodeId, ip, port); + for (size_t j = 0; j < slots->num_ranges; j++) + len += snprintf(entry + len, cap - len, " %d-%d", (int)slots->ranges[j].start, (int)slots->ranges[j].end); + RedisModule_ClusterFreeSlotRanges(mr_staticCtx, slots); + entries = array_append(entries, entry); + } + RedisModule_FreeClusterNodesList(nodeList); + + qsort(entries, array_len(entries), sizeof(*entries), compare_snapshot_entries); + + size_t total = 1; + for (size_t i = 0; i < array_len(entries); i++) + total += strlen(entries[i]) + 1; + char* snapshot = MR_ALLOC(total); + size_t pos = 0; + for (size_t i = 0; i < array_len(entries); i++) { + pos += snprintf(snapshot + pos, total - pos, "%s\n", entries[i]); + MR_FREE(entries[i]); + } + snapshot[pos] = '\0'; + array_free(entries); + return snapshot; +} + +/* Returns true when the incoming cluster-set command carries a topology different + * from the one the current cluster was built from. The long form is compared by + * its arguments; the MYID slot is excluded, as it names the receiving shard rather + * than the topology, and is stored as NULL (see CopyClusterSetArgs). The short form + * derives the topology from the server's cluster state, so on top of its arguments + * (just the AUTH password) it compares `snapshot`, the freshly derived master set. */ +static bool ClusterSetIsNewTopology(RedisModuleString** argv, int argc, const char* snapshot){ Cluster* cur = clusterCtx.CurrCluster; if (!cur || !cur->clusterSetCommand || cur->clusterSetCommandSize != argc) - return false; - if (!IsLongFormClusterSet(argc)) - return false; + return true; for (int i = 1 ; i < argc ; ++i) { - if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) + if (IsLongFormClusterSet(argc) && i == CLUSTERSET_MYID_LONG_FORM_INDEX) continue; - const char* arg = RedisModule_StringPtrLen(argv[i], NULL); - if (strcmp(arg, cur->clusterSetCommand[i]) != 0) - return false; - } - return true; + size_t argLen; + const char* arg = RedisModule_StringPtrLen(argv[i], &argLen); + if (argLen != strlen(cur->clusterSetCommand[i]) || + memcmp(arg, cur->clusterSetCommand[i], argLen) != 0) + return true; + } + if (IsShortFormClusterSet(argc)) + return !snapshot || !cur->shortFormTopology || strcmp(snapshot, cur->shortFormTopology) != 0; + return false; } static int MR_SetClusterData(RedisModuleString** argv, int argc){ + if (!(IsLongFormClusterSet(argc) || IsShortFormClusterSet(argc))) { + RedisModule_Log(mr_staticCtx, "warning", "Could not parse cluster set arguments"); + return REDISMODULE_ERR; + } + + char* snapshot = IsShortFormClusterSet(argc) ? ShortFormTopologySnapshot() : NULL; + /* The topology is re-broadcast on many events that do not change it (node events, * shard reconnects, delivery retries). Rebuilding for an identical topology would * drop all inter-shard connections and abort in-flight executions for nothing. */ - if (ClusterSetCommandIsUnchanged(argv, argc)) { + if (!ClusterSetIsNewTopology(argv, argc, snapshot)) { RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command with an unchanged topology, skipping the rebuild"); + if (snapshot) + MR_FREE(snapshot); return REDISMODULE_OK; } @@ -1417,12 +1495,15 @@ static int MR_SetClusterData(RedisModuleString** argv, int argc){ if (IsLongFormClusterSet(argc)) { SetClusterDataLongForm(argv, argc); return REDISMODULE_OK; - } else if (IsShortFormClusterSet(argc)) { - return SetClusterDataShortForm(argv, argc); - } else { - RedisModule_Log(mr_staticCtx, "warning", "Could not parse cluster set arguments"); - return REDISMODULE_ERR; } + + int res = SetClusterDataShortForm(argv, argc); + if (res == REDISMODULE_OK && clusterCtx.CurrCluster) { + clusterCtx.CurrCluster->shortFormTopology = snapshot; // now owned by the cluster + } else if (snapshot) { + MR_FREE(snapshot); + } + return res; } /* runs in the event loop so its safe to update cluster From fafda414d394d1d0dd292449fa969867827d074a Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 13 Jul 2026 15:08:42 +0300 Subject: [PATCH 11/19] MOD-16399 Reconcile CLUSTERSET through the shared topology mechanism Both CLUSTERSET forms now feed the same reconcile the topology-change event uses (MR_TryApplyTopologyInPlace): the command arguments (long form) or the server's cluster state (short form) are reduced to a topology view and diffed against the connected shard set. An unchanged set -- including a reshard that only moves slot ranges between the same shards -- is applied in place, keeping every connection, in-flight execution and the run id; a shard entering/leaving the set, or changing its address or credentials, still tears down and rebuilds. This supersedes the short-form string-snapshot compare (removed) and extends the long-form identical-args skip: resharding no longer aborts in-flight multi-shard commands. Co-Authored-By: Claude Opus 4.8 --- src/cluster.c | 227 +++++++++++-------- tests/mr_test_module/pytests/test_network.py | 52 ++++- 2 files changed, 170 insertions(+), 109 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 60af6c6a..e1a17af7 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -69,6 +69,7 @@ static bool IsShortFormClusterSet(int argc) { #define CLUSTER_REFRESH_COMMAND xstr(MODULE_NAME)".REFRESHCLUSTER" #define CLUSTER_SET_COMMAND xstr(MODULE_NAME)".CLUSTERSET" #define CLUSTER_SET_FROM_SHARD_COMMAND xstr(MODULE_NAME)".CLUSTERSETFROMSHARD" +#define INTERNAL_PASSWORD_MAX_SIZE 100 #define CLUSTER_INFO_COMMAND xstr(MODULE_NAME)".INFOCLUSTER" #define NETWORK_TEST_COMMAND xstr(MODULE_NAME)".NETWORKTEST" #define FORCE_SHARDS_CONNECTION xstr(MODULE_NAME)".FORCESHARDSCONNECTION" @@ -149,7 +150,6 @@ typedef struct Cluster { Node* slots[NUMBER_OF_SLOTS]; size_t clusterSetCommandSize; char** clusterSetCommand; - char* shortFormTopology; /* canonical master-set snapshot; set only by the short form */ char runId[RUN_ID_SIZE + 1]; }Cluster; @@ -812,10 +812,6 @@ static void MR_ClusterFree(){ MR_FREE(clusterCtx.CurrCluster->clusterSetCommand); } - if(clusterCtx.CurrCluster->shortFormTopology){ - MR_FREE(clusterCtx.CurrCluster->shortFormTopology); - } - MR_FREE(clusterCtx.CurrCluster); clusterCtx.CurrCluster = NULL; clusterCtx.minSlot = 0; @@ -967,6 +963,8 @@ typedef struct TopologyViewEntry { char id[REDISMODULE_NODE_ID_LEN + 1]; char ip[INET6_ADDRSTRLEN]; unsigned short port; + char password[INTERNAL_PASSWORD_MAX_SIZE + 1]; + bool comparePassword; /* the view's source carries per-shard passwords */ long long minSlot; long long maxSlot; } TopologyViewEntry; @@ -1044,6 +1042,13 @@ static bool MR_TryApplyTopologyInPlace(const TopologyViewEntry* view, size_t n){ mr_dictRelease(seen); return false; } + if (view[i].comparePassword && + strcmp(node->password ? node->password : "", view[i].password) != 0) { + RedisModule_Log(mr_staticCtx, "notice", + "Topology reconcile: shard %s changed its password; doing a full topology refresh", view[i].id); + mr_dictRelease(seen); + return false; + } mr_dictAdd(seen, (void*)view[i].id, NULL); /* duplicate ranges for a shard are ignored */ } size_t shardsNow = mr_dictSize(seen); @@ -1133,7 +1138,6 @@ static void InitClusterData(Cluster* cluster, RedisModuleString** argv, int argc cluster->nodes = mr_dictCreate(&mr_dictTypeHeapStrings, NULL); } -#define INTERNAL_PASSWORD_MAX_SIZE 100 // Parse a SHARD entry into the output arguments and return the index of the last parsed token static int ParseShardEntry(RedisModuleString** argv, int argc, int index, @@ -1387,75 +1391,14 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } -static int compare_snapshot_entries(const void* a, const void* b){ - return strcmp(*(const char**)a, *(const char**)b); -} - -/* Build a canonical snapshot of the master set currently visible through the - * cluster API: one " : " line per master, sorted by - * node id so the nodes-list iteration order does not matter. The password is - * deliberately absent: it comes from the command arguments, which are compared - * separately. Returns NULL when the cluster API is unavailable; the caller owns - * the result. */ -static char* ShortFormTopologySnapshot(void){ - if (RedisModule_GetClusterNodeSlotRanges == NULL) - return NULL; - - size_t numNodes; - char **nodeList = RedisModule_GetClusterNodesList(mr_staticCtx, &numNodes); - if (!nodeList) - return NULL; - - ARR(char*) entries = array_new(char*, numNodes); - for (size_t i = 0; i < numNodes; i++) { - char nodeId[REDISMODULE_NODE_ID_LEN + 1]; // nodeList[i] is not null-terminated - memcpy(nodeId, nodeList[i], REDISMODULE_NODE_ID_LEN); - nodeId[REDISMODULE_NODE_ID_LEN] = '\0'; - - char ip[INET6_ADDRSTRLEN]; // INET6_ADDRSTRLEN includes the closing '\0' - int port, flags; - if (RedisModule_GetClusterNodeInfo(mr_staticCtx, nodeId, ip, NULL, &port, &flags) != REDISMODULE_OK) - continue; - if (!(flags & REDISMODULE_NODE_MASTER)) continue; // Skip replica nodes - - RedisModuleSlotRangeArray *slots = RedisModule_GetClusterNodeSlotRanges(mr_staticCtx, nodeId); - RedisModule_Assert(slots != NULL); - size_t cap = REDISMODULE_NODE_ID_LEN + INET6_ADDRSTRLEN + slots->num_ranges * 12 + 32; - char* entry = MR_ALLOC(cap); - size_t len = snprintf(entry, cap, "%s %s:%d", nodeId, ip, port); - for (size_t j = 0; j < slots->num_ranges; j++) - len += snprintf(entry + len, cap - len, " %d-%d", (int)slots->ranges[j].start, (int)slots->ranges[j].end); - RedisModule_ClusterFreeSlotRanges(mr_staticCtx, slots); - entries = array_append(entries, entry); - } - RedisModule_FreeClusterNodesList(nodeList); - - qsort(entries, array_len(entries), sizeof(*entries), compare_snapshot_entries); - - size_t total = 1; - for (size_t i = 0; i < array_len(entries); i++) - total += strlen(entries[i]) + 1; - char* snapshot = MR_ALLOC(total); - size_t pos = 0; - for (size_t i = 0; i < array_len(entries); i++) { - pos += snprintf(snapshot + pos, total - pos, "%s\n", entries[i]); - MR_FREE(entries[i]); - } - snapshot[pos] = '\0'; - array_free(entries); - return snapshot; -} - -/* Returns true when the incoming cluster-set command carries a topology different - * from the one the current cluster was built from. The long form is compared by - * its arguments; the MYID slot is excluded, as it names the receiving shard rather - * than the topology, and is stored as NULL (see CopyClusterSetArgs). The short form - * derives the topology from the server's cluster state, so on top of its arguments - * (just the AUTH password) it compares `snapshot`, the freshly derived master set. */ -static bool ClusterSetIsNewTopology(RedisModuleString** argv, int argc, const char* snapshot){ +/* Returns true when the incoming cluster-set command is argument-identical to + * the one the current cluster was built from. The MYID slot is excluded: it + * names the receiving shard rather than the topology, and is stored as NULL + * (see CopyClusterSetArgs). */ +static bool ClusterSetArgsUnchanged(RedisModuleString** argv, int argc){ Cluster* cur = clusterCtx.CurrCluster; if (!cur || !cur->clusterSetCommand || cur->clusterSetCommandSize != argc) - return true; + return false; for (int i = 1 ; i < argc ; ++i) { if (IsLongFormClusterSet(argc) && i == CLUSTERSET_MYID_LONG_FORM_INDEX) continue; @@ -1463,47 +1406,133 @@ static bool ClusterSetIsNewTopology(RedisModuleString** argv, int argc, const ch const char* arg = RedisModule_StringPtrLen(argv[i], &argLen); if (argLen != strlen(cur->clusterSetCommand[i]) || memcmp(arg, cur->clusterSetCommand[i], argLen) != 0) - return true; + return false; } - if (IsShortFormClusterSet(argc)) - return !snapshot || !cur->shortFormTopology || strcmp(snapshot, cur->shortFormTopology) != 0; - return false; + return true; } -static int MR_SetClusterData(RedisModuleString** argv, int argc){ - if (!(IsLongFormClusterSet(argc) || IsShortFormClusterSet(argc))) { - RedisModule_Log(mr_staticCtx, "warning", "Could not parse cluster set arguments"); - return REDISMODULE_ERR; +/* Returns true when the long-form MYID argument still names the shard the + * current cluster was built for (padded the way SetMyId pads it). */ +static bool LongFormMyIdUnchanged(RedisModuleString** argv, int argc){ + RedisModule_Assert(CLUSTERSET_MYID_LONG_FORM_INDEX < argc); + size_t myIdLen; + const char* myId = RedisModule_StringPtrLen(argv[CLUSTERSET_MYID_LONG_FORM_INDEX], &myIdLen); + char padded[REDISMODULE_NODE_ID_LEN + 1]; + size_t zerosPadding = REDISMODULE_NODE_ID_LEN - myIdLen; + memset(padded, '0', zerosPadding); + memcpy(padded + zerosPadding, myId, myIdLen); + padded[REDISMODULE_NODE_ID_LEN] = '\0'; + return clusterCtx.CurrCluster && clusterCtx.CurrCluster->myId && + strcmp(clusterCtx.CurrCluster->myId, padded) == 0; +} + +/* Reduce the long-form CLUSTERSET arguments to a topology view (the same walk + * as SetClusterDataLongForm, without touching any state). The entries carry + * the per-shard passwords. The caller owns the returned array. */ +static ARR(TopologyViewEntry) BuildLongFormView(RedisModuleString** argv, int argc){ + size_t index = CLUSTERSET_MYID_LONG_FORM_INDEX + 1; + const char *token = RedisModule_StringPtrLen(argv[index], NULL); + if (strcasecmp(token, "HASREPLICATION") == 0) { // skip this token; we ignore it + index++; + RedisModule_Assert(index < argc); + token = RedisModule_StringPtrLen(argv[index], NULL); } + RedisModule_Assert(strcasecmp(token, "RANGES") == 0); + index++; + RedisModule_Assert(index < argc); + long long numOfRanges; + RedisModule_Assert(RedisModule_StringToLongLong(argv[index], &numOfRanges) == REDISMODULE_OK); + index++; - char* snapshot = IsShortFormClusterSet(argc) ? ShortFormTopologySnapshot() : NULL; + ARR(TopologyViewEntry) view = array_new(TopologyViewEntry, numOfRanges ? numOfRanges : 1); + for (size_t j = 0 ; j < (size_t)numOfRanges ; ++j) { + TopologyViewEntry e = { .comparePassword = true }; + bool shouldSkip; + index = ParseShardEntry(argv, argc, index, e.id, e.ip, &e.port, e.password, + &e.minSlot, &e.maxSlot, &shouldSkip); + if (index >= argc) + break; + if (shouldSkip) + continue; + view = array_append(view, e); + index++; + } + return view; +} - /* The topology is re-broadcast on many events that do not change it (node events, - * shard reconnects, delivery retries). Rebuilding for an identical topology would - * drop all inter-shard connections and abort in-flight executions for nothing. */ - if (!ClusterSetIsNewTopology(argv, argc, snapshot)) { - RedisModule_Log(mr_staticCtx, "notice", - "Got cluster set command with an unchanged topology, skipping the rebuild"); - if (snapshot) - MR_FREE(snapshot); - return REDISMODULE_OK; +/* Replace the stored cluster-set command, so the next identical-args compare + * checks against what was last applied, without touching any other state. */ +static void ReplaceClusterSetArgs(Cluster* cluster, RedisModuleString** argv, int argc){ + if (cluster->clusterSetCommand) { + for (size_t i = 0 ; i < cluster->clusterSetCommandSize ; ++i) { + if (cluster->clusterSetCommand[i]) + MR_FREE(cluster->clusterSetCommand[i]); + } + MR_FREE(cluster->clusterSetCommand); } + cluster->clusterSetCommand = MR_ALLOC(sizeof(char*) * argc); + cluster->clusterSetCommandSize = argc; + cluster->clusterSetCommand[0] = MR_STRDUP(CLUSTER_SET_FROM_SHARD_COMMAND); + CopyClusterSetArgs(cluster, argv, argc); +} - if(clusterCtx.CurrCluster) - MR_ClusterFree(); +static int MR_SetClusterData(RedisModuleString** argv, int argc){ + if (!(IsLongFormClusterSet(argc) || IsShortFormClusterSet(argc))) { + RedisModule_Log(mr_staticCtx, "warning", "Could not parse cluster set arguments"); + return REDISMODULE_ERR; + } + /* The topology is re-broadcast on many events that do not change it (node + * events, shard reconnects, delivery retries), and resharding moves slot + * ranges between the same shards. Neither needs to drop the inter-shard + * connections or abort in-flight executions: an argument-identical long + * form is skipped outright, and any other change goes through the same + * reconcile mechanism the topology-change event uses -- only a change of + * the shard set itself (or of a shard's address or credentials) tears + * down and rebuilds. */ if (IsLongFormClusterSet(argc)) { + if (ClusterSetArgsUnchanged(argv, argc)) { + RedisModule_Log(mr_staticCtx, "notice", + "Got cluster set command with an unchanged topology, skipping the rebuild"); + return REDISMODULE_OK; + } + if (clusterCtx.CurrCluster && LongFormMyIdUnchanged(argv, argc)) { + ARR(TopologyViewEntry) view = BuildLongFormView(argv, argc); + bool applied = MR_TryApplyTopologyInPlace(view, array_len(view)); + array_free(view); + if (applied) { + ReplaceClusterSetArgs(clusterCtx.CurrCluster, argv, argc); + RedisModule_Log(mr_staticCtx, "notice", + "Got cluster set command with the same shard set, applied the slot map in place"); + return REDISMODULE_OK; + } + } + if (clusterCtx.CurrCluster) + MR_ClusterFree(); SetClusterDataLongForm(argv, argc); return REDISMODULE_OK; } - int res = SetClusterDataShortForm(argv, argc); - if (res == REDISMODULE_OK && clusterCtx.CurrCluster) { - clusterCtx.CurrCluster->shortFormTopology = snapshot; // now owned by the cluster - } else if (snapshot) { - MR_FREE(snapshot); + /* Short form: the arguments only carry credentials, the topology comes + * from the server's cluster state -- so an unchanged command line still + * requires a reconcile against that state (including slotless masters, + * which a CLUSTERSET-built cluster tracks), and changed credentials + * require a full rebuild (every connection re-authenticates). */ + if (ClusterSetArgsUnchanged(argv, argc)) { + ARR(TopologyViewEntry) view = BuildClusterApiView(true); + if (view) { + bool applied = MR_TryApplyTopologyInPlace(view, array_len(view)); + array_free(view); + if (applied) { + RedisModule_Log(mr_staticCtx, "notice", + "Got cluster set command with the same shard set, applied the slot map in place"); + return REDISMODULE_OK; + } + } } - return res; + if (clusterCtx.CurrCluster) + MR_ClusterFree(); + return SetClusterDataShortForm(argv, argc); } /* runs in the event loop so its safe to update cluster diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index 73e9ab84..6b01d394 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -210,7 +210,7 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _cluster_set_args(self, mock_shard_slot_start=8193): + def _cluster_set_args(self, mock_shard_slot_start=8193, mock_shard_id='2', password='password'): # IPv6 endpoints must be bracketed in host:port strings endpoint_host = '[%s]' % self.host if ':' in self.host else self.host # Build arguments according to MR_SetClusterData parser: @@ -228,12 +228,12 @@ def _cluster_set_args(self, mock_shard_slot_start=8193): # Shard 1 (current Redis) - HARDCODED PORT 6379 'SHARD', '1', 'SLOTRANGE', '0', str(mock_shard_slot_start - 1), - 'ADDR', 'password@%s:6379' % endpoint_host, + 'ADDR', '%s@%s:6379' % (password, endpoint_host), 'MASTER', # Shard 2 (mock shard) - 'SHARD', '2', + 'SHARD', mock_shard_id, 'SLOTRANGE', str(mock_shard_slot_start), '16383', - 'ADDR', 'password@%s:%d' % (endpoint_host, self.port), + 'ADDR', '%s@%s:%d' % (password, endpoint_host, self.port), 'MASTER' ] @@ -259,9 +259,9 @@ def __enter__(self): def __exit__(self, type, value, traceback): self.stream_server.stop() - def GetConnection(self, runid='1', sendHelloResponse=True): + def GetConnection(self, runid='1', sendHelloResponse=True, password='password'): conn = self.new_conns.get(block=True, timeout=None) - self.env.assertEqual(conn.read_request(), ['AUTH', 'password']) + self.env.assertEqual(conn.read_request(), ['AUTH', password]) conn.send_status('OK') # auth response if(sendHelloResponse): self.env.assertEqual(conn.read_request(), ['MRTESTS.HELLO']) @@ -683,10 +683,11 @@ def testMassiveClusterSet(env, conn): with ShardMock(env, host) as shardMock: for i in range(1000): conn = shardMock.GetConnection(sendHelloResponse=False) - # Alternate the slot boundary so every command carries a changed - # topology and forces a rebuild — identical re-sends are a no-op. + # Alternate the mock shard's id so every command carries a changed + # shard set and forces a rebuild — an unchanged set (identical or + # merely reshuffled slot ranges) is reconciled in place instead. promote_internal_client_if_supported(env=env) - env.cmd('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_slot_start=8194 - (i % 2))) + env.cmd('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_id=str(2 + (i % 2)))) env.cmd('MRTESTS.FORCESHARDSCONNECTION') @MRTestDecorator(skipOnCluster=True) @@ -768,8 +769,39 @@ def testIdenticalClusterSetIsNoOp(env, conn): time.sleep(0.5) env.assertTrue(shardMock.new_conns.empty()) - # A genuinely different topology (moved slot boundary) must still rebuild. + # Moving the slot boundary between the same two shards is reconciled + # in place: the run id and the live connection survive the reshard. env.expect('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_slot_start=4096)).equal('OK') + env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') + time.sleep(0.5) + env.assertTrue(shardMock.new_conns.empty()) + + # Re-sending the moved boundary is argument-identical again -> no-op + # (the stored command follows what was last applied). + env.expect('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_slot_start=4096)).equal('OK') + env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + time.sleep(0.5) + env.assertTrue(shardMock.new_conns.empty()) + + # A changed shard set (a new shard id) must still tear down and rebuild. + env.expect('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_id='3')).equal('OK') env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) env.cmd('MRTESTS.FORCESHARDSCONNECTION') conn = shardMock.GetConnection() + + +@MRTestDecorator(skipOnCluster=True) +def testClusterSetPasswordChangeRebuilds(env, conn): + for host in _get_hosts(): + with ShardMock(env, host) as shardMock: + conn = shardMock.GetConnection() + + run_id = env.cmd('MRTESTS.INFOCLUSTER')[3] + + # The same shard set at the same addresses but with new credentials: + # every connection must re-authenticate, so a full rebuild is required. + env.expect('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(password='password2')).equal('OK') + env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') + conn = shardMock.GetConnection(password='password2') From 1b5d6c34fcf47d57609474b8b2db5b12975a8186 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 13 Jul 2026 22:55:30 +0300 Subject: [PATCH 12/19] MOD-16399 Fix testMassiveClusterSet alternation phase (CI hang) The rewritten alternation started at mock shard id '2' -- argument- identical to the topology ShardMock.__enter__ had just applied -- so the identical-args skip made iteration 0 a no-op and GetConnection() blocked forever (45-minute CI cancel, buffered stdout hid the earlier results). Start the alternation at '3' so every iteration changes the shard set relative to the last-applied command, and give GetConnection a finite timeout so a regression fails visibly instead of wedging the job. Co-Authored-By: Claude Opus 4.8 --- tests/mr_test_module/pytests/test_network.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index 6b01d394..be2ece89 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -259,8 +259,10 @@ def __enter__(self): def __exit__(self, type, value, traceback): self.stream_server.stop() - def GetConnection(self, runid='1', sendHelloResponse=True, password='password'): - conn = self.new_conns.get(block=True, timeout=None) + def GetConnection(self, runid='1', sendHelloResponse=True, password='password', timeout=60): + # A finite timeout so a test expecting a reconnect that never comes + # fails visibly instead of wedging the whole CI job. + conn = self.new_conns.get(block=True, timeout=timeout) self.env.assertEqual(conn.read_request(), ['AUTH', password]) conn.send_status('OK') # auth response if(sendHelloResponse): @@ -686,8 +688,9 @@ def testMassiveClusterSet(env, conn): # Alternate the mock shard's id so every command carries a changed # shard set and forces a rebuild — an unchanged set (identical or # merely reshuffled slot ranges) is reconciled in place instead. + # Start at '3': the '2' topology was just applied by __enter__. promote_internal_client_if_supported(env=env) - env.cmd('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_id=str(2 + (i % 2)))) + env.cmd('MRTESTS.CLUSTERSET', *shardMock._cluster_set_args(mock_shard_id=str(3 - (i % 2)))) env.cmd('MRTESTS.FORCESHARDSCONNECTION') @MRTestDecorator(skipOnCluster=True) From 20bc188abfc9431a68ba19c44a3bcd5bdecd01f3 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 27 Jul 2026 13:17:33 +0300 Subject: [PATCH 13/19] MOD-16399 Reuse shared topology comparison for CLUSTERSET Build long-form CLUSTERSET into a candidate Cluster and feed it through the same MR_UpdateClusterTopologyIfNeeded path used by topology events and short form. Unchanged topology now preserves connections and in-flight executions; changed nodes, slots, credentials, or MYID still rebuild. Keep the existing CLUSTERSET stress test meaningful by alternating shard IDs and add a focused no-op/password-change regression. --- src/cluster.c | 37 +++++++-------- tests/mr_test_module/pytests/test_network.py | 48 +++++++++++++++----- 2 files changed, 53 insertions(+), 32 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 9624852d..f1ca4295 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1187,8 +1187,15 @@ Cluster* MR_BuildCluster(RedisModuleString** argv, int argc, const char* passwor return cluster; } +static bool SameNullableString(const char* a, const char* b) { + return a == b || (a != NULL && b != NULL && strcmp(a, b) == 0); +} + static bool SameNode(Node* a, Node* b) { - return strcmp(a->id, b->id) == 0 && strcmp(a->ip, b->ip) == 0 && a->port == b->port; + return strcmp(a->id, b->id) == 0 && + strcmp(a->ip, b->ip) == 0 && + a->port == b->port && + SameNullableString(a->password, b->password); } static bool SameSlotRanges(Node* a, Node* b) { @@ -1212,6 +1219,8 @@ static bool SameCluster(Cluster* a, Cluster* b) { return true; if (a == NULL || b == NULL) return false; + if (strcmp(a->myId, b->myId) != 0) + return false; if (mr_dictSize(a->nodes) != mr_dictSize(b->nodes)) return false; mr_dictIterator* iter = mr_dictGetIterator(a->nodes); @@ -1287,9 +1296,6 @@ static int SetClusterDataShortForm(RedisModuleString** argv, int argc){ RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command (short form)"); } - if(clusterCtx.CurrCluster) - MR_ClusterFree(); - // RedisModule_GetClusterNodeSlotRanges may be NULL when the host Redis // build does not export it (e.g. OSS Redis without the backport). Reject // the command with an error instead of crashing or silently no-op'ing, so @@ -1341,12 +1347,8 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command (long form)"); } - if(clusterCtx.CurrCluster) - MR_ClusterFree(); - - clusterCtx.CurrCluster = MR_CALLOC(1, sizeof(*clusterCtx.CurrCluster)); - InitClusterData(clusterCtx.CurrCluster, argv, argc); - memcpy(clusterCtx.myId, clusterCtx.CurrCluster->myId, REDISMODULE_NODE_ID_LEN + 1); + Cluster* cluster = MR_CALLOC(1, sizeof(*cluster)); + InitClusterData(cluster, argv, argc); size_t index = CLUSTERSET_MYID_LONG_FORM_INDEX + 1; const char *token = RedisModule_StringPtrLen(argv[index], NULL); @@ -1382,28 +1384,21 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ continue; // Create a new node or update an existing one - Node* aMasterNode = MR_GetNode(clusterCtx.CurrCluster, realId); + Node* aMasterNode = MR_GetNode(cluster, realId); if(!aMasterNode){ - aMasterNode = MR_CreateNode(clusterCtx.CurrCluster, realId, ip, port, password, NULL, minSlot, maxSlot); + aMasterNode = MR_CreateNode(cluster, realId, ip, port, password, NULL, minSlot, maxSlot); } else { RedisModule_Assert(minSlot <= maxSlot); // slotless nodes are only created (above) mr_listAddNodeTail(aMasterNode->slotRanges, NewSlotRange(minSlot, maxSlot)); } for(int k = minSlot ; k <= maxSlot ; ++k){ - clusterCtx.CurrCluster->slots[k] = aMasterNode; - } - - if (aMasterNode->isMe) { - // fill the fallback single-range; see the comment at the declaration of minSlot and maxSlot - clusterCtx.minSlot = minSlot; - clusterCtx.maxSlot = maxSlot; + cluster->slots[k] = aMasterNode; } index++; } - clusterCtx.clusterSize = mr_dictSize(clusterCtx.CurrCluster->nodes); - mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); + MR_UpdateClusterTopologyIfNeeded(cluster); } static int MR_SetClusterData(RedisModuleString** argv, int argc){ diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index c62e6614..ac6f1110 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -211,15 +211,13 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _send_cluster_set(self): - # try to promote to internal connection - promote_internal_client_if_supported(env=self.env) + def _cluster_set_args(self, mock_shard_id='2', password='password'): # IPv6 endpoints must be bracketed in host:port strings endpoint_host = '[%s]' % self.host if ':' in self.host else self.host # Build arguments according to MR_SetClusterData parser: # argv[6] => myId, argv[7] => "RANGES", argv[8] => numOfRanges, then repeating: # "SHARD" "SLOTRANGE" "ADDR" ["MASTER"] - args = [ + return [ 'NO-USED', # [1] 'NO-USED', # [2] 'NO-USED', # [3] @@ -231,15 +229,20 @@ def _send_cluster_set(self): # Shard 1 (current Redis) - HARDCODED PORT 6379 'SHARD', '1', 'SLOTRANGE', '0', '8192', - 'ADDR', 'password@%s:6379' % endpoint_host, + 'ADDR', '%s@%s:6379' % (password, endpoint_host), 'MASTER', # Shard 2 (mock shard) - 'SHARD', '2', + 'SHARD', mock_shard_id, 'SLOTRANGE', '8193', '16383', - 'ADDR', 'password@%s:%d' % (endpoint_host, self.port), + 'ADDR', '%s@%s:%d' % (password, endpoint_host, self.port), 'MASTER' ] - self.env.cmd('MRTESTS.CLUSTERSET', *args) + + def _send_cluster_set(self, mock_shard_id='2', password='password'): + # try to promote to internal connection + promote_internal_client_if_supported(env=self.env) + self.env.cmd('MRTESTS.CLUSTERSET', + *self._cluster_set_args(mock_shard_id, password)) self.env.cmd('MRTESTS.FORCESHARDSCONNECTION') def __enter__(self): @@ -258,9 +261,9 @@ def __enter__(self): def __exit__(self, type, value, traceback): self.stream_server.stop() - def GetConnection(self, runid='1', sendHelloResponse=True): + def GetConnection(self, runid='1', sendHelloResponse=True, password='password'): conn = self.new_conns.get(block=True, timeout=None) - self.env.assertEqual(conn.read_request(), ['AUTH', 'password']) + self.env.assertEqual(conn.read_request(), ['AUTH', password]) conn.send_status('OK') # auth response if(sendHelloResponse): self.env.assertEqual(conn.read_request(), ['MRTESTS.HELLO']) @@ -793,7 +796,30 @@ def testMassiveClusterSet(env, conn): with ShardMock(env, host) as shardMock: for i in range(1000): conn = shardMock.GetConnection(sendHelloResponse=False) - shardMock._send_cluster_set() + # The previous test relied on every identical CLUSTERSET causing + # a reconnect. Alternate the shard id so this remains a rebuild + # stress test now that no-op updates preserve the cluster. + shardMock._send_cluster_set(mock_shard_id=str(3 - (i % 2))) + + +@MRTestDecorator(skipOnCluster=True) +def testIdenticalClusterSetIsNoOp(env, conn): + for host in _get_hosts(): + with ShardMock(env, host) as shardMock: + conn = shardMock.GetConnection() + run_id = env.cmd('MRTESTS.INFOCLUSTER')[3] + + promote_internal_client_if_supported(env=env) + env.expect('MRTESTS.CLUSTERSET', + *shardMock._cluster_set_args()).equal('OK') + env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + + env.expect('MRTESTS.CLUSTERSET', + *shardMock._cluster_set_args(password='password2')).equal('OK') + env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') + shardMock.GetConnection(password='password2') + @MRTestDecorator(skipOnCluster=True) def testMassiveClusterSetFromShard(env, conn): From 5d6182b7289f50f62fcd13a391eec7422b7ff290 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 27 Jul 2026 13:41:44 +0300 Subject: [PATCH 14/19] MOD-16399 Use long-form argument guard Keep the existing short-form and topology-event comparison unchanged. Skip an identical long-form CLUSTERSET by comparing its stored arguments before parsing or rebuilding. --- src/cluster.c | 68 +++++++++++++++----- tests/mr_test_module/pytests/test_network.py | 18 +++--- 2 files changed, 61 insertions(+), 25 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index f1ca4295..50edf4ef 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1187,15 +1187,8 @@ Cluster* MR_BuildCluster(RedisModuleString** argv, int argc, const char* passwor return cluster; } -static bool SameNullableString(const char* a, const char* b) { - return a == b || (a != NULL && b != NULL && strcmp(a, b) == 0); -} - static bool SameNode(Node* a, Node* b) { - return strcmp(a->id, b->id) == 0 && - strcmp(a->ip, b->ip) == 0 && - a->port == b->port && - SameNullableString(a->password, b->password); + return strcmp(a->id, b->id) == 0 && strcmp(a->ip, b->ip) == 0 && a->port == b->port; } static bool SameSlotRanges(Node* a, Node* b) { @@ -1219,8 +1212,6 @@ static bool SameCluster(Cluster* a, Cluster* b) { return true; if (a == NULL || b == NULL) return false; - if (strcmp(a->myId, b->myId) != 0) - return false; if (mr_dictSize(a->nodes) != mr_dictSize(b->nodes)) return false; mr_dictIterator* iter = mr_dictGetIterator(a->nodes); @@ -1296,6 +1287,9 @@ static int SetClusterDataShortForm(RedisModuleString** argv, int argc){ RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command (short form)"); } + if(clusterCtx.CurrCluster) + MR_ClusterFree(); + // RedisModule_GetClusterNodeSlotRanges may be NULL when the host Redis // build does not export it (e.g. OSS Redis without the backport). Reject // the command with an error instead of crashing or silently no-op'ing, so @@ -1347,8 +1341,12 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command (long form)"); } - Cluster* cluster = MR_CALLOC(1, sizeof(*cluster)); - InitClusterData(cluster, argv, argc); + if(clusterCtx.CurrCluster) + MR_ClusterFree(); + + clusterCtx.CurrCluster = MR_CALLOC(1, sizeof(*clusterCtx.CurrCluster)); + InitClusterData(clusterCtx.CurrCluster, argv, argc); + memcpy(clusterCtx.myId, clusterCtx.CurrCluster->myId, REDISMODULE_NODE_ID_LEN + 1); size_t index = CLUSTERSET_MYID_LONG_FORM_INDEX + 1; const char *token = RedisModule_StringPtrLen(argv[index], NULL); @@ -1384,25 +1382,63 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ continue; // Create a new node or update an existing one - Node* aMasterNode = MR_GetNode(cluster, realId); + Node* aMasterNode = MR_GetNode(clusterCtx.CurrCluster, realId); if(!aMasterNode){ - aMasterNode = MR_CreateNode(cluster, realId, ip, port, password, NULL, minSlot, maxSlot); + aMasterNode = MR_CreateNode(clusterCtx.CurrCluster, realId, ip, port, password, NULL, minSlot, maxSlot); } else { RedisModule_Assert(minSlot <= maxSlot); // slotless nodes are only created (above) mr_listAddNodeTail(aMasterNode->slotRanges, NewSlotRange(minSlot, maxSlot)); } for(int k = minSlot ; k <= maxSlot ; ++k){ - cluster->slots[k] = aMasterNode; + clusterCtx.CurrCluster->slots[k] = aMasterNode; + } + + if (aMasterNode->isMe) { + // fill the fallback single-range; see the comment at the declaration of minSlot and maxSlot + clusterCtx.minSlot = minSlot; + clusterCtx.maxSlot = maxSlot; } index++; } - MR_UpdateClusterTopologyIfNeeded(cluster); + clusterCtx.clusterSize = mr_dictSize(clusterCtx.CurrCluster->nodes); + mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); +} + +/* Long-form CLUSTERSET carries the complete topology in its arguments. MYID + * identifies the receiving shard rather than the topology and is not retained + * in clusterSetCommand (see CopyClusterSetArgs). */ +static bool LongFormClusterSetIsUnchanged(RedisModuleString** argv, int argc){ + Cluster* cluster = clusterCtx.CurrCluster; + if (!IsLongFormClusterSet(argc) || + !cluster || + !cluster->clusterSetCommand || + cluster->clusterSetCommandSize != argc) { + return false; + } + + for (int i = 1; i < argc; ++i) { + if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) + continue; + + size_t argLen; + const char* arg = RedisModule_StringPtrLen(argv[i], &argLen); + if (argLen != strlen(cluster->clusterSetCommand[i]) || + memcmp(arg, cluster->clusterSetCommand[i], argLen) != 0) { + return false; + } + } + return true; } static int MR_SetClusterData(RedisModuleString** argv, int argc){ if (IsLongFormClusterSet(argc)) { + if (LongFormClusterSetIsUnchanged(argv, argc)) { + RedisModule_Log(mr_staticCtx, "notice", + "Got cluster set command with an unchanged topology, skipping the rebuild"); + return REDISMODULE_OK; + } SetClusterDataLongForm(argv, argc); return REDISMODULE_OK; } else if (IsShortFormClusterSet(argc)) { diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index ac6f1110..be0b636a 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -211,7 +211,7 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _cluster_set_args(self, mock_shard_id='2', password='password'): + def _cluster_set_args(self, mock_shard_id='2'): # IPv6 endpoints must be bracketed in host:port strings endpoint_host = '[%s]' % self.host if ':' in self.host else self.host # Build arguments according to MR_SetClusterData parser: @@ -229,20 +229,20 @@ def _cluster_set_args(self, mock_shard_id='2', password='password'): # Shard 1 (current Redis) - HARDCODED PORT 6379 'SHARD', '1', 'SLOTRANGE', '0', '8192', - 'ADDR', '%s@%s:6379' % (password, endpoint_host), + 'ADDR', 'password@%s:6379' % endpoint_host, 'MASTER', # Shard 2 (mock shard) 'SHARD', mock_shard_id, 'SLOTRANGE', '8193', '16383', - 'ADDR', '%s@%s:%d' % (password, endpoint_host, self.port), + 'ADDR', 'password@%s:%d' % (endpoint_host, self.port), 'MASTER' ] - def _send_cluster_set(self, mock_shard_id='2', password='password'): + def _send_cluster_set(self, mock_shard_id='2'): # try to promote to internal connection promote_internal_client_if_supported(env=self.env) self.env.cmd('MRTESTS.CLUSTERSET', - *self._cluster_set_args(mock_shard_id, password)) + *self._cluster_set_args(mock_shard_id)) self.env.cmd('MRTESTS.FORCESHARDSCONNECTION') def __enter__(self): @@ -261,9 +261,9 @@ def __enter__(self): def __exit__(self, type, value, traceback): self.stream_server.stop() - def GetConnection(self, runid='1', sendHelloResponse=True, password='password'): + def GetConnection(self, runid='1', sendHelloResponse=True): conn = self.new_conns.get(block=True, timeout=None) - self.env.assertEqual(conn.read_request(), ['AUTH', password]) + self.env.assertEqual(conn.read_request(), ['AUTH', 'password']) conn.send_status('OK') # auth response if(sendHelloResponse): self.env.assertEqual(conn.read_request(), ['MRTESTS.HELLO']) @@ -815,10 +815,10 @@ def testIdenticalClusterSetIsNoOp(env, conn): env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) env.expect('MRTESTS.CLUSTERSET', - *shardMock._cluster_set_args(password='password2')).equal('OK') + *shardMock._cluster_set_args(mock_shard_id='3')).equal('OK') env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) env.cmd('MRTESTS.FORCESHARDSCONNECTION') - shardMock.GetConnection(password='password2') + shardMock.GetConnection() @MRTestDecorator(skipOnCluster=True) From 4243940d06afba6670625840be72d5d878cdfd48 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 27 Jul 2026 13:17:33 +0300 Subject: [PATCH 15/19] MOD-16399 Reuse shared topology comparison for CLUSTERSET Build long-form CLUSTERSET into a candidate Cluster and feed it through the same MR_UpdateClusterTopologyIfNeeded path used by topology events and short form. Unchanged topology now preserves connections and in-flight executions; changed nodes, slots, credentials, or MYID still rebuild. Keep the existing CLUSTERSET stress test meaningful by alternating shard IDs and add a focused no-op/password-change regression. --- src/cluster.c | 37 +++++++-------- tests/mr_test_module/pytests/test_network.py | 48 +++++++++++++++----- 2 files changed, 53 insertions(+), 32 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 9624852d..f1ca4295 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1187,8 +1187,15 @@ Cluster* MR_BuildCluster(RedisModuleString** argv, int argc, const char* passwor return cluster; } +static bool SameNullableString(const char* a, const char* b) { + return a == b || (a != NULL && b != NULL && strcmp(a, b) == 0); +} + static bool SameNode(Node* a, Node* b) { - return strcmp(a->id, b->id) == 0 && strcmp(a->ip, b->ip) == 0 && a->port == b->port; + return strcmp(a->id, b->id) == 0 && + strcmp(a->ip, b->ip) == 0 && + a->port == b->port && + SameNullableString(a->password, b->password); } static bool SameSlotRanges(Node* a, Node* b) { @@ -1212,6 +1219,8 @@ static bool SameCluster(Cluster* a, Cluster* b) { return true; if (a == NULL || b == NULL) return false; + if (strcmp(a->myId, b->myId) != 0) + return false; if (mr_dictSize(a->nodes) != mr_dictSize(b->nodes)) return false; mr_dictIterator* iter = mr_dictGetIterator(a->nodes); @@ -1287,9 +1296,6 @@ static int SetClusterDataShortForm(RedisModuleString** argv, int argc){ RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command (short form)"); } - if(clusterCtx.CurrCluster) - MR_ClusterFree(); - // RedisModule_GetClusterNodeSlotRanges may be NULL when the host Redis // build does not export it (e.g. OSS Redis without the backport). Reject // the command with an error instead of crashing or silently no-op'ing, so @@ -1341,12 +1347,8 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command (long form)"); } - if(clusterCtx.CurrCluster) - MR_ClusterFree(); - - clusterCtx.CurrCluster = MR_CALLOC(1, sizeof(*clusterCtx.CurrCluster)); - InitClusterData(clusterCtx.CurrCluster, argv, argc); - memcpy(clusterCtx.myId, clusterCtx.CurrCluster->myId, REDISMODULE_NODE_ID_LEN + 1); + Cluster* cluster = MR_CALLOC(1, sizeof(*cluster)); + InitClusterData(cluster, argv, argc); size_t index = CLUSTERSET_MYID_LONG_FORM_INDEX + 1; const char *token = RedisModule_StringPtrLen(argv[index], NULL); @@ -1382,28 +1384,21 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ continue; // Create a new node or update an existing one - Node* aMasterNode = MR_GetNode(clusterCtx.CurrCluster, realId); + Node* aMasterNode = MR_GetNode(cluster, realId); if(!aMasterNode){ - aMasterNode = MR_CreateNode(clusterCtx.CurrCluster, realId, ip, port, password, NULL, minSlot, maxSlot); + aMasterNode = MR_CreateNode(cluster, realId, ip, port, password, NULL, minSlot, maxSlot); } else { RedisModule_Assert(minSlot <= maxSlot); // slotless nodes are only created (above) mr_listAddNodeTail(aMasterNode->slotRanges, NewSlotRange(minSlot, maxSlot)); } for(int k = minSlot ; k <= maxSlot ; ++k){ - clusterCtx.CurrCluster->slots[k] = aMasterNode; - } - - if (aMasterNode->isMe) { - // fill the fallback single-range; see the comment at the declaration of minSlot and maxSlot - clusterCtx.minSlot = minSlot; - clusterCtx.maxSlot = maxSlot; + cluster->slots[k] = aMasterNode; } index++; } - clusterCtx.clusterSize = mr_dictSize(clusterCtx.CurrCluster->nodes); - mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); + MR_UpdateClusterTopologyIfNeeded(cluster); } static int MR_SetClusterData(RedisModuleString** argv, int argc){ diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index c62e6614..ac6f1110 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -211,15 +211,13 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _send_cluster_set(self): - # try to promote to internal connection - promote_internal_client_if_supported(env=self.env) + def _cluster_set_args(self, mock_shard_id='2', password='password'): # IPv6 endpoints must be bracketed in host:port strings endpoint_host = '[%s]' % self.host if ':' in self.host else self.host # Build arguments according to MR_SetClusterData parser: # argv[6] => myId, argv[7] => "RANGES", argv[8] => numOfRanges, then repeating: # "SHARD" "SLOTRANGE" "ADDR" ["MASTER"] - args = [ + return [ 'NO-USED', # [1] 'NO-USED', # [2] 'NO-USED', # [3] @@ -231,15 +229,20 @@ def _send_cluster_set(self): # Shard 1 (current Redis) - HARDCODED PORT 6379 'SHARD', '1', 'SLOTRANGE', '0', '8192', - 'ADDR', 'password@%s:6379' % endpoint_host, + 'ADDR', '%s@%s:6379' % (password, endpoint_host), 'MASTER', # Shard 2 (mock shard) - 'SHARD', '2', + 'SHARD', mock_shard_id, 'SLOTRANGE', '8193', '16383', - 'ADDR', 'password@%s:%d' % (endpoint_host, self.port), + 'ADDR', '%s@%s:%d' % (password, endpoint_host, self.port), 'MASTER' ] - self.env.cmd('MRTESTS.CLUSTERSET', *args) + + def _send_cluster_set(self, mock_shard_id='2', password='password'): + # try to promote to internal connection + promote_internal_client_if_supported(env=self.env) + self.env.cmd('MRTESTS.CLUSTERSET', + *self._cluster_set_args(mock_shard_id, password)) self.env.cmd('MRTESTS.FORCESHARDSCONNECTION') def __enter__(self): @@ -258,9 +261,9 @@ def __enter__(self): def __exit__(self, type, value, traceback): self.stream_server.stop() - def GetConnection(self, runid='1', sendHelloResponse=True): + def GetConnection(self, runid='1', sendHelloResponse=True, password='password'): conn = self.new_conns.get(block=True, timeout=None) - self.env.assertEqual(conn.read_request(), ['AUTH', 'password']) + self.env.assertEqual(conn.read_request(), ['AUTH', password]) conn.send_status('OK') # auth response if(sendHelloResponse): self.env.assertEqual(conn.read_request(), ['MRTESTS.HELLO']) @@ -793,7 +796,30 @@ def testMassiveClusterSet(env, conn): with ShardMock(env, host) as shardMock: for i in range(1000): conn = shardMock.GetConnection(sendHelloResponse=False) - shardMock._send_cluster_set() + # The previous test relied on every identical CLUSTERSET causing + # a reconnect. Alternate the shard id so this remains a rebuild + # stress test now that no-op updates preserve the cluster. + shardMock._send_cluster_set(mock_shard_id=str(3 - (i % 2))) + + +@MRTestDecorator(skipOnCluster=True) +def testIdenticalClusterSetIsNoOp(env, conn): + for host in _get_hosts(): + with ShardMock(env, host) as shardMock: + conn = shardMock.GetConnection() + run_id = env.cmd('MRTESTS.INFOCLUSTER')[3] + + promote_internal_client_if_supported(env=env) + env.expect('MRTESTS.CLUSTERSET', + *shardMock._cluster_set_args()).equal('OK') + env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + + env.expect('MRTESTS.CLUSTERSET', + *shardMock._cluster_set_args(password='password2')).equal('OK') + env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + env.cmd('MRTESTS.FORCESHARDSCONNECTION') + shardMock.GetConnection(password='password2') + @MRTestDecorator(skipOnCluster=True) def testMassiveClusterSetFromShard(env, conn): From 83a5deac8374364922ef04eb30af3f0ac446ef00 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 27 Jul 2026 13:41:44 +0300 Subject: [PATCH 16/19] MOD-16399 Use long-form argument guard Keep the existing short-form and topology-event comparison unchanged. Skip an identical long-form CLUSTERSET by comparing its stored arguments before parsing or rebuilding. --- src/cluster.c | 68 +++++++++++++++----- tests/mr_test_module/pytests/test_network.py | 18 +++--- 2 files changed, 61 insertions(+), 25 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index f1ca4295..50edf4ef 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1187,15 +1187,8 @@ Cluster* MR_BuildCluster(RedisModuleString** argv, int argc, const char* passwor return cluster; } -static bool SameNullableString(const char* a, const char* b) { - return a == b || (a != NULL && b != NULL && strcmp(a, b) == 0); -} - static bool SameNode(Node* a, Node* b) { - return strcmp(a->id, b->id) == 0 && - strcmp(a->ip, b->ip) == 0 && - a->port == b->port && - SameNullableString(a->password, b->password); + return strcmp(a->id, b->id) == 0 && strcmp(a->ip, b->ip) == 0 && a->port == b->port; } static bool SameSlotRanges(Node* a, Node* b) { @@ -1219,8 +1212,6 @@ static bool SameCluster(Cluster* a, Cluster* b) { return true; if (a == NULL || b == NULL) return false; - if (strcmp(a->myId, b->myId) != 0) - return false; if (mr_dictSize(a->nodes) != mr_dictSize(b->nodes)) return false; mr_dictIterator* iter = mr_dictGetIterator(a->nodes); @@ -1296,6 +1287,9 @@ static int SetClusterDataShortForm(RedisModuleString** argv, int argc){ RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command (short form)"); } + if(clusterCtx.CurrCluster) + MR_ClusterFree(); + // RedisModule_GetClusterNodeSlotRanges may be NULL when the host Redis // build does not export it (e.g. OSS Redis without the backport). Reject // the command with an error instead of crashing or silently no-op'ing, so @@ -1347,8 +1341,12 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ RedisModule_Log(mr_staticCtx, "notice", "Got cluster set command (long form)"); } - Cluster* cluster = MR_CALLOC(1, sizeof(*cluster)); - InitClusterData(cluster, argv, argc); + if(clusterCtx.CurrCluster) + MR_ClusterFree(); + + clusterCtx.CurrCluster = MR_CALLOC(1, sizeof(*clusterCtx.CurrCluster)); + InitClusterData(clusterCtx.CurrCluster, argv, argc); + memcpy(clusterCtx.myId, clusterCtx.CurrCluster->myId, REDISMODULE_NODE_ID_LEN + 1); size_t index = CLUSTERSET_MYID_LONG_FORM_INDEX + 1; const char *token = RedisModule_StringPtrLen(argv[index], NULL); @@ -1384,25 +1382,63 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ continue; // Create a new node or update an existing one - Node* aMasterNode = MR_GetNode(cluster, realId); + Node* aMasterNode = MR_GetNode(clusterCtx.CurrCluster, realId); if(!aMasterNode){ - aMasterNode = MR_CreateNode(cluster, realId, ip, port, password, NULL, minSlot, maxSlot); + aMasterNode = MR_CreateNode(clusterCtx.CurrCluster, realId, ip, port, password, NULL, minSlot, maxSlot); } else { RedisModule_Assert(minSlot <= maxSlot); // slotless nodes are only created (above) mr_listAddNodeTail(aMasterNode->slotRanges, NewSlotRange(minSlot, maxSlot)); } for(int k = minSlot ; k <= maxSlot ; ++k){ - cluster->slots[k] = aMasterNode; + clusterCtx.CurrCluster->slots[k] = aMasterNode; + } + + if (aMasterNode->isMe) { + // fill the fallback single-range; see the comment at the declaration of minSlot and maxSlot + clusterCtx.minSlot = minSlot; + clusterCtx.maxSlot = maxSlot; } index++; } - MR_UpdateClusterTopologyIfNeeded(cluster); + clusterCtx.clusterSize = mr_dictSize(clusterCtx.CurrCluster->nodes); + mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); +} + +/* Long-form CLUSTERSET carries the complete topology in its arguments. MYID + * identifies the receiving shard rather than the topology and is not retained + * in clusterSetCommand (see CopyClusterSetArgs). */ +static bool LongFormClusterSetIsUnchanged(RedisModuleString** argv, int argc){ + Cluster* cluster = clusterCtx.CurrCluster; + if (!IsLongFormClusterSet(argc) || + !cluster || + !cluster->clusterSetCommand || + cluster->clusterSetCommandSize != argc) { + return false; + } + + for (int i = 1; i < argc; ++i) { + if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) + continue; + + size_t argLen; + const char* arg = RedisModule_StringPtrLen(argv[i], &argLen); + if (argLen != strlen(cluster->clusterSetCommand[i]) || + memcmp(arg, cluster->clusterSetCommand[i], argLen) != 0) { + return false; + } + } + return true; } static int MR_SetClusterData(RedisModuleString** argv, int argc){ if (IsLongFormClusterSet(argc)) { + if (LongFormClusterSetIsUnchanged(argv, argc)) { + RedisModule_Log(mr_staticCtx, "notice", + "Got cluster set command with an unchanged topology, skipping the rebuild"); + return REDISMODULE_OK; + } SetClusterDataLongForm(argv, argc); return REDISMODULE_OK; } else if (IsShortFormClusterSet(argc)) { diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index ac6f1110..be0b636a 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -211,7 +211,7 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _cluster_set_args(self, mock_shard_id='2', password='password'): + def _cluster_set_args(self, mock_shard_id='2'): # IPv6 endpoints must be bracketed in host:port strings endpoint_host = '[%s]' % self.host if ':' in self.host else self.host # Build arguments according to MR_SetClusterData parser: @@ -229,20 +229,20 @@ def _cluster_set_args(self, mock_shard_id='2', password='password'): # Shard 1 (current Redis) - HARDCODED PORT 6379 'SHARD', '1', 'SLOTRANGE', '0', '8192', - 'ADDR', '%s@%s:6379' % (password, endpoint_host), + 'ADDR', 'password@%s:6379' % endpoint_host, 'MASTER', # Shard 2 (mock shard) 'SHARD', mock_shard_id, 'SLOTRANGE', '8193', '16383', - 'ADDR', '%s@%s:%d' % (password, endpoint_host, self.port), + 'ADDR', 'password@%s:%d' % (endpoint_host, self.port), 'MASTER' ] - def _send_cluster_set(self, mock_shard_id='2', password='password'): + def _send_cluster_set(self, mock_shard_id='2'): # try to promote to internal connection promote_internal_client_if_supported(env=self.env) self.env.cmd('MRTESTS.CLUSTERSET', - *self._cluster_set_args(mock_shard_id, password)) + *self._cluster_set_args(mock_shard_id)) self.env.cmd('MRTESTS.FORCESHARDSCONNECTION') def __enter__(self): @@ -261,9 +261,9 @@ def __enter__(self): def __exit__(self, type, value, traceback): self.stream_server.stop() - def GetConnection(self, runid='1', sendHelloResponse=True, password='password'): + def GetConnection(self, runid='1', sendHelloResponse=True): conn = self.new_conns.get(block=True, timeout=None) - self.env.assertEqual(conn.read_request(), ['AUTH', password]) + self.env.assertEqual(conn.read_request(), ['AUTH', 'password']) conn.send_status('OK') # auth response if(sendHelloResponse): self.env.assertEqual(conn.read_request(), ['MRTESTS.HELLO']) @@ -815,10 +815,10 @@ def testIdenticalClusterSetIsNoOp(env, conn): env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) env.expect('MRTESTS.CLUSTERSET', - *shardMock._cluster_set_args(password='password2')).equal('OK') + *shardMock._cluster_set_args(mock_shard_id='3')).equal('OK') env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) env.cmd('MRTESTS.FORCESHARDSCONNECTION') - shardMock.GetConnection(password='password2') + shardMock.GetConnection() @MRTestDecorator(skipOnCluster=True) From c23cc6a63df21177fa53b4ec8be32b4b1a9c615e Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 27 Jul 2026 13:55:17 +0300 Subject: [PATCH 17/19] MOD-16399 Skip identical long-form CLUSTERSET Compare the long-form command with the stored arguments before parsing or rebuilding. Keep short-form and topology-event handling unchanged, and preserve the existing cluster for an identical DMC re-broadcast. --- src/cluster.c | 23 ++++++++++++++++++++ tests/mr_test_module/pytests/test_network.py | 22 ++++++++++++++++--- 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 9624852d..4501d9dc 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1406,8 +1406,31 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } +/* MYID identifies the receiving shard and is not retained in clusterSetCommand. */ +static bool IsSameLongFormClusterSet(RedisModuleString** argv, int argc){ + Cluster* current = clusterCtx.CurrCluster; + if (!current || + !current->clusterSetCommand || + current->clusterSetCommandSize != argc) + return false; + + for (int i = 1; i < argc; ++i) { + if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) + continue; + const char* arg = RedisModule_StringPtrLen(argv[i], NULL); + if (strcmp(arg, current->clusterSetCommand[i]) != 0) + return false; + } + return true; +} + static int MR_SetClusterData(RedisModuleString** argv, int argc){ if (IsLongFormClusterSet(argc)) { + if (IsSameLongFormClusterSet(argv, argc)) { + RedisModule_Log(mr_staticCtx, "notice", + "Skipping identical long-form cluster set"); + return REDISMODULE_OK; + } SetClusterDataLongForm(argv, argc); return REDISMODULE_OK; } else if (IsShortFormClusterSet(argc)) { diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index c62e6614..c5e5b489 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -211,7 +211,7 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _send_cluster_set(self): + def _send_cluster_set(self, mock_shard_id='2'): # try to promote to internal connection promote_internal_client_if_supported(env=self.env) # IPv6 endpoints must be bracketed in host:port strings @@ -234,7 +234,7 @@ def _send_cluster_set(self): 'ADDR', 'password@%s:6379' % endpoint_host, 'MASTER', # Shard 2 (mock shard) - 'SHARD', '2', + 'SHARD', mock_shard_id, 'SLOTRANGE', '8193', '16383', 'ADDR', 'password@%s:%d' % (endpoint_host, self.port), 'MASTER' @@ -793,7 +793,23 @@ def testMassiveClusterSet(env, conn): with ShardMock(env, host) as shardMock: for i in range(1000): conn = shardMock.GetConnection(sendHelloResponse=False) - shardMock._send_cluster_set() + # Keep exercising rebuilds now that identical updates are skipped. + shardMock._send_cluster_set(mock_shard_id=str(3 - (i % 2))) + + +@MRTestDecorator(skipOnCluster=True) +def testIdenticalLongFormClusterSetIsNoOp(env, conn): + for host in _get_hosts(): + with ShardMock(env, host) as shardMock: + shardMock.GetConnection() + run_id = env.cmd('MRTESTS.INFOCLUSTER')[3] + + shardMock._send_cluster_set() + env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + + shardMock._send_cluster_set(mock_shard_id='3') + env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + shardMock.GetConnection() @MRTestDecorator(skipOnCluster=True) def testMassiveClusterSetFromShard(env, conn): From 60a4f327d126068daecdd0efdc01689e2031cb77 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 27 Jul 2026 17:29:42 +0300 Subject: [PATCH 18/19] MOD-16399 Compare long-form arguments by length Use RedisModuleString lengths with memcmp so embedded NUL bytes safely force a rebuild. Add a regression case covering the binary-string mismatch. --- src/cluster.c | 6 ++++-- tests/mr_test_module/pytests/test_network.py | 11 +++++++++-- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 4501d9dc..e52414c5 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1417,8 +1417,10 @@ static bool IsSameLongFormClusterSet(RedisModuleString** argv, int argc){ for (int i = 1; i < argc; ++i) { if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) continue; - const char* arg = RedisModule_StringPtrLen(argv[i], NULL); - if (strcmp(arg, current->clusterSetCommand[i]) != 0) + size_t argLen; + const char* arg = RedisModule_StringPtrLen(argv[i], &argLen); + if (argLen != strlen(current->clusterSetCommand[i]) || + memcmp(arg, current->clusterSetCommand[i], argLen) != 0) return false; } return true; diff --git a/tests/mr_test_module/pytests/test_network.py b/tests/mr_test_module/pytests/test_network.py index c5e5b489..a8d6b3f6 100644 --- a/tests/mr_test_module/pytests/test_network.py +++ b/tests/mr_test_module/pytests/test_network.py @@ -211,7 +211,7 @@ def _handle_conn(self, sock, client_addr): conn = Connection(sock) self.new_conns.put(conn) - def _send_cluster_set(self, mock_shard_id='2'): + def _send_cluster_set(self, mock_shard_id='2', first_arg='NO-USED'): # try to promote to internal connection promote_internal_client_if_supported(env=self.env) # IPv6 endpoints must be bracketed in host:port strings @@ -220,7 +220,7 @@ def _send_cluster_set(self, mock_shard_id='2'): # argv[6] => myId, argv[7] => "RANGES", argv[8] => numOfRanges, then repeating: # "SHARD" "SLOTRANGE" "ADDR" ["MASTER"] args = [ - 'NO-USED', # [1] + first_arg, # [1] 'NO-USED', # [2] 'NO-USED', # [3] 'NO-USED', # [4] @@ -807,6 +807,13 @@ def testIdenticalLongFormClusterSetIsNoOp(env, conn): shardMock._send_cluster_set() env.assertEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + # RedisModuleString can carry embedded NUL bytes. Its explicit length + # must participate in the comparison so this safely rebuilds. + shardMock._send_cluster_set(first_arg=b'NO-USED\0changed') + env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) + shardMock.GetConnection() + + run_id = env.cmd('MRTESTS.INFOCLUSTER')[3] shardMock._send_cluster_set(mock_shard_id='3') env.assertNotEqual(env.cmd('MRTESTS.INFOCLUSTER')[3], run_id) shardMock.GetConnection() From 4d1759e6e853d5f248173fa39dd74d800fc4d41a Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Mon, 27 Jul 2026 18:01:55 +0300 Subject: [PATCH 19/19] MOD-16399 Place MYID comment by comparison guard --- src/cluster.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cluster.c b/src/cluster.c index e52414c5..cd60a402 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1406,7 +1406,6 @@ static void SetClusterDataLongForm(RedisModuleString** argv, int argc){ mr_dictEmpty(clusterCtx.nodesMsgIds, NULL); } -/* MYID identifies the receiving shard and is not retained in clusterSetCommand. */ static bool IsSameLongFormClusterSet(RedisModuleString** argv, int argc){ Cluster* current = clusterCtx.CurrCluster; if (!current || @@ -1415,6 +1414,7 @@ static bool IsSameLongFormClusterSet(RedisModuleString** argv, int argc){ return false; for (int i = 1; i < argc; ++i) { + /* MYID identifies the receiving shard and is not retained in clusterSetCommand. */ if (i == CLUSTERSET_MYID_LONG_FORM_INDEX) continue; size_t argLen;