fix(stream): skip metadata subkeys during boundary recalculation

Author: kirito632

src/commands/cmd_stream.cc

@@ -309,6 +309,10 @@ class CommandXDelEx : public Commander {
       entry_ids_.emplace_back(id);
     }
 
+    if (parser.Good()) {
+      return parser.InvalidSyntax();
+    }
+
     return Status::OK();
   }
 

src/storage/batch_extractor.cc

@@ -270,6 +270,7 @@ rocksdb::Status WriteBatchExtractor::PutCF(uint32_t column_family_id, const Slic
     Slice entry_id_check = ikey.GetSubKey();
     uint64_t delimiter = 0;
     GetFixed64(&entry_id_check, &delimiter);
+    // Non-entry stream metadata uses UINT64_MAX-prefixed subkeys.
     if (delimiter == UINT64_MAX) {
       return rocksdb::Status::OK();
     }
@@ -406,6 +407,7 @@ rocksdb::Status WriteBatchExtractor::DeleteCF(uint32_t column_family_id, const S
     redis::StreamEntryID entry_id;
     GetFixed64(&encoded_id, &entry_id.ms);
 
+    // Non-entry stream metadata uses UINT64_MAX-prefixed subkeys.
     if (entry_id.ms == UINT64_MAX) {
       return rocksdb::Status::OK();
     }

src/types/redis_stream.cc

@@ -416,8 +416,9 @@ rocksdb::Status Stream::getGroupNames(engine::Context &ctx, const std::string &n
 
   auto iter = util::UniqueIterator(ctx, read_options, stream_cf_handle_);
   for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+    // Group metadata keys sort before consumer/PEL keys, so no more groups once we see a different type.
     if (identifySubkeyType(iter->key()) != StreamSubkeyType::StreamConsumerGroupMetadata) {
-      continue;
+      break;
     }
     group_names->push_back(groupNameFromInternalKey(iter->key()));
   }
@@ -452,6 +453,7 @@ rocksdb::Status Stream::processOneEntryDeletion(
   }
 
   if (s.IsNotFound()) {
+    // DelRef cleans orphan PELs; ACKED preserves them without the entry.
     if (option == StreamDeleteOption::DelRef) {
       s = cleanPelFromAllGroups(ctx, ns_key, *metadata, id, batch, batch_modified, all_groups, group_pending_decrements,
                                 consumer_pending_decrements);
@@ -683,7 +685,9 @@ rocksdb::Status Stream::DeleteEntriesWithOption(engine::Context &ctx, const Slic
 
       if (first_deleted) {
         iter->SeekToFirst();
-        while (iter->Valid() && deleted_entry_keys.count(iter->key().ToString()) > 0) {
+        // Skip non-entry subkeys (group/consumer/PEL metadata) when recalculating stream boundaries.
+        while (iter->Valid() && (identifySubkeyType(iter->key()) != StreamSubkeyType::StreamEntry ||
+                                 deleted_entry_keys.count(iter->key().ToString()) > 0)) {
           iter->Next();
         }
         if (iter->Valid()) {
@@ -696,7 +700,9 @@ rocksdb::Status Stream::DeleteEntriesWithOption(engine::Context &ctx, const Slic
       }
       if (last_deleted) {
         iter->SeekToLast();
-        while (iter->Valid() && deleted_entry_keys.count(iter->key().ToString()) > 0) {
+        // Skip non-entry subkeys (group/consumer/PEL metadata) when recalculating stream boundaries.
+        while (iter->Valid() && (identifySubkeyType(iter->key()) != StreamSubkeyType::StreamEntry ||
+                                 deleted_entry_keys.count(iter->key().ToString()) > 0)) {
           iter->Prev();
         }
         if (iter->Valid()) {

src/types/redis_stream.h

@@ -27,7 +27,6 @@
 #include <string>
 #include <vector>
 
-#include "common/db_util.h"
 #include "storage/redis_db.h"
 #include "storage/redis_metadata.h"
 

tests/gocase/unit/type/stream/stream_test.go

@@ -1032,6 +1032,58 @@ var streamTests = func(t *testing.T, configs util.KvrocksServerConfigs) {
 		require.Equal(t, "2-0", msgs[0].ID)
 		require.NoError(t, rdb.Del(ctx, streamName).Err())
 	})
+
+	t.Run("XDELEX KEEPREF delete last entry with consumer groups recalculates boundary", func(t *testing.T) {
+		streamName := "xdelex_boundary_last_" + strconv.Itoa(rand.Int())
+		require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
+			Stream: streamName, ID: "1-0", Values: []string{"f", "v"},
+		}).Err())
+		require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
+			Stream: streamName, ID: "2-0", Values: []string{"f", "v"},
+		}).Err())
+		require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
+			Stream: streamName, ID: "3-0", Values: []string{"f", "v"},
+		}).Err())
+		// Group/PEL keys sort after entry keys; SeekToLast must filter them out.
+		require.NoError(t, rdb.XGroupCreateMkStream(ctx, streamName, "g1", "0").Err())
+		r, err := rdb.Do(ctx, "XDELEX", streamName, "KEEPREF", "IDS", "1", "3-0").Result()
+		require.NoError(t, err)
+		require.Equal(t, []interface{}{int64(1)}, r)
+		require.Equal(t, int64(2), rdb.XLen(ctx, streamName).Val())
+		// XRANGE should end at 2-0 now, not parse garbage from group keys
+		msgs, err := rdb.XRange(ctx, streamName, "-", "+").Result()
+		require.NoError(t, err)
+		require.Equal(t, 2, len(msgs))
+		require.Equal(t, "2-0", msgs[1].ID)
+		require.NoError(t, rdb.Del(ctx, streamName).Err())
+	})
+
+	t.Run("XDELEX DELREF delete last entry with consumer groups recalculates boundary", func(t *testing.T) {
+		streamName := "xdelex_boundary_last_delref_" + strconv.Itoa(rand.Int())
+		require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
+			Stream: streamName, ID: "1-0", Values: []string{"f", "v"},
+		}).Err())
+		require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
+			Stream: streamName, ID: "2-0", Values: []string{"f", "v"},
+		}).Err())
+		require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
+			Stream: streamName, ID: "3-0", Values: []string{"f", "v"},
+		}).Err())
+		require.NoError(t, rdb.XGroupCreateMkStream(ctx, streamName, "g1", "0").Err())
+		_, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
+			Group: "g1", Consumer: "c1", Streams: []string{streamName, ">"}, Count: 1,
+		}).Result()
+		require.NoError(t, err)
+		r, err := rdb.Do(ctx, "XDELEX", streamName, "DELREF", "IDS", "1", "3-0").Result()
+		require.NoError(t, err)
+		require.Equal(t, []interface{}{int64(1)}, r)
+		require.Equal(t, int64(2), rdb.XLen(ctx, streamName).Val())
+		msgs, err := rdb.XRange(ctx, streamName, "-", "+").Result()
+		require.NoError(t, err)
+		require.Equal(t, 2, len(msgs))
+		require.Equal(t, "2-0", msgs[1].ID)
+		require.NoError(t, rdb.Del(ctx, streamName).Err())
+	})
 }
 
 // streamSimulateXRANGE simulates Redis XRANGE implementation in Golang.