fix can restart cn when tn crash

pkg/lockservice/lock_table_allocator.go

@@ -206,8 +206,9 @@ func (l *lockTableAllocator) setRestartService(serviceID string) {
 			zap.String("serviceID", serviceID))
 		return
 	}
-	getLogger().Info("set restart lock service",
-		zap.String("serviceID", serviceID))
+	logServiceStatus("set restart lock service",
+		serviceID,
+		b.getStatus())
 	b.setStatus(pb.Status_ServiceLockWaiting)
 }
 
@@ -221,7 +222,8 @@ func (l *lockTableAllocator) remainTxnInService(serviceID string) int32 {
 	txnIDs := b.getTxnIds()
 	getLogger().Error("remain txn in restart service",
 		bytesArrayField("txnIDs", txnIDs),
-		zap.String("serviceID", serviceID))
+		zap.String("serviceID", serviceID),
+		zap.String("status", b.getStatus().String()))
 
 	c := len(txnIDs)
 	if c == 0 {
@@ -231,7 +233,8 @@ func (l *lockTableAllocator) remainTxnInService(serviceID string) int32 {
 			// -1 means can not get right remain txn in restart lock service
 			c = -1
 			getLogger().Error("can not get right remain txn in restart lock service",
-				zap.String("serviceID", serviceID))
+				zap.String("serviceID", serviceID),
+				zap.String("status", b.getStatus().String()))
 		}
 
 	}
@@ -255,6 +258,9 @@ func (l *lockTableAllocator) canRestartService(serviceID string) bool {
 			zap.String("serviceID", serviceID))
 		return true
 	}
+	logServiceStatus("can restart lock service",
+		serviceID,
+		b.getStatus())
 	return b.isStatus(pb.Status_ServiceCanRestart)
 }
 
@@ -612,9 +618,16 @@ func (b *serviceBinds) isStatus(status pb.Status) bool {
 	return b.status == status
 }
 
+func (b *serviceBinds) getStatus() pb.Status {
+	b.RLock()
+	defer b.RUnlock()
+	return b.status
+}
+
 func (b *serviceBinds) setStatus(status pb.Status) {
 	b.Lock()
 	defer b.Unlock()
+	logStatusChange(b.status, status)
 	b.status = status
 }
 
