Skip to content

fix(stream): keep pending counters consistent #3596

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wengsht wants to merge 1 commit into
base: unstable
Choose a base branch
from
+133 −25
Draft

fix(stream): keep pending counters consistent #3596

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 41 additions & 25 deletions src/types/redis_stream.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
#include <rocksdb/status.h>

#include <memory>
#include <set>
#include <utility>
#include <vector>

Expand Down Expand Up @@ -352,7 +353,11 @@ rocksdb::Status Stream::DeletePelEntries(engine::Context &ctx, const Slice &stre
if (!s.ok()) return s;

std::map<std::string, uint64_t> consumer_acknowledges;
std::set<StreamEntryID> seen;
for (const auto &id : entry_ids) {
if (!seen.insert(id).second) {
continue;
}
std::string entry_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata, group_name, id);
std::string value;
s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_, entry_key, &value);
Expand All @@ -371,7 +376,8 @@ rocksdb::Status Stream::DeletePelEntries(engine::Context &ctx, const Slice &stre
}
if (*acknowledged > 0) {
StreamConsumerGroupMetadata group_metadata = decodeStreamConsumerGroupMetadataValue(get_group_value);
group_metadata.pending_number -= *acknowledged;
group_metadata.pending_number =
group_metadata.pending_number >= *acknowledged ? group_metadata.pending_number - *acknowledged : 0;
std::string group_value = encodeStreamConsumerGroupMetadataValue(group_metadata);
s = batch->Put(stream_cf_handle_, group_key, group_value);
if (!s.ok()) return s;
Expand All @@ -385,7 +391,8 @@ rocksdb::Status Stream::DeletePelEntries(engine::Context &ctx, const Slice &stre
}
if (s.ok()) {
auto consumer_metadata = decodeStreamConsumerMetadataValue(consumer_meta_original);
consumer_metadata.pending_number -= ack_count;
consumer_metadata.pending_number =
consumer_metadata.pending_number >= ack_count ? consumer_metadata.pending_number - ack_count : 0;
s = batch->Put(stream_cf_handle_, consumer_meta_key, encodeStreamConsumerMetadataValue(consumer_metadata));
if (!s.ok()) return s;
}
Expand Down Expand Up @@ -437,7 +444,12 @@ rocksdb::Status Stream::ClaimPelEntries(engine::Context &ctx, const Slice &strea
s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;

std::map<std::string, uint64_t> original_consumer_decrements;
std::set<StreamEntryID> seen;
for (const auto &id : entry_ids) {
if (!seen.insert(id).second) {
continue;
Comment on lines 449 to +451
}
std::string raw_value;
rocksdb::Status s = getEntryRawValue(ctx, ns_key, metadata, id, &raw_value);
if (!s.ok() && !s.IsNotFound()) {
Expand Down Expand Up @@ -473,25 +485,14 @@ rocksdb::Status Stream::ClaimPelEntries(engine::Context &ctx, const Slice &strea
result->entries.emplace_back(id.ToString(), std::move(values));
}

if (pel_entry.consumer_name != "") {
std::string original_consumer_key =
internalKeyFromConsumerName(ns_key, metadata, group_name, pel_entry.consumer_name);
std::string get_original_consumer_value;
s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_, original_consumer_key,
&get_original_consumer_value);
if (!s.ok()) {
return s;
}
StreamConsumerMetadata original_consumer_metadata =
decodeStreamConsumerMetadataValue(get_original_consumer_value);
original_consumer_metadata.pending_number -= 1;
s = batch->Put(stream_cf_handle_, original_consumer_key,
encodeStreamConsumerMetadataValue(original_consumer_metadata));
if (!s.ok()) return s;
if (!pel_entry.consumer_name.empty() && pel_entry.consumer_name != consumer_name) {
original_consumer_decrements[pel_entry.consumer_name] += 1;
consumer_metadata.pending_number += 1;
Comment on lines +488 to +490
} else if (pel_entry.consumer_name.empty()) {
consumer_metadata.pending_number += 1;
}

pel_entry.consumer_name = consumer_name;
consumer_metadata.pending_number += 1;
if (options.with_time) {
pel_entry.last_delivery_time_ms = options.last_delivery_time_ms;
} else {
Expand All @@ -514,6 +515,22 @@ rocksdb::Status Stream::ClaimPelEntries(engine::Context &ctx, const Slice &strea
}
}

for (const auto &[original_consumer, decrement] : original_consumer_decrements) {
std::string original_consumer_key = internalKeyFromConsumerName(ns_key, metadata, group_name, original_consumer);
std::string original_consumer_value;
s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_, original_consumer_key, &original_consumer_value);
if (!s.ok()) {
return s;
}
StreamConsumerMetadata original_consumer_metadata = decodeStreamConsumerMetadataValue(original_consumer_value);
original_consumer_metadata.pending_number = original_consumer_metadata.pending_number >= decrement
? original_consumer_metadata.pending_number - decrement
: 0;
s = batch->Put(stream_cf_handle_, original_consumer_key,
encodeStreamConsumerMetadataValue(original_consumer_metadata));
if (!s.ok()) return s;
}

if (options.with_last_id && options.last_delivered_id > group_metadata.last_delivered_id) {
group_metadata.last_delivered_id = options.last_delivered_id;
}
Expand Down Expand Up @@ -630,14 +647,13 @@ rocksdb::Status Stream::AutoClaim(engine::Context &ctx, const Slice &stream_name
++total_claimed_count;
claimed_consumer_entity_count[penl_entry.consumer_name] += 1;
penl_entry.consumer_name = consumer_name;
penl_entry.last_delivery_time_ms = now_ms;
// Increment the delivery attempts counter unless JUSTID option provided
if (!options.just_id) {
penl_entry.last_delivery_count += 1;
}
s = batch->Put(stream_cf_handle_, iter->key(), encodeStreamPelEntryValue(penl_entry));
if (!s.ok()) return s;
}
penl_entry.last_delivery_time_ms = now_ms;
if (!options.just_id) {
penl_entry.last_delivery_count += 1;
}
s = batch->Put(stream_cf_handle_, iter->key(), encodeStreamPelEntryValue(penl_entry));
if (!s.ok()) return s;
}

// A claim keeps the group total (entry moves between consumers); a deleted dangling
Expand Down
92 changes: 92 additions & 0 deletions tests/gocase/unit/type/stream/stream_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1658,6 +1658,32 @@ func TestStreamOffset(t *testing.T) {
require.Equal(t, int64(3), r)
})

t.Run("XACK acknowledges duplicate IDs once", func(t *testing.T) {
streamName := "xack-duplicates"
groupName := "group"
consumerName := "consumer"
require.NoError(t, rdb.Del(ctx, streamName).Err())
require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
Stream: streamName,
ID: "1-0",
Values: []string{"field", "value"},
}).Err())
require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err())
require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
Group: groupName,
Consumer: consumerName,
Streams: []string{streamName, ">"},
Count: 1,
}).Err())

