fix(stream): correct XPENDING end parsing and inclusive PEL range scan

Author: songqing

src/commands/cmd_stream.cc

@@ -864,7 +864,7 @@ class CommandXPending : public Commander {
       }
 
       if (end_id != "+") {
-        auto s = ParseStreamEntryID(start_id, &options_.end_id);
+        auto s = ParseStreamEntryID(end_id, &options_.end_id);
         if (!s.IsOK()) {
           return s;
         }

src/types/redis_stream.cc

@@ -1758,10 +1758,26 @@ rocksdb::Status Stream::GetPendingEntries(engine::Context &ctx, StreamPendingOpt
   }
 
   std::string prefix_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, options.start_id);
-  std::string end_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, options.end_id);
+  // XPENDING extended form follows XRANGE-style ranges (closed interval by default).
+  // RocksDB iterate_upper_bound is exclusive; use a key strictly after end_id for the scan bound.
+  std::string end_key_exclusive;
+  if (options.end_id == StreamEntryID::Maximum()) {
+    // "+" means no upper ID limit; place the bound after all keys in this stream metadata version
+    // (same pattern as Stream::trim). A PEL-specific key cannot be "past" maximum ID.
+    end_key_exclusive = InternalKey(ns_key, "", metadata.version + 1, storage_->IsSlotIdEncoded()).Encode();
+  } else {
+    StreamEntryID end_next = options.end_id;
+    Status inc_st = IncrementStreamEntryID(&end_next);
+    if (!inc_st.IsOK()) {
+      end_key_exclusive = InternalKey(ns_key, "", metadata.version + 1, storage_->IsSlotIdEncoded()).Encode();
+    } else {
+      // Next ID after end_id in PEL key space — exclusive iterator upper bound (XRANGE closed end).
+      end_key_exclusive = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, end_next);
+    }
+  }
 
   rocksdb::ReadOptions read_options = ctx.DefaultScanOptions();
-  rocksdb::Slice upper_bound(end_key);
+  rocksdb::Slice upper_bound(end_key_exclusive);
   read_options.iterate_upper_bound = &upper_bound;
   rocksdb::Slice lower_bound(prefix_key);
   read_options.iterate_lower_bound = &lower_bound;

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

@@ -2480,6 +2480,48 @@ func TestStreamOffset(t *testing.T) {
 		require.ErrorContains(t, rdb.Do(ctx, "XREVRANGE", "mystream").Err(), "wrong number of arguments")
 		require.ErrorContains(t, rdb.Do(ctx, "XREVRANGE", "mystream", "+").Err(), "wrong number of arguments")
 	})
+
+	t.Run("XPENDING with specific end ID should filter correctly", func(t *testing.T) {
+		streamKey := "xpending-endid-test"
+		group := "grp"
+		consumer := "con"
+		require.NoError(t, rdb.Del(ctx, streamKey).Err())
+
+		id1, err := rdb.XAdd(ctx, &redis.XAddArgs{Stream: streamKey, Values: []string{"f", "1"}}).Result()
+		require.NoError(t, err)
+		id2, err := rdb.XAdd(ctx, &redis.XAddArgs{Stream: streamKey, Values: []string{"f", "2"}}).Result()
+		require.NoError(t, err)
+		id3, err := rdb.XAdd(ctx, &redis.XAddArgs{Stream: streamKey, Values: []string{"f", "3"}}).Result()
+		require.NoError(t, err)
+
+		require.NoError(t, rdb.XGroupCreateMkStream(ctx, streamKey, group, "0").Err())
+		// Read all entries so they become pending
+		_, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{Group: group, Consumer: consumer, Streams: []string{streamKey, ">"}, Count: 10}).Result()
+		require.NoError(t, err)
+
+		// XPENDING extended form: same ID range rules as XRANGE (see Redis docs). Use XPENDING with end_id = id1.
+		result, err := rdb.Do(ctx, "XPENDING", streamKey, group, id1, id1, "10").Result()
+		require.NoError(t, err)
+		entries, ok := result.([]interface{})
+		require.True(t, ok)
+		require.Len(t, entries, 1, "XPENDING with end_id=id1 should return only 1 entry")
+
+		// Use XPENDING with range [id1, id2] (should return 2 entries).
+		result, err = rdb.Do(ctx, "XPENDING", streamKey, group, id1, id2, "10").Result()
+		require.NoError(t, err)
+		entries, ok = result.([]interface{})
+		require.True(t, ok)
+		require.Len(t, entries, 2, "XPENDING with range [id1,id2] should return 2 entries")
+
+		// Use XPENDING with range [id1, id3] (should return all 3 entries).
+		result, err = rdb.Do(ctx, "XPENDING", streamKey, group, id1, id3, "10").Result()
+		require.NoError(t, err)
+		entries, ok = result.([]interface{})
+		require.True(t, ok)
+		require.Len(t, entries, 3, "XPENDING with range [id1,id3] should return 3 entries")
+
+		require.NoError(t, rdb.Del(ctx, streamKey).Err())
+	})
 }
 
 func parseStreamEntryID(id string) (ts int64, seqNum int64) {