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 c278e4a06 fix(stream): potential data race between blocking
XREAD/XREADGROUP and transaction (#3685)
c278e4a06 is described below
commit c278e4a0623bbc1722c9ccfd0b00c96998cdb122
Author: TC <[email protected]>
AuthorDate: Sat Oct 17 10:30:45 2026 -0500
fix(stream): potential data race between blocking XREAD/XREADGROUP and
transaction (#3685)
This is the same issue as #2900 / #2910, for the stream commands.
`CommandXRead::OnWrite` and `CommandXReadGroup::OnWrite` replace the
connection's bufferevent callbacks, as `BlockingCommander` does, but
they don't take `WorkConcurrencyGuard()` before reading the stream. A
reader woken by an `XADD` that runs inside `MULTI/EXEC` can therefore
execute while the `EXEC` still holds the storage in transaction mode. It
then reads, or for `XREADGROUP` writes, the shared transaction write
batch from another worker. That corrupts the batch (`Corruption: unknown
WriteBatch tag` / `WriteBatch has wrong count`).
This change:
- takes the guard in both `OnWrite` paths, mirroring
`BlockingCommander::OnWrite`;
- adds `TestPotentialDataRaceInBlockingStreamRead`, modelled on
`TestPotentialDataRaceInBlockingCommand`.
Validation on `unstable` @ 07079ab:
- **The new test:** it fails 5 of 5 runs without the fix and passes 5 of
5 with it.
- **Existing suites:** `unit/type/stream`, `unit/multi` and
`unit/type/list` pass with the fix.
- **Lint:** clang-format-18, gofmt, go vet and golangci-lint are clean.
- **Deadlock:** the guard is taken before the key locks, the same order
`ExecuteCommands` uses. Inside the guarded scope there is only the Range
read and the Reply. Worker bufferevents use `BEV_OPT_DEFER_CALLBACKS`,
so nothing re-enters while the guard is held.
It closes #3684.
### AI assistance
Per the contributing guide: this change was produced with **Claude Code
(Anthropic, model Claude Opus 5.5)**.
- **Diagnosis:** we hit the corruption in our own deployment, which uses
blocked `XREADGROUP` consumers alongside `MULTI/EXEC` batches. Claude
Code read the KVRocks sources and the RocksDB logs, related them to
#2900 and #2910, and traced the unguarded `OnWrite` paths in
`cmd_stream.cc`.
- **Patch and test:** it wrote both, modelled on #2910.
- **Validation:** it built `unstable` with and without the patch on our
test machine, and ran the new test, the existing stream, multi and list
suites, and the linters.
The results above come from those runs.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-authored-by: jihuayu <[email protected]>
---
src/commands/cmd_stream.cc | 16 ++
tests/gocase/unit/type/stream/stream_test.go | 243 +++++++++++++++++++++++++++
2 files changed, 259 insertions(+)
diff --git a/src/commands/cmd_stream.cc b/src/commands/cmd_stream.cc
index f2facb982..ce8bb3fa2 100644
--- a/src/commands/cmd_stream.cc
+++ b/src/commands/cmd_stream.cc
@@ -1360,6 +1360,14 @@ class CommandXRead : public Commander,
conn_->SetCB(bev);
bufferevent_enable(bev, EV_READ);
+ // The read below runs from the wake-up callback, outside ExecuteCommands,
so it
+ // must not overlap an exclusive command: while EXEC runs, the storage is
in
+ // transaction mode and this read would walk (and, for XREADGROUP, write
into)
+ // the shared transaction write batch from another worker thread. This
mirrors
+ // BlockingCommander::OnWrite. For more context, please refer to:
+ // https://github.com/apache/kvrocks/issues/2900
+ auto concurrency = srv_->WorkConcurrencyGuard();
+
redis::Stream stream_db(srv_->storage, conn_->GetNamespace());
std::vector<StreamReadResult> results;
@@ -1675,6 +1683,14 @@ class CommandXReadGroup : public Commander,
conn_->SetCB(bev);
bufferevent_enable(bev, EV_READ);
+ // The read below runs from the wake-up callback, outside ExecuteCommands,
so it
+ // must not overlap an exclusive command: while EXEC runs, the storage is
in
+ // transaction mode and this read would walk (and, for XREADGROUP, write
into)
+ // the shared transaction write batch from another worker thread. This
mirrors
+ // BlockingCommander::OnWrite. For more context, please refer to:
+ // https://github.com/apache/kvrocks/issues/2900
+ auto concurrency = srv_->WorkConcurrencyGuard();
+
redis::Stream stream_db(srv_->storage, conn_->GetNamespace());
std::vector<StreamReadResult> results;
diff --git a/tests/gocase/unit/type/stream/stream_test.go
b/tests/gocase/unit/type/stream/stream_test.go
index 7e3b65751..3560ff066 100644
--- a/tests/gocase/unit/type/stream/stream_test.go
+++ b/tests/gocase/unit/type/stream/stream_test.go
@@ -22,11 +22,13 @@ package stream
import (
"context"
+ "errors"
"fmt"
"math/rand"
"slices"
"strconv"
"strings"
+ "sync"
"testing"
"time"
@@ -2768,3 +2770,244 @@ func parseStreamEntryID(id string) (ts int64, seqNum
int64) {
return
}
+
+func TestBlockingXReadWokenByTransaction(t *testing.T) {
+ srv := util.StartServer(t, map[string]string{})
+ defer srv.Close()
+ ctx := context.Background()
+ rdb := srv.NewClient()
+ defer func() { require.NoError(t, rdb.Close()) }()
+
+ const readers = 4
+ clients := make([]*redis.Client, readers)
+ for i := range clients {
+ clients[i] = srv.NewClientWithOption(&redis.Options{MaxRetries:
-1})
+ defer func() { require.NoError(t, clients[i].Close()) }()
+ }
+
+ for round := 0; round < 4; round++ {
+ results := make(chan *redis.XStreamSliceCmd, readers)
+ for _, client := range clients {
+ go func() {
+ results <- client.XRead(ctx, &redis.XReadArgs{
+ Streams: []string{"mystream", "$"},
+ Count: 1,
+ Block: 10 * time.Second,
+ })
+ }()
+ }
+ require.Eventually(t, func() bool {
+ return util.FindInfoEntry(rdb, "blocked_clients") ==
strconv.Itoa(readers)
+ }, 5*time.Second, 10*time.Millisecond)
+
+ pipe := rdb.TxPipeline()
+ value := strconv.Itoa(round)
+ added := pipe.XAdd(ctx, &redis.XAddArgs{Stream: "mystream",
Values: []string{"field", value}})
+ for i := 0; i < 64; i++ {
+ pipe.ZAdd(ctx, "myzset", redis.Z{Score: float64(i),
Member: fmt.Sprintf("member-%d-%d", round, i)})
+ }
+ _, err := pipe.Exec(ctx)
+ require.NoError(t, err)
+ id, err := added.Result()
+ require.NoError(t, err)
+
+ require.Eventually(t, func() bool { return len(results) ==
readers }, 5*time.Second, 10*time.Millisecond)
+ for i := 0; i < readers; i++ {
+ streams, err := (<-results).Result()
+ require.NoError(t, err)
+ require.Equal(t, []redis.XStream{{
+ Stream: "mystream",
+ Messages: []redis.XMessage{{ID: id, Values:
map[string]interface{}{"field": value}}},
+ }}, streams)
+ }
+ }
+ require.NoError(t, rdb.Set(ctx, "after", "ok", 0).Err())
+}
+
+func TestBlockingXReadGroupWokenByTransaction(t *testing.T) {
+ srv := util.StartServer(t, map[string]string{})
+ defer srv.Close()
+ ctx := context.Background()
+ rdb := srv.NewClient()
+ defer func() { require.NoError(t, rdb.Close()) }()
+ require.NoError(t, rdb.XGroupCreateMkStream(ctx, "mystream", "mygroup",
"0").Err())
+
+ const readers = 4
+ const rounds = 4
+ clients := make([]*redis.Client, readers)
+ consumerPending := make(map[string]int64, readers)
+ for i := range clients {
+ clients[i] = srv.NewClientWithOption(&redis.Options{MaxRetries:
-1})
+ defer func() { require.NoError(t, clients[i].Close()) }()
+ consumerPending[fmt.Sprintf("consumer-%d", i)] = rounds
+ }
+
+ type readResult struct {
+ consumer string
+ cmd *redis.XStreamSliceCmd
+ }
+ expected := make([]redis.XMessage, 0, readers*rounds)
+ owners := make(map[string]string, readers*rounds)
+ for round := 0; round < rounds; round++ {
+ results := make(chan readResult, readers)
+ for i, client := range clients {
+ go func() {
+ consumer := fmt.Sprintf("consumer-%d", i)
+ results <- readResult{consumer: consumer, cmd:
client.XReadGroup(ctx, &redis.XReadGroupArgs{
+ Group: "mygroup",
+ Consumer: consumer,
+ Streams: []string{"mystream", ">"},
+ Count: 1,
+ Block: 10 * time.Second,
+ })}
+ }()
+ }
+ require.Eventually(t, func() bool {
+ return util.FindInfoEntry(rdb, "blocked_clients") ==
strconv.Itoa(readers)
+ }, 5*time.Second, 10*time.Millisecond)
+
+ pipe := rdb.TxPipeline()
+ for i := 0; i < readers; i++ {
+ id := fmt.Sprintf("%d-0", round*readers+i+1)
+ value := strconv.Itoa(round*readers + i)
+ pipe.XAdd(ctx, &redis.XAddArgs{Stream: "mystream", ID:
id, Values: []string{"field", value}})
+ expected = append(expected, redis.XMessage{ID: id,
Values: map[string]interface{}{"field": value}})
+ }
+ for i := 0; i < 64; i++ {
+ pipe.ZAdd(ctx, "myzset", redis.Z{Score: float64(i),
Member: fmt.Sprintf("member-%d-%d", round, i)})
+ }
+ _, err := pipe.Exec(ctx)
+ require.NoError(t, err)
+
+ require.Eventually(t, func() bool { return len(results) ==
readers }, 5*time.Second, 10*time.Millisecond)
+ messages := make([]redis.XMessage, 0, readers)
+ for i := 0; i < readers; i++ {
+ result := <-results
+ streams, err := result.cmd.Result()
+ require.NoError(t, err)
+ require.Len(t, streams, 1)
+ require.Equal(t, "mystream", streams[0].Stream)
+ require.Len(t, streams[0].Messages, 1)
+ message := streams[0].Messages[0]
+ require.NotContains(t, owners, message.ID, "a message
must only be assigned once")
+ owners[message.ID] = result.consumer
+ messages = append(messages, message)
+ }
+ require.ElementsMatch(t, expected[round*readers:], messages)
+ }
+
+ pending, err := rdb.XPending(ctx, "mystream", "mygroup").Result()
+ require.NoError(t, err)
+ require.Equal(t, &redis.XPending{
+ Count: readers * rounds,
+ Lower: expected[0].ID,
+ Higher: expected[len(expected)-1].ID,
+ Consumers: consumerPending,
+ }, pending)
+ entries, err := rdb.XPendingExt(ctx, &redis.XPendingExtArgs{
+ Stream: "mystream",
+ Group: "mygroup",
+ Start: "-",
+ End: "+",
+ Count: readers*rounds + 1,
+ }).Result()
+ require.NoError(t, err)
+ require.Len(t, entries, len(expected))
+ for i, entry := range entries {
+ require.Equal(t, expected[i].ID, entry.ID)
+ require.Equal(t, owners[entry.ID], entry.Consumer)
+ require.EqualValues(t, 1, entry.RetryCount)
+ }
+
+ groups, err := rdb.XInfoGroups(ctx, "mystream").Result()
+ require.NoError(t, err)
+ require.Len(t, groups, 1)
+ require.Equal(t, "mygroup", groups[0].Name)
+ require.EqualValues(t, readers, groups[0].Consumers)
+ require.EqualValues(t, readers*rounds, groups[0].Pending)
+ require.Equal(t, expected[len(expected)-1].ID,
groups[0].LastDeliveredID)
+ consumers, err := rdb.XInfoConsumers(ctx, "mystream",
"mygroup").Result()
+ require.NoError(t, err)
+ require.Len(t, consumers, readers)
+ for _, consumer := range consumers {
+ require.Contains(t, consumerPending, consumer.Name)
+ require.Equal(t, consumerPending[consumer.Name],
consumer.Pending)
+ }
+ require.NoError(t, rdb.Set(ctx, "after", "ok", 0).Err())
+}
+
+// TestPotentialDataRaceInBlockingStreamRead checks that the wake-up callback
of a
+// blocked XREAD/XREADGROUP doesn't race with a concurrent transaction: the
XADD
+// inside MULTI/EXEC wakes the reader while the EXEC still holds the storage in
+// transaction mode, the same shape as TestPotentialDataRaceInBlockingCommand.
+//
+// For more information, please refer to:
https://github.com/apache/kvrocks/issues/2900
+func TestPotentialDataRaceInBlockingStreamRead(t *testing.T) {
+ srv := util.StartServer(t, map[string]string{})
+ defer srv.Close()
+
+ ctx, cancelFn := context.WithCancel(context.Background())
+ rdb := srv.NewClient()
+ defer func() { require.NoError(t, rdb.Close()) }()
+
+ streamKey := "mystream"
+ require.NoError(t, rdb.Del(ctx, streamKey).Err())
+ require.NoError(t, rdb.XGroupCreateMkStream(ctx, streamKey, "mygroup",
"$").Err())
+
+ var wg sync.WaitGroup
+ for i := 0; i < 4; i++ {
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ for {
+ err := rdb.XRead(ctx, &redis.XReadArgs{
+ Streams: []string{streamKey, "$"},
+ Count: 10,
+ Block: 3500 * time.Millisecond,
+ }).Err()
+ if errors.Is(err, redis.Nil) {
+ continue
+ } else if errors.Is(err, context.Canceled) {
+ return
+ } else {
+ require.NoError(t, err)
+ }
+ }
+ }()
+ go func(consumer string) {
+ defer wg.Done()
+ for {
+ err := rdb.XReadGroup(ctx,
&redis.XReadGroupArgs{
+ Group: "mygroup",
+ Consumer: consumer,
+ Streams: []string{streamKey, ">"},
+ Count: 10,
+ Block: 3500 * time.Millisecond,
+ }).Err()
+ if errors.Is(err, redis.Nil) {
+ continue
+ } else if errors.Is(err, context.Canceled) {
+ return
+ } else {
+ require.NoError(t, err)
+ }
+ }
+ }(fmt.Sprintf("consumer-%d", i))
+ }
+
+ for i := 0; i < 64; i++ {
+ pipe := rdb.TxPipeline()
+ pipe.XAdd(ctx, &redis.XAddArgs{Stream: streamKey, Values:
[]string{"field", "value"}})
+ for j := 0; j < 64; j++ {
+ pipe.ZAdd(ctx, "myzset", redis.Z{Score: float64(j),
Member: fmt.Sprintf("member-%d-%d", i, j)})
+ }
+ _, err := pipe.Exec(ctx)
+ require.NoError(t, err)
+ time.Sleep(time.Millisecond * 100)
+ }
+ // The server must still accept writes after the transactions.
+ require.NoError(t, rdb.Set(ctx, "after", "ok", 0).Err())
+
+ cancelFn()
+ wg.Wait()
+}