acked, err := rdb.XAck(ctx, streamName, groupName, "1-0", "1-0").Result()
require.NoError(t, err)
require.Equal(t, int64(1), acked)
require.Equal(t, int64(0), rdb.XInfoGroups(ctx, streamName).Val()[0].Pending)
require.Equal(t, int64(0), rdb.XInfoConsumers(ctx, streamName, groupName).Val()[0].Pending)
require.Equal(t, int64(0), rdb.XPending(ctx, streamName, groupName).Val().Count)
})

t.Run("Simple XCLAIM command tests", func(t *testing.T) {
streamName := "mystream"
groupName := "mygroup"
Expand Down Expand Up @@ -1840,6 +1866,72 @@ func TestStreamOffset(t *testing.T) {
require.Equal(t, "1-0", claimedIDs[0], "Expected claimed message ID to match")
})

t.Run("XCLAIM to the current consumer keeps pending counts unchanged", func(t *testing.T) {
streamName := "xclaim-current-consumer"
groupName := "group"
consumerName := "consumer"
require.NoError(t, rdb.Del(ctx, streamName).Err())
require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
Stream: streamName,
ID: "1-0",
Values: []string{"field", "value"},
}).Err())
require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err())
require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
Group: groupName,
Consumer: consumerName,
Streams: []string{streamName, ">"},
Count: 1,
}).Err())

require.NoError(t, rdb.XClaimJustID(ctx, &redis.XClaimArgs{
Stream: streamName,
Group: groupName,
Consumer: consumerName,
MinIdle: 0,
Messages: []string{"1-0"},
}).Err())
require.Equal(t, int64(1), rdb.XInfoGroups(ctx, streamName).Val()[0].Pending)
require.Equal(t, int64(1), rdb.XInfoConsumers(ctx, streamName, groupName).Val()[0].Pending)
})

t.Run("XAUTOCLAIM refreshes idle time for the current consumer", func(t *testing.T) {
streamName := "xautoclaim-current-consumer"
groupName := "group"
consumerName := "consumer"
require.NoError(t, rdb.Del(ctx, streamName).Err())
require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
Stream: streamName,
ID: "1-0",
Values: []string{"field", "value"},
}).Err())
require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err())
require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
Group: groupName,
Consumer: consumerName,
Streams: []string{streamName, ">"},
Count: 1,
}).Err())
require.NoError(t, rdb.Do(ctx, "XCLAIM", streamName, groupName, consumerName, 0, "1-0",
"IDLE", 100000, "JUSTID").Err())

first := rdb.XAutoClaimJustID(ctx, &redis.XAutoClaimArgs{
Stream: streamName, Group: groupName, Consumer: consumerName,
MinIdle: 50 * time.Second, Start: "0-0", Count: 1,
})
require.NoError(t, first.Err())
firstIDs, _ := first.Val()
require.Equal(t, []string{"1-0"}, firstIDs)

second := rdb.XAutoClaimJustID(ctx, &redis.XAutoClaimArgs{
Stream: streamName, Group: groupName, Consumer: consumerName,
MinIdle: 50 * time.Second, Start: "0-0", Count: 1,
})
require.NoError(t, second.Err())
secondIDs, _ := second.Val()
require.Empty(t, secondIDs)
})

t.Run("XAUTOCLAIM can claim PEL items from another consume", func(t *testing.T) {

streamName := "mystream"
Expand Down