This is an automated email from the ASF dual-hosted git repository.
jihuayu pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new 3c1ac1817 fix(stream): decrement pending_number when XAUTOCLAIM
removes deleted PEL entries (#3578)
3c1ac1817 is described below
commit 3c1ac18172777cef83bbd6d287c1d28fd4231fba
Author: shitao <[email protected]>
AuthorDate: Tue Aug 11 00:52:47 2026 -0700
fix(stream): decrement pending_number when XAUTOCLAIM removes deleted PEL
entries (#3578)
The group's and the owning consumer's `pending_number` can be serialized
as a RESP integer larger than `INT64_MAX`, which breaks any client that
decodes integer replies as signed 64-bit — the reply fails to parse and
the connection is dropped. Hit in production.
## XAUTOCLAIM `pending_number` drift
When XAUTOCLAIM meets a PEL entry whose stream entry was trimmed or
XDEL'd, it deletes the PEL entry but never decrements the group's or the
owning consumer's `pending_number`. The counts drift upward on every
such sweep and, being unsigned, eventually underflow-wrap on a later
decrement — after which `XINFO GROUPS` and `XPENDING` report garbage.
The fix mirrors `DeletePelEntries` (XACK): when a deleted entry is
removed from the PEL, decrement the owning consumer and the group. A
claim only moves ownership within the group (net-zero for the group
count), so only deleted entries reduce it. Subtractions are saturating
so a counter that has already drifted can't wrap.
Test: read entries into a consumer, XDEL them, XAUTOCLAIM, then check
the group and consumer pending counts return to 0. The existing XDEL
autoclaim tests only asserted the returned deleted-id list.
> This PR previously also carried an `XINFO GROUPS` lag-underflow fix.
Per review it is a separate issue (SRP + clean cherry-pick), so I've
dropped it here and will open a follow-up PR.
---
AI assistance: diagnosis and drafting were done with AI help; I've
reviewed the changes and tests and understand the behavior.
---------
Signed-off-by: Shitao Weng <[email protected]>
Co-authored-by: Edward Xu <[email protected]>
---
src/types/redis_stream.cc | 43 ++++++++++++++++++++++++++--
tests/gocase/unit/type/stream/stream_test.go | 36 +++++++++++++++++++++++
2 files changed, 76 insertions(+), 3 deletions(-)
diff --git a/src/types/redis_stream.cc b/src/types/redis_stream.cc
index 580629c9f..78ec5cc02 100644
--- a/src/types/redis_stream.cc
+++ b/src/types/redis_stream.cc
@@ -582,6 +582,7 @@ rocksdb::Status Stream::AutoClaim(engine::Context &ctx,
const Slice &stream_name
auto iter = util::UniqueIterator(ctx, read_options, stream_cf_handle_);
uint64_t total_claimed_count = 0;
+ std::map<std::string, uint64_t> deleted_consumer_count;
for (iter->SeekToFirst(); iter->Valid() && count > 0 && attempts > 0;
iter->Next()) {
std::string tmp_group_name;
StreamEntryID entry_id = groupAndEntryIdFromPelInternalKey(iter->key(),
tmp_group_name);
@@ -605,6 +606,9 @@ rocksdb::Status Stream::AutoClaim(engine::Context &ctx,
const Slice &stream_name
deleted_entries.push_back(entry_id);
s = batch->Delete(stream_cf_handle_, iter->key());
if (!s.ok()) return s;
+ // The referenced entry was trimmed/XDEL'd; dropping this PEL record
must
+ // decrement pending_number below, otherwise the counter drifts up and
wraps.
+ deleted_consumer_count[penl_entry.consumer_name] += 1;
--count;
continue;
}
@@ -636,14 +640,34 @@ rocksdb::Status Stream::AutoClaim(engine::Context &ctx,
const Slice &stream_name
}
}
- if (total_claimed_count > 0 && !pending_entries.empty()) {
+ // A claim keeps the group total (entry moves between consumers); a deleted
dangling
+ // entry leaves the PEL, so decrement both its consumer and the group.
Saturating.
+ const uint64_t deleted_count = deleted_entries.size();
+ // The current consumer never appears in claimed_consumer_entity_count (we
only claim from other
+ // consumers), so keep its own deleted-entry decrement in a scalar and let
the map hold only the
+ // other consumers, avoiding a find-and-erase on the map.
+ uint64_t current_consumer_decrement = 0;
+ std::map<std::string, uint64_t> consumer_pending_decrements =
claimed_consumer_entity_count;
+ for (const auto &[consumer, cnt] : deleted_consumer_count) {
+ if (consumer == consumer_name) {
+ current_consumer_decrement += cnt;
+ } else {
+ consumer_pending_decrements[consumer] += cnt;
+ }
+ }
+
+ if (total_claimed_count > 0 || deleted_count > 0) {
current_consumer_metadata.pending_number += total_claimed_count;
+ current_consumer_metadata.pending_number =
+ current_consumer_metadata.pending_number >= current_consumer_decrement
+ ? current_consumer_metadata.pending_number -
current_consumer_decrement
+ : 0;
current_consumer_metadata.last_attempted_interaction_ms = now_ms;
s = batch->Put(stream_cf_handle_, consumer_key,
encodeStreamConsumerMetadataValue(current_consumer_metadata));
if (!s.ok()) return s;
- for (const auto &[consumer, count] : claimed_consumer_entity_count) {
+ for (const auto &[consumer, dec] : consumer_pending_decrements) {
std::string tmp_consumer_key = internalKeyFromConsumerName(ns_key,
metadata, group_name, consumer);
std::string tmp_consumer_value;
s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_,
tmp_consumer_key, &tmp_consumer_value);
@@ -651,10 +675,23 @@ rocksdb::Status Stream::AutoClaim(engine::Context &ctx,
const Slice &stream_name
return s;
}
StreamConsumerMetadata tmp_consumer_metadata =
decodeStreamConsumerMetadataValue(tmp_consumer_value);
- tmp_consumer_metadata.pending_number -= count;
+ tmp_consumer_metadata.pending_number =
+ tmp_consumer_metadata.pending_number >= dec ?
tmp_consumer_metadata.pending_number - dec : 0;
s = batch->Put(stream_cf_handle_, tmp_consumer_key,
encodeStreamConsumerMetadataValue(tmp_consumer_metadata));
if (!s.ok()) return s;
}
+
+ if (deleted_count > 0) {
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata,
group_name);
+ std::string get_group_value;
+ s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_,
group_key, &get_group_value);
+ if (!s.ok()) return s;
+ StreamConsumerGroupMetadata group_metadata =
decodeStreamConsumerGroupMetadataValue(get_group_value);
+ group_metadata.pending_number =
+ group_metadata.pending_number >= deleted_count ?
group_metadata.pending_number - deleted_count : 0;
+ s = batch->Put(stream_cf_handle_, group_key,
encodeStreamConsumerGroupMetadataValue(group_metadata));
+ if (!s.ok()) return s;
+ }
}
bool has_next_entry = false;
diff --git a/tests/gocase/unit/type/stream/stream_test.go
b/tests/gocase/unit/type/stream/stream_test.go
index 00354848c..1be57ede2 100644
--- a/tests/gocase/unit/type/stream/stream_test.go
+++ b/tests/gocase/unit/type/stream/stream_test.go
@@ -2155,6 +2155,42 @@ func TestStreamOffset(t *testing.T) {
// add xpending to this test case when it is supported
})
+ t.Run("XAUTOCLAIM decrements pending_number when sweeping deleted
entries", func(t *testing.T) {
+ streamName := "x"
+ groupName := "grp"
+ require.NoError(t, rdb.Del(ctx, streamName).Err())
+ for _, id := range []string{"1-0", "2-0", "3-0"} {
+ require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
+ Stream: streamName,
+ ID: id,
+ Values: map[string]interface{}{"f": "v"},
+ }).Err())
+ }
+ require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName,
"0").Err())
+
+ // Alice reads all three, so the group and the consumer each
hold 3 pending.
+ require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
+ Group: groupName,
+ Consumer: "Alice",
+ Count: 10,
+ Streams: []string{streamName, ">"},
+ }).Err())
+ require.Equal(t, int64(3), rdb.XInfoGroups(ctx,
streamName).Val()[0].Pending)
+ consumers := rdb.XInfoConsumers(ctx, streamName,
groupName).Val()
+ require.Len(t, consumers, 1)
+ require.Equal(t, int64(3), consumers[0].Pending)
+
+ // Delete the entries so their PEL records dangle, then let
XAUTOCLAIM sweep them.
+ // Before the fix the sweep dropped the records without
decrementing pending_number.
+ require.NoError(t, rdb.XDel(ctx, streamName, "1-0", "2-0",
"3-0").Err())
+ require.NoError(t, rdb.Do(ctx, "XAUTOCLAIM", streamName,
groupName, "Bob", 0, "0-0").Err())
+
+ require.Equal(t, int64(0), rdb.XInfoGroups(ctx,
streamName).Val()[0].Pending)
+ for _, c := range rdb.XInfoConsumers(ctx, streamName,
groupName).Val() {
+ require.Equalf(t, int64(0), c.Pending, "consumer %s
should have 0 pending", c.Name)
+ }
+ })
+
t.Run("XAUTOCLAIM with out of range count", func(t *testing.T) {
err := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{
Stream: "x",
