fix dead lock (#21279)

fix dead lock

Approved by: @ouyuanning

Author: zhangxu19830126

pkg/txn/client/client.go

@@ -743,6 +743,9 @@ func (client *txnClient) removeFromWaitActiveLocked(txnID []byte) bool {
 
 func (client *txnClient) waitMarkAllActiveAbortedLocked() {
 	if client.mu.waitMarkAllActiveAbortedC != nil {
-		<-client.mu.waitMarkAllActiveAbortedC
+		c := client.mu.waitMarkAllActiveAbortedC
+		client.mu.Unlock()
+		<-c
+		client.mu.Lock()
 	}
 }

pkg/txn/client/client_test.go

@@ -275,3 +275,16 @@ func TestNewWithUpdateSnapshotTimeout(t *testing.T) {
 	assert.Equal(t, 0, len(v.mu.waitActiveTxns))
 	v.mu.Unlock()
 }
+
+func TestWaitAbortMarked(t *testing.T) {
+	c := make(chan struct{})
+	tc := &txnClient{}
+	tc.mu.waitMarkAllActiveAbortedC = c
+	tc.mu.state = normal
+	tc.mu.activeTxns = map[string]*txnOperator{}
+	go func() {
+		close(c)
+	}()
+	op := &txnOperator{}
+	require.NoError(t, tc.openTxn(op))
+}

pkg/txn/client/operator.go

@@ -24,8 +24,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"go.uber.org/zap"
-
 	"github.com/matrixorigin/matrixone/pkg/common/log"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
@@ -38,6 +36,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/txn/rpc"
 	"github.com/matrixorigin/matrixone/pkg/txn/util"
 	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
+	"go.uber.org/zap"
 )
 
 var (