@@ -754,8 +767,14 @@ func (l *lockTableAllocator) handleKeepLockTableBind(
 	}
 	b := l.getServiceBinds(req.KeepLockTableBind.ServiceID)
 	if b.isStatus(pb.Status_ServiceLockEnable) {
-		writeResponse(ctx, cancel, resp, nil, cs)
-		return
+		if req.KeepLockTableBind.Status != pb.Status_ServiceLockEnable {
+			getLogger().Error("tn has abnormal lock service status",
+				zap.String("serviceID", b.serviceID),
+				zap.String("status", req.KeepLockTableBind.Status.String()))
+		} else {
+			writeResponse(ctx, cancel, resp, nil, cs)
+			return
+		}
 	}
 	b.setTxnIds(req.KeepLockTableBind.TxnIDs)
 	switch req.KeepLockTableBind.Status {

pkg/lockservice/lock_table_keeper.go

@@ -16,6 +16,7 @@ package lockservice
 
 import (
 	"context"
+	"go.uber.org/zap"
 	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
@@ -209,6 +210,12 @@ func (k *lockTableKeeper) doKeepLockTableBind(ctx context.Context) {
 	if resp.KeepLockTableBind.OK {
 		switch resp.KeepLockTableBind.Status {
 		case pb.Status_ServiceLockEnable:
+			if !k.service.isStatus(pb.Status_ServiceLockEnable) {
+				getLogger().Error("tn has abnormal lock service status",
+					zap.String("serviceID", k.serviceID),
+					zap.String("status", k.service.getStatus().String()))
+			}
+			return
 		case pb.Status_ServiceLockWaiting:
 			// maybe pb.Status_ServiceUnLockSucc
 			if k.service.isStatus(pb.Status_ServiceLockEnable) {

pkg/lockservice/log.go

@@ -369,10 +369,12 @@ func logPingFailed(
 	}
 }
 
-func logCanLockOnService() {
+func logCanLockOnService(
+	serviceID string) {
 	logger := getWithSkipLogger()
 	if logger.Enabled(zap.InfoLevel) {
-		logger.Error("if lock on service")
+		logger.Error("if lock on service",
+			zap.String("serviceID", serviceID))
 	}
 }
 
@@ -500,6 +502,19 @@ func logStatus(
 	}
 }
 
+func logServiceStatus(
+	info string,
+	serviceID string,
+	status pb.Status) {
+	logger := getWithSkipLogger()
+	if logger.Enabled(zap.InfoLevel) {
+		logger.Info("service status",
+			zap.String("info", info),
+			zap.String("serviceID", serviceID),
+			zap.String("status", status.String()))
+	}
+}
+
 func logStatusChange(
 	from pb.Status,
 	to pb.Status) {

pkg/lockservice/service.go

@@ -318,11 +318,11 @@ func (s *service) canLockOnServiceStatus(
 	if s.isStatus(pb.Status_ServiceLockEnable) {
 		return true
 	}
-	defer logCanLockOnService()
 	if opts.Sharding == pb.Sharding_ByRow {
 		tableID = shardingByRow(rows[0])
 	}
 	if !s.validGroupTable(opts.Group, tableID) {
+		logCanLockOnService(s.serviceID)
 		return false
 	}
 	if s.activeTxnHolder.hasActiveTxn(txnID) {
@@ -386,8 +386,8 @@ func (s *service) Close() error {
 func (s *service) setStatus(status pb.Status) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s.mu.status = status
 	logStatusChange(s.mu.status, status)
+	s.mu.status = status
 }
 
 func (s *service) getStatus() pb.Status {

pkg/lockservice/service_test.go

@@ -1430,6 +1430,193 @@ func TestReLockSuccWithLockTableBindChanged(t *testing.T) {
 	)
 }
 
+func TestIssue3537(t *testing.T) {
+	runLockServiceTestsWithLevel(
+		t,
+		zapcore.DebugLevel,
+		[]string{"s1"},
+		time.Second*1,
+		func(alloc *lockTableAllocator, s []*service) {
+			l1 := s[0]
+
+			ctx, cancel := context.WithTimeout(
+				context.Background(),
+				time.Second*10)
+			defer cancel()
+			option := pb.LockOptions{
+				Granularity: pb.Granularity_Row,
+				Mode:        pb.LockMode_Exclusive,
+				Policy:      pb.WaitPolicy_Wait,
+			}
+
+			_, err := l1.Lock(
+				ctx,
+				0,
+				[][]byte{{1}},
+				[]byte("txn1"),
+				option)
+			require.NoError(t, err)
+
+			alloc.setRestartService("s1")
+			for {
+				if l1.isStatus(pb.Status_ServiceLockWaiting) {
+					break
+				}
+				select {
+				case <-ctx.Done():
+					panic("timeout bug")
+				default:
+				}
+			}
+
+			b := alloc.getServiceBindsWithoutPrefix("s1")
+			b.setStatus(pb.Status_ServiceLockEnable)
+
+			for {
+				if b.isStatus(pb.Status_ServiceLockWaiting) {
+					break
+				}
+				select {
+				case <-ctx.Done():
+					panic("timeout bug")
+				default:
+				}
+			}
+
+			require.NoError(t, l1.Unlock(ctx, []byte("txn1"), timestamp.Timestamp{}))
+			for {
+				if b.isStatus(pb.Status_ServiceCanRestart) {
+					break
+				}
+				select {
+				case <-ctx.Done():
+					panic("timeout bug")
+				default:
+				}
+			}
+		},
+		nil,
+	)
+}
+
+func TestIssue3537_2(t *testing.T) {
+	runLockServiceTestsWithLevel(
+		t,
+		zapcore.DebugLevel,
+		[]string{"s1"},
+		time.Second*1,
+		func(alloc *lockTableAllocator, s []*service) {
+			l1 := s[0]
+
+			ctx, cancel := context.WithTimeout(
+				context.Background(),
+				time.Second*10)
+			defer cancel()
+			option := pb.LockOptions{
+				Granularity: pb.Granularity_Row,
+				Mode:        pb.LockMode_Exclusive,
+				Policy:      pb.WaitPolicy_Wait,
+			}
+
+			_, err := l1.Lock(
+				ctx,
+				0,
+				[][]byte{{1}},
+				[]byte("txn1"),
+				option)
+			require.NoError(t, err)
+
+			alloc.setRestartService("s1")
+			for {
+				if l1.isStatus(pb.Status_ServiceLockWaiting) {
+					break
+				}
+				select {
+				case <-ctx.Done():
+					panic("timeout bug")
+				default:
+				}
+			}
+
+			b := alloc.getServiceBindsWithoutPrefix("s1")
+			b.setStatus(pb.Status_ServiceLockEnable)
+
+			require.NoError(t, l1.Unlock(ctx, []byte("txn1"), timestamp.Timestamp{}))
+
+			for {
+				if b.isStatus(pb.Status_ServiceCanRestart) {
+					break
+				}
+				select {
+				case <-ctx.Done():
+					panic("timeout bug")
+				default:
+				}
+			}
+		},
+		nil,
+	)
+}
+
+func TestIssue3537_3(t *testing.T) {
+	runLockServiceTestsWithLevel(
+		t,
+		zapcore.DebugLevel,
+		[]string{"s1"},
+		time.Second*1,
+		func(alloc *lockTableAllocator, s []*service) {
+			l1 := s[0]
+
+			ctx, cancel := context.WithTimeout(
+				context.Background(),
+				time.Second*10)
+			defer cancel()
+			option := pb.LockOptions{
+				Granularity: pb.Granularity_Row,
+				Mode:        pb.LockMode_Exclusive,
+				Policy:      pb.WaitPolicy_Wait,
+			}
+
+			_, err := l1.Lock(
+				ctx,
+				0,
+				[][]byte{{1}},
+				[]byte("txn1"),
+				option)
+			require.NoError(t, err)
+
+			alloc.setRestartService("s1")
+			for {
+				if l1.isStatus(pb.Status_ServiceLockWaiting) {
+					break
+				}
+				select {
+				case <-ctx.Done():
+					panic("timeout bug")
+				default:
+				}
+			}
+
+			require.NoError(t, l1.Unlock(ctx, []byte("txn1"), timestamp.Timestamp{}))
+
+			b := alloc.getServiceBindsWithoutPrefix("s1")
+			b.setStatus(pb.Status_ServiceLockEnable)
+
+			for {
+				if b.isStatus(pb.Status_ServiceCanRestart) {
+					break
+				}
+				select {
+				case <-ctx.Done():
+					panic("timeout bug")
+				default:
+				}
+			}
+		},
+		nil,
+	)
+}
+
 func TestIssue3288(t *testing.T) {
 	runLockServiceTestsWithLevel(
 		t,