Repository: matrixorigin/matrixone (go)

Issue:
[Bug]: DN report shard not found after Pod re-creation
### Is there an existing issue for the same bug?

- [X] I have checked the existing issues.

### Environment

```Markdown
- Version or commit-id (e.g. v0.1.0 or 8b23a93): b5cfd63a
- Hardware parameters:
- OS type:
- Others:
```


### Actual Behavior

After I recreate DN and CN Pod on k8s with new persistent volume (the formerly persisted data will lost), DN report shard not found and CN cannot connect to DN:

DN log:

```
{"level":"ERROR","time":"2023/03/22 03:27:59.007169 +0000","caller":"dnservice/store_rpc_handler.go:166","msg":"error: dn shard uuid 33376565-3237-3162-3439-323631326537, id 2 not found","span":{"trace_id":"e4975cea-c860-11ed-adf8-cacea6ac1916","kind":"remote"},"stacktrace":"github.com/matrixorigin/matrixone/pkg/dnservice.(*store).validDNShard\n\t/go/src/github.com/matrixorigin/matrixone/pkg/dnservice/store_rpc_handler.go:166\ngithub.com/matrixorigin/matrixone/pkg/dnservice.(*store).doRead\n\t/go/src/github.com/matrixorigin/matrixone/pkg/dnservice/store_rpc_handler.go:70\ngithub.com/matrixorigin/matrixone/pkg/dnservice.(*store).handleWithRetry\n\t/go/src/github.com/matrixorigin/matrixone/pkg/dnservice/store_rpc_handler.go:184\ngithub.com/matrixorigin/matrixone/pkg/dnservice.(*store).handleRead\n\t/go/src/github.com/matrixorigin/matrixone/pkg/dnservice/store_rpc_handler.go:58\ngithub.com/matrixorigin/matrixone/pkg/txn/rpc.(*server).onMessage\n\t/go/src/github.com/matrixorigin/matrixone/pkg/txn/rpc/server.go:163\ngithub.com/matrixorigin/matrixone/pkg/common/morpc.(*server).onMessage\n\t/go/src/github.com/matrixorigin/matrixone/pkg/common/morpc/server.go:227\ngithub.com/fagongzi/goetty/v2.(*server).doConnection\n\t/go/pkg/mod/github.com/fagongzi/goetty/v2@v2.0.3-0.20221212132037-abf2d4c05484/application.go:381\ngithub.com/fagongzi/goetty/v2.(*server).doStart.func2.1\n\t/go/pkg/mod/github.com/fagongzi/goetty/v2@v2.0.3-0.20221212132037-abf2d4c05484/application.go:344"}
{"level":"INFO","time":"2023/03/22 03:27:59.890790 +0000","caller":"blockio/writer.go:120","msg":"[WriteEnd]","operation":"name: sys/logs/2023/03/22/rawlog/1679455678_33376565-3237-3162-3439-323631326537_DN_985619455.tae, block count: 1, size: 5323"}
{"level":"INFO","time":"2023/03/22 03:27:59.929997 +0000","caller":"blockio/writer.go:120","msg":"[WriteEnd]","operation":"name: sys/logs/2023/03/22/metric/1679455679_33376565-3237-3162-3439-323631326537_DN_005677618.tae, block count: 1, size: 742"}
{"level":"ERROR","time":"2023/03/22 03:28:00.007319 +0000","caller":"dnservice/store_rpc_handler.go:166","msg":"error: dn shard uuid 33376565-3237-3162-3439-323631326537, id 2 not found","span":{"trace_id":"e4975cea-c860-11ed-adf8-cacea6ac1916","kind":"remote"},"stacktrace":"github.com/matrixorigin/matrixone/pkg/dnservice.(*store).validDNShard\n\t/go/src/github.com/matrixorigin/matrixone/pkg/dnservice/store_rpc_handler.go:166\ngithub.com/matrixorigin/matrixone/pkg/dnservice.(*store).doRead\n\t/go/src/github.com/matrixorigin/matrixone/pkg/dnservice/store_rpc_handler.go:70\ngithub.com/matrixorigin/matrixone/pkg/dnservice.(*store).handleWithRetry\n\t/go/src/github.com/matrixorigin/matrixone/pkg/dnservice/store_rpc_handler.go:184\ngithub.com/matrixorigin/matrixone/pkg/dnservice.(*store).handleRead\n\t/go/src/github.com/matrixorigin/matrixone/pkg/dnservice/store_rpc_handler.go:58\ngithub.com/matrixorigin/matrixone/pkg/txn/rpc.(*server).onMessage\n\t/go/src/github.com/matrixorigin/matrixone/pkg/txn/rpc/server.go:163\ngithub.com/matrixorigin/matrixone/pkg/common/morpc.(*server).onMessage\n\t/go/src/github.com/matrixorigin/matrixone/pkg/common/morpc/server.go:227\ngithub.com/fagongzi/goetty/v2.(*server).doConnection\n\t/go/pkg/mod/github.com/fagongzi/goetty/v2@v2.0.3-0.20221212132037-abf2d4c05484/application.go:381\ngithub.com/fagongzi/goetty/v2.(*server).doStart.func2.1\n\t/go/pkg/mod/github.com/fagongzi/goetty/v2@v2.0.3-0.20221212132037-abf2d4c05484/application.go:344"}
```

