Improve timeout check for FT.AGGREGATE in cluster mode - [MOD-12434] (#7359)

GuyAv46 · web-flow · commit 78329f6e9d39 · 2025-11-17T16:19:20.000Z
* allow empty replies to be pushed to the query channel

* fix empty reply check

* actually fix empty reply check

* address review comment
diff --git a/coord/src/dist_aggregate.c b/coord/src/dist_aggregate.c
@@ -130,7 +130,7 @@ static void netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep) {
     MRReply *results = NULL;
     if (map && MRReply_Type(map) == MR_REPLY_MAP) {
       results = MRReply_MapElement(map, "results");
-      if (results && MRReply_Type(results) == MR_REPLY_ARRAY && MRReply_Length(results) > 0) {
+      if (results && MRReply_Type(results) == MR_REPLY_ARRAY) {
         MRIteratorCallback_AddReply(ctx, rep); // to be picked up by getNextReply
         // User code now owns the reply, so we can't free it here ourselves!
         rep = NULL;
@@ -140,7 +140,7 @@ static void netCursorCallback(MRIteratorCallbackCtx *ctx, MRReply *rep) {
   else // RESP2
   {
     MRReply *results = MRReply_ArrayElement(rep, 0);
-    if (results && MRReply_Type(results) == MR_REPLY_ARRAY && MRReply_Length(results) > 1) {
+    if (results && MRReply_Type(results) == MR_REPLY_ARRAY && MRReply_Length(results) >= 1) {
       MRIteratorCallback_AddReply(ctx, rep); // to be picked up by getNextReply
       // User code now owns the reply, so we can't free it here ourselves!
       rep = NULL;
@@ -265,13 +265,20 @@ static int getNextReply(RPNet *nc) {
   }
 
   MRReply *rows = MRReply_ArrayElement(root, 0);
-  if (   rows == NULL
-      || (MRReply_Type(rows) != MR_REPLY_ARRAY && MRReply_Type(rows) != MR_REPLY_MAP)
-      || MRReply_Length(rows) == 0) {
+  // Perform sanity check to avoid processing empty replies
+  bool is_empty;
+  if (nc->cmd.protocol == 3) { // RESP3
+    MRReply *results = MRReply_MapElement(rows, "results");
+    is_empty = MRReply_Length(results) == 0;
+  } else { // RESP2
+    is_empty = MRReply_Length(rows) == 1;
+  }
+
+  if (is_empty) {
     MRReply_Free(root);
     root = NULL;
     rows = NULL;
-    RedisModule_Log(RSDummyContext, "warning", "An empty reply was received from a shard");
+    RedisModule_Log(RSDummyContext, "verbose", "An empty reply was received from a shard");
   }
 
   // invariant: either rows == NULL or least one row exists
