fix(core): potential data race between blocking command and transaction (#2910)

Currently, the EXEC is an exclusive command and will enter the transaction mode
when running, but the blocking command's callback function doesn't have this lock.
So the callback might get trapped in the wrong transaction mode and break the write batch.

For example, the client A sends the command: `BRPOP list 10` and it will enter the blocking mode due to no data.
And the client B uses the transaction to send the command: `MULTI; LPUSH list a; EXEC`.
Then the client A's callback will be revoked after the `LPUSH` command is executed, but the EXEC is running.
So the callback's operation(removing an element from the list) will be counted in the transaction.

It closes #2900.
It closes #2766.

Author: git-hulk

src/commands/blocking_commander.h

@@ -24,6 +24,7 @@
 #include "common/lock_manager.h"
 #include "event_util.h"
 #include "server/redis_connection.h"
+#include "server/server.h"
 
 namespace redis {
 
@@ -70,6 +71,13 @@ class BlockingCommander : public Commander,
   void OnWrite(bufferevent *bev) {
     bool done{false};
     {
+      // The blocking command should not be executed when the server is in exclusive state,
+      // because it might have the data race when the server is in transaction mode and run
+      // the callback here might cause the current execution also in transaction mode.
+      //
+      // For more context, please refer to: https://github.com/apache/kvrocks/issues/2900
+      auto concurrency = conn_->GetServer()->WorkConcurrencyGuard();
+
       auto guard = GetLocks();
       done = OnBlockingWrite();
     }
@@ -125,7 +133,7 @@ class BlockingCommander : public Commander,
     UnblockKeys();
     auto bev = conn_->GetBufferEvent();
     conn_->SetCB(bev);
-    bufferevent_enable(bev, EV_READ);
+    bufferevent_enable(bev, EV_READ | EV_WRITE);
   }
 
  protected:

tests/gocase/unit/type/list/list_test.go

@@ -21,10 +21,12 @@ package list
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math/rand"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -1513,3 +1515,47 @@ func testList(t *testing.T, configs util.KvrocksServerConfigs) {
 		})
 	}
 }
+
+// TestPotentialDataRaceInBlockingCommand is to test blocking command's callback
+// shouldn't have data race with concurrent transaction behavior.
+//
+// For more information, please refer to: https://github.com/apache/kvrocks/issues/2900
+func TestPotentialDataRaceInBlockingCommand(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()) }()
+
+	listKey := "mylist"
+	rdb.Del(ctx, listKey)
+	var wg sync.WaitGroup
+	for i := 0; i < 4; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				err := rdb.BLPop(ctx, 3500*time.Millisecond, listKey).Err()
+				if errors.Is(err, redis.Nil) {
+					continue
+				} else if errors.Is(err, context.Canceled) {
+					return
+				} else {
+					require.NoError(t, err)
+				}
+			}
+		}()
+	}
+
+	for i := 0; i < 64; i++ {
+		pipe := rdb.TxPipeline()
+		pipe.LPush(ctx, listKey, "element")
+		_, err := pipe.Exec(ctx)
+		require.NoError(t, err)
+		time.Sleep(time.Millisecond * 100)
+	}
+
+	cancelFn()
+	wg.Wait()
+}