CN log:

```
{"level":"INFO","time":"2023/03/22 03:33:50.810931 +0000","caller":"mo-service/launch.go:191","msg":"shard not ready"}
{"level":"INFO","time":"2023/03/22 03:33:51.812921 +0000","caller":"mo-service/launch.go:191","msg":"shard not ready"}
{"level":"INFO","time":"2023/03/22 03:33:52.815108 +0000","caller":"mo-service/launch.go:191","msg":"shard not ready"}
{"level":"INFO","time":"2023/03/22 03:33:53.817263 +0000","caller":"mo-service/launch.go:191","msg":"shard not ready"}
{"level":"INFO","time":"2023/03/22 03:33:54.819301 +0000","caller":"mo-service/launch.go:191","msg":"shard not ready"}
{"level":"INFO","time":"2023/03/22 03:33:55.821372 +0000","caller":"mo-service/launch.go:191","msg":"shard not ready"}
{"level":"INFO","time":"2023/03/22 03:33:56.823435 +0000","caller":"mo-service/launch.go:191","msg":"shard not ready"}
{"level":"INFO","time":"2023/03/22 03:33:57.825407 +0000","caller":"mo-service/launch.go:191","msg":"shard not ready"}
```

full log:
[cn.log](https://github.com/matrixorigin/matrixone/files/11035969/cn.log)

[dn.log](https://github.com/matrixorigin/matrixone/files/11036003/dn.log)



### Expected Behavior

DN should restart and connect to HAKeeper without error.

### Steps to Reproduce

_No response_

### Additional information

_No response_

Pull request description:
fix #8599: move task schedule to a separate goroutine
## What type of PR is this?

- [ ] API-change
- [*] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #8599 

## What this PR does / why we need it:
moving the task schedule from the ticker normal routine to a separate goroutine can avoid the hakeeper's health check and tick update operations being blocked by the task schedule, or the tick will be skipped and can not correctly estimate the time passing.

and I have already added some unit tests.



Patch:
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index fa225b445..2654ca64a 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -667,6 +667,34 @@ func (l *store) queryLog(ctx context.Context, shardID uint64,
 	}
 }
 
+func (l *store) tickerForTaskSchedule(ctx context.Context, duration time.Duration) {
+	ticker := time.NewTicker(duration)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			state, _ := l.getCheckerStateFromLeader()
+			if state != nil && state.State == pb.HAKeeperRunning {
+				l.taskSchedule(state)
+			}
+
+		case <-ctx.Done():
+			return
+		}
+
+		// l.taskSchedule could be blocking a long time, this extra select
+		// can give a chance immediately to check the ctx status when it resumes.
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			// nothing to do
+		}
+	}
+
+}
+
 func (l *store) ticker(ctx context.Context) {
 	if l.cfg.HAKeeperTickInterval.Duration == 0 {
 		panic("invalid HAKeeperTickInterval")
@@ -685,6 +713,12 @@ func (l *store) ticker(ctx context.Context) {
 	haTicker := time.NewTicker(l.cfg.HAKeeperCheckInterval.Duration)
 	defer haTicker.Stop()
 
+	// moving task schedule from the ticker normal routine to a
+	// separate goroutine can avoid the hakeeper's health check and tick update
+	// operations being blocked by task schedule, or the tick will be skipped and
+	// can not correctly estimate the time passing.
+	go l.tickerForTaskSchedule(ctx, l.cfg.HAKeeperCheckInterval.Duration)
+
 	for {
 		select {
 		case <-ticker.C:
diff --git a/pkg/logservice/store_hakeeper_check.go b/pkg/logservice/store_hakeeper_check.go
index 3a8230e25..006c8d7f2 100644
--- a/pkg/logservice/store_hakeeper_check.go
+++ b/pkg/logservice/store_hakeeper_check.go
@@ -111,23 +111,32 @@ func (l *store) updateIDAlloc(count uint64) error {
 	return nil
 }
 
-func (l *store) hakeeperCheck() {
+func (l *store) getCheckerStateFromLeader() (*pb.CheckerState, uint64) {
 	isLeader, term, err := l.isLeaderHAKeeper()
 	if err != nil {
 		l.runtime.Logger().Error("failed to get HAKeeper Leader ID", zap.Error(err))
-		return
+		return nil, term
 	}
 
 	if !isLeader {
 		l.taskScheduler.StopScheduleCronTask()
-		return
+		return nil, term
 	}
 	state, err := l.getCheckerState()
 	if err != nil {
 		// TODO: check whether this is temp error
 		l.runtime.Logger().Error("failed to get checker state", zap.Error(err))
+		return nil, term
+	}
+	return state, term
+}
+
+func (l *store) hakeeperCheck() {
+	state, term := l.getCheckerStateFromLeader()
+	if state == nil {
 		return
 	}
+
 	switch state.State {
 	case pb.HAKeeperCreated:
 		l.runtime.Logger().Warn("waiting for initial cluster info to be set, check skipped")
@@ -140,7 +149,6 @@ func (l *store) hakeeperCheck() {
 		l.handleBootstrapFailure()
 	case pb.HAKeeperRunning:
 		l.healthCheck(term, state)
-		l.taskSchedule(state)
 	default:
 		panic("unknown HAKeeper state")
 	}


Test patch:
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index 30f9feffa..227727860 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -494,6 +494,37 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 	runHAKeeperClusterTest(t, fn)
 }
 
+func TestGetCheckerStateFromLeader(t *testing.T) {
+	fn := func(t *testing.T, store *store) {
+		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*10))
+		defer cancel()
+
+		for {
+			select {
+			case <-ctx.Done():
+				t.Error("test deadline reached")
+				return
+
+			default:
+				isLeader, termA, err := store.isLeaderHAKeeper()
+				state, termB := store.getCheckerStateFromLeader()
+				require.NoError(t, err)
+				assert.Equal(t, termB, termA)
+
+				if !isLeader {
+					assert.Equal(t, (*pb.CheckerState)(nil), state)
+				} else {
+					assert.NotEqual(t, (*pb.CheckerState)(nil), state)
+					return
+				}
+				time.Sleep(time.Second)
+			}
+		}
+	}
+
+	runHAKeeperStoreTest(t, false, fn)
+}
+
 func TestGetCheckerState(t *testing.T) {
 	fn := func(t *testing.T, store *store) {
 		state, err := store.getCheckerState()
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 49d44ef55..051c436da 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -17,6 +17,7 @@ package logservice
 import (
 	"context"
 	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/pb/task"
 	"math"
 	"sync/atomic"
 	"testing"
@@ -298,6 +299,97 @@ func TestQueryLog(t *testing.T) {
 	runStoreTest(t, fn)
 }
 
+func proceedHAKeeperToRunning(t *testing.T, store *store) {
+	state, err := store.getCheckerState()
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperCreated, state.State)
+
+	err = store.setInitialClusterInfo(1, 1, 1)
+	assert.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	hb := store.getHeartbeatMessage()
+	_, err = store.addLogStoreHeartbeat(ctx, hb)
+	assert.NoError(t, err)
+
+	state, err = store.getCheckerState()
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperBootstrapping, state.State)
+
+	_, term, err := store.isLeaderHAKeeper()
+	assert.NoError(t, err)
+
+	store.bootstrap(term, state)
+	state, err = store.getCheckerState()
+
+	assert.NoError(t, err)
+	assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State)
+
+	cmd, err := store.getCommandBatch(ctx, store.id())
+	require.NoError(t, err)
+	require.Equal(t, 1, len(cmd.Commands))
+	assert.True(t, cmd.Commands[0].Bootstrapping)
+
+	// handle startReplica to make sure logHeartbeat msg contain shards info,
+	// which used in store.checkBootstrap to determine if all log shards ready
+	service := &Service{store: store}
+	service.handleStartReplica(cmd.Commands[0])
+
+	for state.State != pb.HAKeeperRunning && store.bootstrapCheckCycles > 0 {
+		func() {
+			ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+
+			_, err = store.addLogStoreHeartbeat(ctx, store.getHeartbeatMessage())
+			assert.NoError(t, err)
+
+			store.checkBootstrap(state)
+			state, err = store.getCheckerState()
+			assert.NoError(t, err)
+
+			time.Sleep(time.Millisecond * 100)
+		}()
+	}
+
+	assert.Equal(t, pb.HAKeeperRunning, state.State)
+}
+
+// test if the tickerForTaskSchedule can push forward these routine
+func TestTickerForTaskSchedule(t *testing.T) {
+	fn := func(t *testing.T, store *store, taskService taskservice.TaskService) {
+		// making hakeeper state proceeds to running before test task schedule
+		tickerCxt, tickerCancel := context.WithCancel(context.Background())
+		defer tickerCancel()
+
+		//do task schedule background
+		go store.tickerForTaskSchedule(tickerCxt, time.Millisecond*10)
+
+		proceedHAKeeperToRunning(t, store)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		err := taskService.Create(ctx, task.TaskMetadata{ID: "1234"})
+		assert.NoError(t, err)
+
+		cnUUID := uuid.New().String()
+		cmd := pb.CNStoreHeartbeat{UUID: cnUUID}
+		_, err = store.addCNStoreHeartbeat(ctx, cmd)
+		assert.NoError(t, err)
+
+		// waiting the background taskSchedule operation
+		// schedule task that we created to CN node
+		time.Sleep(time.Millisecond * 100)
+
+		tasks, err := taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID))
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(tasks))
+	}
+
+	runHakeeperTaskServiceTest(t, fn)
+}
+
 func TestHAKeeperTick(t *testing.T) {
 	fn := func(t *testing.T, store *store) {
 		peers := make(map[uint64]dragonboat.Target)