Fix edge case in clusterset - [MOD-13562] (#8068)

GuyAv46 · github-actions[bot] · commit c3490311801f · 2026-01-19T17:41:52.000Z
* collect all shards (including replicas and no slots) and filter only at the end * Add API description * Align usage * add a test * avoid string arg consuming of standalone flag params * fix unit tests * add another test * align and add flow tests * add another case (cherry picked from commit a2279bd)
diff --git a/src/coord/rmr/redise.c b/src/coord/rmr/redise.c
@@ -15,6 +15,7 @@
 typedef struct {
   arrayof(RedisModuleSlotRange) slotRanges;
   MRClusterNode node;
+  bool isMaster;
 } RLShard;
 
 static void RLShard_Free(RLShard *sh) {
@@ -81,18 +82,22 @@ static void MRTopology_AddRLShard(MRClusterTopology *t, RLShard *sh) {
     }                                                     \
   })
 
+#define STR_MATCH(str, len, lit) (sizeof(lit) - 1 == len && strcasecmp(str, lit) == 0)
+
 MRClusterTopology *RedisEnterprise_ParseTopology(RedisModuleCtx *ctx, RedisModuleString **argv,
                                                  int argc, uint32_t *my_shard_idx) {
   ArgsCursor ac; // Name is important for error macros, same goes for `ctx`
   ArgsCursor_InitRString(&ac, argv, argc);
   AC_Advance(&ac); // Skip command name
+  size_t myID_offset;
   const char *myID = NULL;                 // Mandatory. No default.
   uint32_t numRanges = 0;                  // Mandatory. No default.
   uint32_t numSlots = 16384;               // Default.
 
   // Parse general arguments. No allocation is done here, so we can just return on error
   while (!AC_IsAtEnd(&ac)) {
     if (AC_AdvanceIfMatch(&ac, "MYID")) {
+      myID_offset = ac.offset;
       myID = AC_GetStringNC(&ac, NULL);  // Verified after breaking out of loop
     } else if (AC_AdvanceIfMatch(&ac, "HASHFUNC")) {
       const char *hashFuncStr = AC_GetStringNC(&ac, NULL);
@@ -156,7 +161,6 @@ MRClusterTopology *RedisEnterprise_ParseTopology(RedisModuleCtx *ctx, RedisModul
     const char *idstr = RedisModule_StringPtrLen(shardIDStr, &len);
     sh->node.id = rm_strndup(idstr, len);
 
-    bool is_master = false;
     while (!AC_IsAtEnd(&ac)) {
       if (AC_AdvanceIfMatch(&ac, "SLOTRANGE")) {
         if (array_len(sh->slotRanges) > 0) {
@@ -196,39 +200,27 @@ MRClusterTopology *RedisEnterprise_ParseTopology(RedisModuleCtx *ctx, RedisModul
 
       } else if (AC_AdvanceIfMatch(&ac, "UNIXADDR")) {
         /* Optional UNIXADDR <unix_addr> */
+        if (sh->node.endpoint.unixSock) {
+          ERROR_FMT("Multiple UNIXADDR specified for shard `%s`", sh->node.id);
+          goto error;
+        }
         size_t len;
         const char *unixSock;
-        if (!(unixSock = AC_GetStringNC(&ac, &len))) {
+        if (AC_GetString(&ac, &unixSock, &len, AC_F_NOADVANCE) != AC_OK ||
+            STR_MATCH(unixSock, len, "MASTER") || // Avoid consuming MASTER flag argument
+            STR_MATCH(unixSock, len, "SHARD")) {  // Avoid consuming next SHARD marker argument
           ERROR_MISSING("UNIXADDR");
           goto error;
         }
-        if (sh->node.endpoint.unixSock) {
-          ERROR_FMT("Multiple UNIXADDR specified for shard `%s`", sh->node.id);
-          goto error;
-        }
         sh->node.endpoint.unixSock = rm_strndup(unixSock, len);
-
+        AC_Advance(&ac);
       } else if (AC_AdvanceIfMatch(&ac, "MASTER")) {
-        is_master = true;
+        sh->isMaster = true;
       } else {
         break;
       }
     }
 
-    // We don't care for replicas using this command anymore
-    if (!is_master) {
-      RLShard_Free(sh);
-      sh = NULL;
-      continue;
-    }
-
-    // Ignore shards with no slot ranges (like replicas)
-    if (array_len(sh->slotRanges) == 0) {
-      RLShard_Free(sh);
-      sh = NULL;
-      continue;
-    }
-
     dictEntry *entry = dictAddOrFind(shards, shardIDStr);
     if (!dictGetVal(entry)) {
       // New shard
@@ -270,15 +262,18 @@ MRClusterTopology *RedisEnterprise_ParseTopology(RedisModuleCtx *ctx, RedisModul
           goto error;
         }
       }
-      RS_ASSERT(array_len(sh->slotRanges) == 1);
-      // Verify slot range starts past existing ones
-      if (array_tail(existing_shard->slotRanges).end + 1 >= sh->slotRanges[0].start) {
-        ERROR_FMT("SLOTRANGE out of order for shard `%s`", sh->node.id);
-        goto error;
-      }
 
-      // Append new slot range
-      array_ensure_append_1(existing_shard->slotRanges, sh->slotRanges[0]);
+      RS_ASSERT(array_len(sh->slotRanges) <= 1);
+      if (array_len(sh->slotRanges) == 1) {
+        // Verify slot range starts past existing ones
+        if (array_len(existing_shard->slotRanges) > 0 && array_tail(existing_shard->slotRanges).end + 1 >= sh->slotRanges[0].start) {
+          ERROR_FMT("SLOTRANGE out of order for shard `%s`", sh->node.id);
+          goto error;
+        }
+
+        // Append new slot range
+        array_ensure_append_1(existing_shard->slotRanges, sh->slotRanges[0]);
+      }
 
       // Discard parsed shard
       RLShard_Free(sh);
@@ -294,13 +289,17 @@ MRClusterTopology *RedisEnterprise_ParseTopology(RedisModuleCtx *ctx, RedisModul
   }
 
   // Now, build the topology.
-  // 1. All shards in the dict are valid masters
+  // 1. All shards in the dict are valid
   // 2. We can identify my shard by myID
   topo = MR_NewTopology(dictSize(shards));
   dictIterator *iter = dictGetIterator(shards);
   dictEntry *de;
   while ((de = dictNext(iter)) != NULL) {
-    MRTopology_AddRLShard(topo, dictGetVal(de));
+    RLShard *sh = dictGetVal(de);
+    // Only add master shards with slots
+    if (sh->isMaster && array_len(sh->slotRanges) > 0) {
+      MRTopology_AddRLShard(topo, sh);
+    }
   }
   dictReleaseIterator(iter);
 
@@ -315,10 +314,14 @@ MRClusterTopology *RedisEnterprise_ParseTopology(RedisModuleCtx *ctx, RedisModul
   }
 
   if (*my_shard_idx == UINT32_MAX) {
-    ERROR_FMT("MYID `%s` does not correspond to any shard", myID);
-    MRClusterTopology_Free(topo);
-    topo = NULL;
-    goto error;
+    // if MyID corresponds to some shard in the dict, this is NOT an error:
+    // It means the local node is not part of the topology we store (e.g., it has no slot, or is a replica)
+    if (dictFind(shards, argv[myID_offset]) == NULL) {
+      ERROR_FMT("MYID `%s` does not correspond to any shard", myID);
+      MRClusterTopology_Free(topo);
+      topo = NULL;
+      goto error;
+    }
   }
 
 error: // Also the normal exit point
diff --git a/src/coord/rmr/redise.h b/src/coord/rmr/redise.h
@@ -15,6 +15,14 @@
 extern "C" {
 #endif
 
+/* Parse the cluster topology from the given arguments.
+ * On success, returns the parsed topology. On failure, replies with an error
+ * using the provided context and returns NULL.
+ *
+ * The `my_shard_idx` output parameter is set to the index of the shard
+ * corresponding to MYID, or UINT32_MAX if MYID does not correspond to any shard
+ * in the topology.
+ */
 MRClusterTopology *RedisEnterprise_ParseTopology(RedisModuleCtx *ctx, RedisModuleString **argv, int argc, uint32_t *my_shard_idx);
 
 #ifdef __cplusplus
diff --git a/src/module.c b/src/module.c
@@ -4022,15 +4022,23 @@ int SetClusterCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
 
   RedisModule_Log(ctx, "notice", "Received new cluster topology with %u shards (%s)", topo->numShards, ranges_info);
 
-  // Take a reference to our own shard slot ranges (MR_UpdateTopology won't consume it)
-  RS_ASSERT(my_shard_idx < topo->numShards);
-  const RedisModuleSlotRangeArray *my_slots = topo->shards[my_shard_idx].slotRanges;
+  if (my_shard_idx != UINT32_MAX) {
+    // Take a reference to our own shard slot ranges (MR_UpdateTopology won't consume it)
+    RS_ASSERT(my_shard_idx < topo->numShards);
+    const RedisModuleSlotRangeArray *my_slots = topo->shards[my_shard_idx].slotRanges;
 
-  // Store the local shard id
-  MR_SetLocalNodeId(topo->shards[my_shard_idx].node.id);
+    // Store the local shard id
+    MR_SetLocalNodeId(topo->shards[my_shard_idx].node.id);
 
-  // send the topology to the cluster
-  MR_UpdateTopology(topo, my_slots);
+    // send the topology to the cluster
+    MR_UpdateTopology(topo, my_slots);
+
+  } else {
+    // Valid topology but this node is not part of it.
+    // We cannot pass NULL as local slots, so we pass an empty slot array.
+    static const RedisModuleSlotRangeArray empty_slots = {0, {{0, 0}}};
+    MR_UpdateTopology(topo, &empty_slots);
+  }
   return RedisModule_ReplyWithSimpleString(ctx, "OK");
 }
 
diff --git a/tests/cpptests/coord_tests/test_cpp_clusterset.cpp b/tests/cpptests/coord_tests/test_cpp_clusterset.cpp
@@ -746,7 +746,7 @@ TEST_F(ClusterSetTest, Error_MissingUNIXADDRValue) {
     MRClusterTopology *topo = RedisEnterprise_ParseTopology(ctx, argv, argv.size(), &my_shard_idx);
 
     EXPECT_EQ(topo, nullptr);
-    EXPECT_EQ(RMCK_GetLastError(ctx), "MYID `shard1` does not correspond to any shard at offset 14");
+    EXPECT_EQ(RMCK_GetLastError(ctx), "Missing value for UNIXADDR at offset 13");
 
 }
 
@@ -806,7 +806,7 @@ TEST_F(ClusterSetTest, Error_MultipleUNIXADDR_SameBlock) {
     MRClusterTopology *topo = RedisEnterprise_ParseTopology(ctx, argv, argv.size(), &my_shard_idx);
 
     EXPECT_EQ(topo, nullptr);
-    EXPECT_EQ(RMCK_GetLastError(ctx), "Multiple UNIXADDR specified for shard `shard1` at offset 16");
+    EXPECT_EQ(RMCK_GetLastError(ctx), "Multiple UNIXADDR specified for shard `shard1` at offset 15");
 
 }
 
@@ -994,3 +994,56 @@ TEST_F(ClusterSetTest, EdgeCase_ManyShards) {
 
     MRClusterTopology_Free(topo);
 }
+
+TEST_F(ClusterSetTest, EdgeCase_LocalShardEmpty) {
+    std::vector<std::string> args = {
+        "search.CLUSTERSET",
+        "MYID", "local_shard",
+        "RANGES", "3",
+        "SHARD", "local_shard", "ADDR", "127.0.0.3:6379", "MASTER",  // No SLOTRANGE - empty shard
+        "SHARD", "shard1", "SLOTRANGE", "0", "8191", "ADDR", "127.0.0.1:6379", "MASTER",
+        "SHARD", "shard2", "SLOTRANGE", "8192", "16383", "ADDR", "127.0.0.2:6379", "MASTER"
+    };
+
+    ArgvList argv(ctx, args);
+    uint32_t my_shard_idx = UINT32_MAX;
+    MRClusterTopology *topo = RedisEnterprise_ParseTopology(ctx, argv, argv.size(), &my_shard_idx);
+
+    ASSERT_NE(topo, nullptr) << "Parsing should succeed with empty local shard";
+    EXPECT_EQ(my_shard_idx, UINT32_MAX) << "Should not find local shard in topology (empty shard ignored)";
+    EXPECT_EQ(topo->numShards, 2) << "Should have 2 shards (empty shard ignored)";
+
+    // Verify that the local shard is not part of the topology
+    for (uint32_t i = 0; i < topo->numShards; i++) {
+        EXPECT_STRNE(topo->shards[i].node.id, "local_shard")
+            << "Local shard should not be in topology";
+    }
+
+    MRClusterTopology_Free(topo);
+}
+
+TEST_F(ClusterSetTest, EdgeCase_LocalShardReplica) {
+    std::vector<std::string> args = {
+        "search.CLUSTERSET",
+        "MYID", "local_shard",
+        "RANGES", "2",
+        "SHARD", "local_shard", "ADDR", "127.0.0.3:6379", "SLOTRANGE", "0", "16383",
+        "SHARD", "other_shard", "ADDR", "127.0.0.1:6381", "SLOTRANGE", "0", "16383", "MASTER",
+    };
+
+    ArgvList argv(ctx, args);
+    uint32_t my_shard_idx = UINT32_MAX;
+    MRClusterTopology *topo = RedisEnterprise_ParseTopology(ctx, argv, argv.size(), &my_shard_idx);
+
+    ASSERT_NE(topo, nullptr) << "Parsing should succeed with empty local shard";
+    EXPECT_EQ(my_shard_idx, UINT32_MAX) << "Should not find local shard in topology (empty shard ignored)";
+    EXPECT_EQ(topo->numShards, 1) << "Should have 1 shard (empty shard ignored)";
+
+    // Verify that the local shard is not part of the topology
+    for (uint32_t i = 0; i < topo->numShards; i++) {
+        EXPECT_STRNE(topo->shards[i].node.id, "local_shard")
+            << "Local shard should not be in topology";
+    }
+
+    MRClusterTopology_Free(topo);
+}
diff --git a/tests/pytests/test.py b/tests/pytests/test.py
@@ -4313,6 +4313,58 @@ def test_cluster_set_multiple_slots(env: Env):
     ]
     env.expect('SEARCH.CLUSTERINFO').equal(expected)
 
+@skip(cluster=False) # this test is only relevant on cluster
+def test_cluster_set_myself_excluded(env: Env):
+    env.cmd(debug_cmd(), 'PAUSE_TOPOLOGY_UPDATER')
+    env.cmd(config_cmd(), 'SET', 'TOPOLOGY_VALIDATION_TIMEOUT', 1)
+
+    # Set two shards, one with all the slots, and one without any slots
+    env.expect(
+        'SEARCH.CLUSTERSET',
+            'HASHFUNC', 'CRC16',
+            'NUMSLOTS', '16384',
+            'MYID', '1',
+            'RANGES', '2',
+            'SHARD', '1',
+            'ADDR', 'localhost:7001',
+            'MASTER',
+            'SHARD', '2',
+            'ADDR', 'localhost:7002',
+            'MASTER',
+            'SLOTRANGE', '0', '16383'
+    ).ok()
+
+    # Expect only the shard with slots to be listed
+    expected = [
+        'num_partitions', 1,
+        'cluster_type', 'redis_oss',
+        'shards', [['slots', [0, 16383], 'id', '2', 'host', 'localhost', 'port', 7002]],
+    ]
+    env.expect('SEARCH.CLUSTERINFO').equal(expected)
+
+    # Set two shards, one master and myself as replica
+    env.expect(
+        'SEARCH.CLUSTERSET',
+            'HASHFUNC', 'CRC16',
+            'NUMSLOTS', '16384',
+            'MYID', '1',
+            'RANGES', '2',
+            'SHARD', '1',
+            'ADDR', 'localhost:7001',
+            'SLOTRANGE', '0', '16383',
+            'SHARD', '2',
+            'ADDR', 'localhost:7002',
+            'MASTER',
+            'SLOTRANGE', '0', '16383'
+    ).ok()
+
+    # Expect only the master shard to be listed
+    expected = [
+        'num_partitions', 1,
+        'cluster_type', 'redis_oss',
+        'shards', [['slots', [0, 16383], 'id', '2', 'host', 'localhost', 'port', 7002]],
+    ]
+    env.expect('SEARCH.CLUSTERINFO').equal(expected)
 
 @skip(cluster=False) # this test is only relevant on cluster
 def test_cluster_set_errors(env: Env):
@@ -4339,7 +4391,7 @@ def test_cluster_set_errors(env: Env):
     env.expect('SEARCH.CLUSTERSET', 'MYID', '1', 'RANGES', '1',
                'SHARD').error().contains('Missing value for SHARD')
     env.expect('SEARCH.CLUSTERSET', 'MYID', '1', 'RANGES', '1',
-               'SHARD', '1').error().contains('MYID `1` does not correspond to any shard')
+               'SHARD', '1').error().contains('Missing value for ADDR at offset 7')
     env.expect('SEARCH.CLUSTERSET', 'MYID', '1', 'RANGES', '1',
                'SHARD', '1', 'SLOTRANGE').error().contains('Missing value for SLOTRANGE')
     env.expect('SEARCH.CLUSTERSET', 'MYID', '1', 'RANGES', '1',
