fix: XREADGROUP key extraction for STREAMS consumer name (#7721)

Author: vyavdoshenko

src/server/stream_family_test.cc

@@ -299,6 +299,18 @@ TEST_F(StreamFamilyTest, XReadGroup) {
   EXPECT_THAT(resp, ArgType(RespExpr::NIL_ARRAY));
 }
 
+TEST_F(StreamFamilyTest, XReadGroupConsumerNamedStreams) {
+  Run({"XADD", "COUNT", "1-0", "field", "value"});
+  Run({"XGROUP", "CREATE", "COUNT", "grp1", "0"});
+  Run({"XADD", "2", "1-0", "field", "value"});
+  Run({"XGROUP", "CREATE", "2", "grp1", "0"});
+  Run({"XADD", "xp1", "1-0", "field", "value"});
+  Run({"XGROUP", "CREATE", "xp1", "grp1", "0"});
+
+  auto resp = Run({"XREADGROUP", "GROUP", "grp1", "STREAMS", "COUNT", "2", "STREAMS", "xp1", ">"});
+  EXPECT_THAT(resp, RespElementsAre(RespArray(ElementsAre("xp1", ArrLen(1)))));
+}
+
 TEST_F(StreamFamilyTest, XReadBlock) {
   Run({"xadd", "foo", "1-*", "k1", "v1"});
   Run({"xadd", "foo", "1-*", "k2", "v2"});

src/server/transaction.cc

@@ -1652,14 +1652,46 @@ OpResult<KeyIndex> DetermineKeys(const CommandId* cid, const facade::ParsedArgs&
 
     string_view name{cid->name()};
 
-    // Determine based on STREAMS argument position
+    // Determine based on the STREAMS option position.
     if (name == "XREAD" || name == "XREADGROUP") {
-      for (size_t i = 0; i < args.size(); ++i) {
+      size_t i = 0;
+      if (name == "XREADGROUP") {
+        i = 3;  // GROUP <group> <consumer>
+      }
+
+      while (i < args.size()) {
         string_view arg = args[i];
         if (absl::EqualsIgnoreCase(arg, "STREAMS")) {
           size_t left = args.size() - i - 1;
           return KeyIndex(i + 1, i + 1 + (left / 2));
         }
+
+        if (absl::EqualsIgnoreCase(arg, "COUNT")) {
+          if (i + 1 >= args.size())
+            return OpStatus::SYNTAX_ERR;
+          uint32_t count;
+          if (!absl::SimpleAtoi(args[i + 1], &count))
+            return OpStatus::INVALID_INT;
+          i += 2;
+          continue;
+        }
+
+        if (absl::EqualsIgnoreCase(arg, "BLOCK")) {
+          if (i + 1 >= args.size())
+            return OpStatus::SYNTAX_ERR;
+          int64_t timeout;
+          if (!absl::SimpleAtoi(args[i + 1], &timeout))
+            return OpStatus::INVALID_INT;
+          i += 2;
+          continue;
+        }
+
+        if (name == "XREADGROUP" && absl::EqualsIgnoreCase(arg, "NOACK")) {
+          ++i;
+          continue;
+        }
+
+        break;
       }
       return OpStatus::SYNTAX_ERR;
     }