This is an automated email from the ASF dual-hosted git repository.
aleksraiden 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 4d364df9c fix: XDEL double-counting repeated IDs and stream length
underflow (#3701)
4d364df9c is described below
commit 4d364df9c59afd6e71b6adaf8973f3905b48fd2f
Author: GAURAV NAGALAPURAM VENKATARAMANAN <[email protected]>
AuthorDate: Wed Oct 21 10:34:47 2026 -0700
fix: XDEL double-counting repeated IDs and stream length underflow (#3701)
Stream::DeleteEntries counted a repeated ID once per occurrence because
the read did not see deletes staged in the same batch, desyncing
metadata.size and underflowing it to 2^64-1. Dedupe IDs per call and
clamp the size subtraction.
Fixes #3700
AI-use disclosure
An AI assistant (Muse) was used during this contribution. It helped
with:
- Verifying the root cause in Stream::DeleteEntries against the reported
issue
- Suggesting the fix approach (deduplicating IDs within a call, clamping
the length subtraction against underflow)
- Reviewing the implementation
- Drafting the Go regression test
The C++ implementation was written by me, and I verified the logic
against the issue's reproduction cases.
Parts needing extra review: the regression test has not been executed
locally yet — my macOS build is blocked by an unrelated range-v3/libc
toolchain incompatibility in the repo's pinned dependencies, so the test
relies on CI validation once the workflows are approved.
---------
Co-authored-by: Aleks Lozovyuk <[email protected]>
---
src/types/redis_stream.cc | 8 +++++++-
tests/gocase/unit/type/stream/stream_test.go | 14 ++++++++++++++
2 files changed, 21 insertions(+), 1 deletion(-)
diff --git a/src/types/redis_stream.cc b/src/types/redis_stream.cc
index 66e05bbda..f506d9b00 100644
--- a/src/types/redis_stream.cc
+++ b/src/types/redis_stream.cc
@@ -23,6 +23,7 @@
#include <rocksdb/status.h>
#include <memory>
+#include <set>
#include <utility>
#include <vector>
@@ -1017,7 +1018,12 @@ rocksdb::Status Stream::DeleteEntries(engine::Context
&ctx, const Slice &stream_
auto iter = util::UniqueIterator(ctx, read_options, stream_cf_handle_);
+ std::set<StreamEntryID> seen_ids;
for (const auto &id : ids) {
+ if (!seen_ids.insert(id).second) {
+ continue;
+ }
+
std::string entry_key = internalKeyFromEntryID(ns_key, metadata, id);
std::string value;
s = storage_->Get(ctx, read_options, stream_cf_handle_, entry_key, &value);
@@ -1062,7 +1068,7 @@ rocksdb::Status Stream::DeleteEntries(engine::Context
&ctx, const Slice &stream_
}
if (*deleted_cnt > 0) {
- metadata.size -= *deleted_cnt;
+ metadata.size = *deleted_cnt > metadata.size ? 0 : metadata.size -
*deleted_cnt;
std::string bytes;
metadata.Encode(&bytes);
diff --git a/tests/gocase/unit/type/stream/stream_test.go
b/tests/gocase/unit/type/stream/stream_test.go
index 3560ff066..0ed58f171 100644
--- a/tests/gocase/unit/type/stream/stream_test.go
+++ b/tests/gocase/unit/type/stream/stream_test.go
@@ -504,6 +504,20 @@ var streamTests = func(t *testing.T, configs
util.KvrocksServerConfigs) {
require.Subset(t, items[0].Values,
map[string]interface{}{"foo": "value0"})
require.Subset(t, items[1].Values,
map[string]interface{}{"foo": "value2"})
})
+ t.Run("XDEL with repeated IDs counts each entry once", func(t
*testing.T) {
+ require.NoError(t, rdb.Del(ctx, "somestream").Err())
+ require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{Stream:
"somestream", ID: "1-0", Values: []string{"foo", "value0"}}).Err())
+ require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{Stream:
"somestream", ID: "2-0", Values: []string{"foo", "value1"}}).Err())
+ // A repeated ID must be counted once; the stream length must
stay consistent.
+ require.EqualValues(t, 1, rdb.XDel(ctx, "somestream", "1-0",
"1-0").Val())
+ require.EqualValues(t, 1, rdb.XLen(ctx, "somestream").Val())
+ items := rdb.XRange(ctx, "somestream", "-", "+").Val()
+ require.Len(t, items, 1)
+ require.Equal(t, "2-0", items[0].ID)
+ // Deleting the last entry with repeats must not underflow the
length.
+ require.EqualValues(t, 1, rdb.XDel(ctx, "somestream", "2-0",
"2-0", "2-0").Val())
+ require.EqualValues(t, 0, rdb.XLen(ctx, "somestream").Val())
+ })
// Here the idea is to check the consistency of the stream data
structure as we remove all the elements down to zero elements.
t.Run("XDEL fuzz test", func(t *testing.T) {
