Repository: matrixorigin/matrixone
Pull request #5082 (2022-09-19 07:38:51)

add taskservice argument when create logservice
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:

Create `TaskService` in `main`.
Add `TaskService` argument when create `logservice`


Patch:
diff --git a/cmd/mo-service/main.go b/cmd/mo-service/main.go
index e49534ab6..912784712 100644
--- a/cmd/mo-service/main.go
+++ b/cmd/mo-service/main.go
@@ -26,18 +26,16 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/matrixorigin/matrixone/pkg/cnservice/cnclient"
-	"github.com/matrixorigin/matrixone/pkg/fileservice"
-	"github.com/matrixorigin/matrixone/pkg/sql/compile"
-
 	"github.com/matrixorigin/matrixone/pkg/cnservice"
-
+	"github.com/matrixorigin/matrixone/pkg/cnservice/cnclient"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/stopper"
 	"github.com/matrixorigin/matrixone/pkg/dnservice"
+	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/logservice"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
-
-	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/sql/compile"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/util"
 	"github.com/matrixorigin/matrixone/pkg/util/export"
 	"github.com/matrixorigin/matrixone/pkg/util/metric"
@@ -96,6 +94,9 @@ func startService(cfg *Config, stopper *stopper.Stopper) error {
 		return err
 	}
 
+	// TODO: Use real task storage.
+	ts := taskservice.NewTaskService(taskservice.NewMemTaskStorage())
+
 	if err = initTraceMetric(context.Background(), cfg, stopper, fs); err != nil {
 		return err
 	}
@@ -106,9 +107,9 @@ func startService(cfg *Config, stopper *stopper.Stopper) error {
 	case dnServiceType:
 		return startDNService(cfg, stopper, fs)
 	case logServiceType:
-		return startLogService(cfg, stopper, fs)
+		return startLogService(cfg, stopper, fs, ts)
 	case standaloneServiceType:
-		return startStandalone(cfg, stopper, fs)
+		return startStandalone(cfg, stopper, fs, ts)
 	default:
 		panic("unknown service type")
 	}
@@ -174,9 +175,10 @@ func startLogService(
 	cfg *Config,
 	stopper *stopper.Stopper,
 	fileService fileservice.FileService,
+	taskService taskservice.TaskService,
 ) error {
 	lscfg := cfg.getLogServiceConfig()
-	s, err := logservice.NewService(lscfg, fileService)
+	s, err := logservice.NewService(lscfg, fileService, taskService)
 	if err != nil {
 		panic(err)
 	}
@@ -202,10 +204,11 @@ func startStandalone(
 	cfg *Config,
 	stopper *stopper.Stopper,
 	fileService fileservice.FileService,
+	taskService taskservice.TaskService,
 ) error {
 
 	// start log service
-	if err := startLogService(cfg, stopper, fileService); err != nil {
+	if err := startLogService(cfg, stopper, fileService, taskService); err != nil {
 		return err
 	}
 
diff --git a/pkg/logservice/config.go b/pkg/logservice/config.go
index 11f1e4787..4a83e398c 100644
--- a/pkg/logservice/config.go
+++ b/pkg/logservice/config.go
@@ -26,7 +26,6 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
-	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/util/toml"
 )
 
@@ -60,7 +59,7 @@ type Config struct {
 	// milliseconds.
 	RTTMillisecond uint64 `toml:"rttmillisecond"`
 	// DataDir is the name of the directory for storing all log service data. It
-	// should a locally mount partition with good write and fsync performance.
+	// should a locally mounted partition with good write and fsync performance.
 	DataDir string `toml:"data-dir"`
 	// ServiceAddress is log service's service address that can be reached by
 	// other nodes such as DN nodes.
@@ -154,8 +153,7 @@ type Config struct {
 
 	// HAKeeperClientConfig is the config for HAKeeperClient
 	HAKeeperClientConfig HAKeeperClientConfig
-	// TaskService is the interface for managing async tasks running on CNs.
-	TaskService taskservice.TaskService
+
 	// DisableWorkers disables the HAKeeper ticker and HAKeeper client in tests.
 	// Never set this field to true in production
 	DisableWorkers bool
@@ -297,9 +295,6 @@ func (c *Config) Fill() {
 	if c.FS == nil {
 		c.FS = vfs.Default
 	}
-	if c.TaskService == nil {
-		c.TaskService = taskservice.NewTaskService(taskservice.NewMemTaskStorage())
-	}
 	if c.RTTMillisecond == 0 {
 		c.RTTMillisecond = 200
 	}
@@ -336,6 +331,9 @@ func (c *Config) Fill() {
 	if c.HAKeeperConfig.DNStoreTimeout.Duration == 0 {
 		c.HAKeeperConfig.DNStoreTimeout.Duration = hakeeper.DefaultDNStoreTimeout
 	}
+	if c.HAKeeperConfig.CNStoreTimeout.Duration == 0 {
+		c.HAKeeperConfig.CNStoreTimeout.Duration = hakeeper.DefaultCNStoreTimeout
+	}
 	if c.HeartbeatInterval.Duration == 0 {
 		c.HeartbeatInterval.Duration = defaultHeartbeatInterval
 	}
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index eee1a4e81..236f7e949 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -31,6 +31,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 )
 
 var (
@@ -82,13 +83,14 @@ type Service struct {
 func NewService(
 	cfg Config,
 	fileService fileservice.FileService,
+	taskService taskservice.TaskService,
 	opts ...Option,
 ) (*Service, error) {
 	cfg.Fill()
 	if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
-	store, err := newLogStore(cfg)
+	store, err := newLogStore(cfg, taskService)
 	if err != nil {
 		plog.Errorf("failed to create log store %v", err)
 		return nil, err
diff --git a/pkg/logservice/service_wrap.go b/pkg/logservice/service_wrap.go
index 3e4c34fe6..48741f01c 100644
--- a/pkg/logservice/service_wrap.go
+++ b/pkg/logservice/service_wrap.go
@@ -18,6 +18,7 @@ import (
 	"github.com/lni/dragonboat/v4"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 )
 
 type WrappedService struct {
@@ -27,9 +28,10 @@ type WrappedService struct {
 func NewWrappedService(
 	c Config,
 	fileService fileservice.FileService,
+	taskService taskservice.TaskService,
 	opts ...Option,
 ) (*WrappedService, error) {
-	svc, err := NewService(c, fileService, opts...)
+	svc, err := NewService(c, fileService, taskService, opts...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 721f55ff8..6c96865eb 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -37,6 +37,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/task"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 )
 
 var (
@@ -142,7 +143,7 @@ type store struct {
 	}
 }
 
-func newLogStore(cfg Config) (*store, error) {
+func newLogStore(cfg Config, taskService taskservice.TaskService) (*store, error) {
 	nh, err := dragonboat.NewNodeHost(getNodeHostConfig(cfg))
 	if err != nil {
 		return nil, err
@@ -154,7 +155,7 @@ func newLogStore(cfg Config) (*store, error) {
 		cfg:           cfg,
 		nh:            nh,
 		checker:       checkers.NewCoordinator(hakeeperConfig),
-		taskScheduler: task.NewTaskScheduler(cfg.TaskService, hakeeperConfig),
+		taskScheduler: task.NewTaskScheduler(taskService, hakeeperConfig),
 		alloc:         newIDAllocator(),
 		stopper:       stopper.NewStopper("log-store"),
 		tickerStopper: stopper.NewStopper("hakeeper-ticker"),


Test patch:
diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go
index 375199ddb..c2c4fd826 100644
--- a/pkg/logservice/client_test.go
+++ b/pkg/logservice/client_test.go
@@ -30,6 +30,7 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 )
 
@@ -40,6 +41,7 @@ func runClientTest(t *testing.T,
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	service, err := NewService(cfg,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -128,6 +130,7 @@ func TestClientCanBeConnectedByReverseProxy(t *testing.T) {
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	service, err := NewService(cfg,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/logservice/hakeeper_client_test.go b/pkg/logservice/hakeeper_client_test.go
index fb17cd8dc..550956d56 100644
--- a/pkg/logservice/hakeeper_client_test.go
+++ b/pkg/logservice/hakeeper_client_test.go
@@ -32,6 +32,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 )
 
@@ -307,6 +308,7 @@ func testNotHAKeeperErrorIsHandled(t *testing.T, fn func(*testing.T, *managedHAK
 	cfg1.Fill()
 	service1, err := NewService(cfg1,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -318,6 +320,7 @@ func testNotHAKeeperErrorIsHandled(t *testing.T, fn func(*testing.T, *managedHAK
 	cfg2.Fill()
 	service2, err := NewService(cfg2,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/logservice/service_commands_test.go b/pkg/logservice/service_commands_test.go
index 80df6e401..042252680 100644
--- a/pkg/logservice/service_commands_test.go
+++ b/pkg/logservice/service_commands_test.go
@@ -27,6 +27,7 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 )
 
@@ -50,6 +51,7 @@ func TestBackgroundTickAndHeartbeat(t *testing.T) {
 	cfg.Fill()
 	service, err := NewService(cfg,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index 02b900caa..86427b46a 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -32,6 +32,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	hapkg "github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 )
 
@@ -66,6 +67,7 @@ func runServiceTest(t *testing.T,
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	service, err := NewService(cfg,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -109,6 +111,7 @@ func TestNewService(t *testing.T) {
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	service, err := NewService(cfg,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -557,6 +560,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	cfg1.Fill()
 	service1, err := NewService(cfg1,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -571,6 +575,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	cfg2.Fill()
 	service2, err := NewService(cfg2,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -690,6 +695,7 @@ func TestGossipInSimulatedCluster(t *testing.T) {
 		configs = append(configs, cfg)
 		service, err := NewService(cfg,
 			testutil.NewFS(),
+			taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 			WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 				return true
 			}),
@@ -798,6 +804,7 @@ func TestGossipInSimulatedCluster(t *testing.T) {
 	time.Sleep(2 * time.Second)
 	service, err := NewService(configs[12],
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index 32fd72473..5d55318f9 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -29,6 +29,8 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/pb/task"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 )
 
@@ -122,7 +124,7 @@ func runHAKeeperStoreTest(t *testing.T, startLogReplica bool, fn func(*testing.T
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := getTestStore(cfg, startLogReplica)
+	store, err := getTestStore(cfg, startLogReplica, nil)
 	assert.NoError(t, err)
 	defer func() {
 		assert.NoError(t, store.close())
@@ -133,6 +135,24 @@ func runHAKeeperStoreTest(t *testing.T, startLogReplica bool, fn func(*testing.T
 	fn(t, store)
 }
 
+func runHakeeperTaskServiceTest(t *testing.T, fn func(*testing.T, *store, taskservice.TaskService)) {
+	defer leaktest.AfterTest(t)()
+	cfg := getStoreTestConfig()
+	cfg.HAKeeperConfig.CNStoreTimeout.Duration = 5 * time.Second
+	defer vfs.ReportLeakedFD(cfg.FS, t)
+
+	taskService := taskservice.NewTaskService(taskservice.NewMemTaskStorage())
+	store, err := getTestStore(cfg, false, taskService)
+	assert.NoError(t, err)
+	defer func() {
+		assert.NoError(t, store.close())
+	}()
+	peers := make(map[uint64]dragonboat.Target)
+	peers[1] = store.id()
+	assert.NoError(t, store.startHAKeeperReplica(1, peers, false))
+	fn(t, store, taskService)
+}
+
 func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	defer leaktest.AfterTest(t)()
 	cfg1 := Config{
@@ -150,6 +170,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg1.HAKeeperConfig.TickPerSecond = 10
 	cfg1.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
 	cfg1.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
+	cfg1.HAKeeperConfig.CNStoreTimeout.Duration = 5 * time.Second
 	cfg2 := Config{
 		UUID:                uuid.New().String(),
 		FS:                  vfs.NewStrictMem(),
@@ -165,6 +186,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg2.HAKeeperConfig.TickPerSecond = 10
 	cfg2.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
 	cfg2.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
+	cfg2.HAKeeperConfig.CNStoreTimeout.Duration = 5 * time.Second
 	cfg3 := Config{
 		UUID:                uuid.New().String(),
 		FS:                  vfs.NewStrictMem(),
@@ -180,6 +202,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg3.HAKeeperConfig.TickPerSecond = 10
 	cfg3.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
 	cfg3.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
+	cfg3.HAKeeperConfig.CNStoreTimeout.Duration = 5 * time.Second
 	cfg4 := Config{
 		UUID:                uuid.New().String(),
 		FS:                  vfs.NewStrictMem(),
@@ -195,9 +218,11 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg4.HAKeeperConfig.TickPerSecond = 10
 	cfg4.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
 	cfg4.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
+	cfg4.HAKeeperConfig.CNStoreTimeout.Duration = 5 * time.Second
 	cfg1.Fill()
 	service1, err := NewService(cfg1,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -209,6 +234,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg2.Fill()
 	service2, err := NewService(cfg2,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -220,6 +246,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg3.Fill()
 	service3, err := NewService(cfg3,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -231,6 +258,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg4.Fill()
 	service4, err := NewService(cfg4,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -441,7 +469,7 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 						break
 					}
 				}
-				plog.Infof("dnRepairted %t, logRepaired %t", dnRepaired, logRepaired)
+				plog.Infof("dnRepaired %t, logRepaired %t", dnRepaired, logRepaired)
 				if !logRepaired || !dnRepaired {
 					return false, nil
 				} else {
@@ -580,3 +608,207 @@ func testBootstrap(t *testing.T, fail bool) {
 	}
 	runHAKeeperStoreTest(t, false, fn)
 }
+
+func TestTaskSchedulerCanScheduleTasksToCNs(t *testing.T) {
+	fn := func(t *testing.T, store *store, taskService taskservice.TaskService) {
+		state, err := store.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperCreated, state.State)
+		require.NoError(t, store.setInitialClusterInfo(1, 1, 1))
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperBootstrapping, state.State)
+		m := store.getHeartbeatMessage()
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		_, err = store.addLogStoreHeartbeat(ctx, m)
+		assert.NoError(t, err)
+
+		_, term, err := store.isLeaderHAKeeper()
+		require.NoError(t, err)
+
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		store.bootstrap(term, state)
+
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State)
+		assert.Equal(t, uint64(checkBootstrapCycles), store.bootstrapCheckCycles)
+		require.NotNil(t, store.bootstrapMgr)
+		assert.False(t, store.bootstrapMgr.CheckBootstrap(state.LogState))
+
+		cb, err := store.getCommandBatch(ctx, store.id())
+		require.NoError(t, err)
+		require.Equal(t, 1, len(cb.Commands))
+		assert.True(t, cb.Commands[0].Bootstrapping)
+		service := &Service{store: store}
+		service.handleStartReplica(cb.Commands[0])
+
+		for i := 0; i < 100; i++ {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+			m := store.getHeartbeatMessage()
+			_, err = store.addLogStoreHeartbeat(ctx, m)
+			assert.NoError(t, err)
+
+			state, err = store.getCheckerState()
+			require.NoError(t, err)
+			store.checkBootstrap(state)
+
+			state, err = store.getCheckerState()
+			require.NoError(t, err)
+			if state.State != pb.HAKeeperRunning {
+				time.Sleep(50 * time.Millisecond)
+			} else {
+				break
+			}
+			if i == 2999 {
+				t.Fatalf("failed to complete bootstrap")
+			}
+		}
+
+		cnUUID1 := uuid.New().String()
+		cnMsg1 := pb.CNStoreHeartbeat{UUID: cnUUID1}
+		err = store.addCNStoreHeartbeat(ctx, cnMsg1)
+		assert.NoError(t, err)
+		err = taskService.Create(ctx, task.TaskMetadata{ID: "a"})
+		assert.NoError(t, err)
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		tasks, err := taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID1))
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(tasks))
+		store.taskSchedule(state)
+		tasks, err = taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID1))
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(tasks))
+
+		cnUUID2 := uuid.New().String()
+		cnMsg2 := pb.CNStoreHeartbeat{UUID: cnUUID2}
+		err = store.addCNStoreHeartbeat(ctx, cnMsg2)
+		assert.NoError(t, err)
+		err = taskService.Create(ctx, task.TaskMetadata{ID: "b"})
+		assert.NoError(t, err)
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		tasks, err = taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID2))
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(tasks))
+		store.taskSchedule(state)
+		tasks, err = taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID2))
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(tasks))
+	}
+	runHakeeperTaskServiceTest(t, fn)
+}
+
+func TestTaskSchedulerCanReScheduleExpiredTasks(t *testing.T) {
+	fn := func(t *testing.T, store *store, taskService taskservice.TaskService) {
+		state, err := store.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperCreated, state.State)
+		require.NoError(t, store.setInitialClusterInfo(1, 1, 1))
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperBootstrapping, state.State)
+		m := store.getHeartbeatMessage()
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		_, err = store.addLogStoreHeartbeat(ctx, m)
+		assert.NoError(t, err)
+
+		_, term, err := store.isLeaderHAKeeper()
+		require.NoError(t, err)
+
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		store.bootstrap(term, state)
+
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State)
+		assert.Equal(t, uint64(checkBootstrapCycles), store.bootstrapCheckCycles)
+		require.NotNil(t, store.bootstrapMgr)
+		assert.False(t, store.bootstrapMgr.CheckBootstrap(state.LogState))
+
+		cb, err := store.getCommandBatch(ctx, store.id())
+		require.NoError(t, err)
+		require.Equal(t, 1, len(cb.Commands))
+		assert.True(t, cb.Commands[0].Bootstrapping)
+		service := &Service{store: store}
+		service.handleStartReplica(cb.Commands[0])
+
+		for i := 0; i < 100; i++ {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+			m := store.getHeartbeatMessage()
+			_, err = store.addLogStoreHeartbeat(ctx, m)
+			assert.NoError(t, err)
+
+			state, err = store.getCheckerState()
+			require.NoError(t, err)
+			store.checkBootstrap(state)
+
+			state, err = store.getCheckerState()
+			require.NoError(t, err)
+			if state.State != pb.HAKeeperRunning {
+				time.Sleep(50 * time.Millisecond)
+			} else {
+				break
+			}
+			if i == 2999 {
+				t.Fatalf("failed to complete bootstrap")
+			}
+		}
+
+		cnUUID1 := uuid.New().String()
+		cnMsg1 := pb.CNStoreHeartbeat{UUID: cnUUID1}
+		err = store.addCNStoreHeartbeat(ctx, cnMsg1)
+		assert.NoError(t, err)
+		err = taskService.Create(ctx, task.TaskMetadata{ID: "a"})
+		assert.NoError(t, err)
+		state, err = store.getCheckerState()
+		require.NoError(t, err)
+		tasks, err := taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID1))
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(tasks))
+		store.taskSchedule(state)
+		tasks, err = taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID1))
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(tasks))
+
+		cnUUID2 := uuid.New().String()
+		for i := 0; i < 1000; i++ {
+			plog.Infof("iteration %d", i)
+			tn := func() bool {
+				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+				defer cancel()
+				cnMsg2 := pb.CNStoreHeartbeat{UUID: cnUUID2}
+				err = store.addCNStoreHeartbeat(ctx, cnMsg2)
+				assert.NoError(t, err)
+				state, err = store.getCheckerState()
+				require.NoError(t, err)
+				store.taskSchedule(state)
+				tasks, err = taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID2))
+				assert.NoError(t, err)
+				if len(tasks) == 0 {
+					plog.Infof("no task found")
+					time.Sleep(50 * time.Millisecond)
+				} else {
+					tasks, err = taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID1))
+					assert.Equal(t, 0, len(tasks))
+					return true
+				}
+				return false
+			}
+			completed := tn()
+			if completed {
+				return
+			}
+			time.Sleep(100 * time.Millisecond)
+		}
+		t.Fatalf("failed to reschedule expired tasks")
+	}
+	runHakeeperTaskServiceTest(t, fn)
+}
diff --git a/pkg/logservice/store_metadata_test.go b/pkg/logservice/store_metadata_test.go
index 51eae81db..bfd0226b7 100644
--- a/pkg/logservice/store_metadata_test.go
+++ b/pkg/logservice/store_metadata_test.go
@@ -90,7 +90,7 @@ func TestStartReplicas(t *testing.T) {
 	cfg.Fill()
 	require.NoError(t, mkdirAll(cfg.DataDir, cfg.FS))
 	func() {
-		store, err := getTestStore(cfg, false)
+		store, err := getTestStore(cfg, false, nil)
 		require.NoError(t, err)
 		members := make(map[uint64]dragonboat.Target)
 		members[1] = store.id()
@@ -101,7 +101,7 @@ func TestStartReplicas(t *testing.T) {
 		require.NoError(t, store.startReplica(20, 1, members, false))
 	}()
 
-	store, err := getTestStore(cfg, false)
+	store, err := getTestStore(cfg, false, nil)
 	require.NoError(t, err)
 	defer func() {
 		require.NoError(t, store.close())
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 6dec48b1b..901eec869 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -16,6 +16,7 @@ package logservice
 
 import (
 	"context"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"math"
 	"sync/atomic"
 	"testing"
@@ -73,7 +74,7 @@ func TestStoreCanBeCreatedAndClosed(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := newLogStore(cfg)
+	store, err := newLogStore(cfg, nil)
 	assert.NoError(t, err)
 	plog.Infof("1")
 	defer func() {
@@ -82,8 +83,8 @@ func TestStoreCanBeCreatedAndClosed(t *testing.T) {
 	plog.Infof("2")
 }
 
-func getTestStore(cfg Config, startLogReplica bool) (*store, error) {
-	store, err := newLogStore(cfg)
+func getTestStore(cfg Config, startLogReplica bool, taskService taskservice.TaskService) (*store, error) {
+	store, err := newLogStore(cfg, taskService)
 	if err != nil {
 		return nil, err
 	}
@@ -102,7 +103,7 @@ func TestHAKeeperCanBeStarted(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := newLogStore(cfg)
+	store, err := newLogStore(cfg, nil)
 	assert.NoError(t, err)
 	peers := make(map[uint64]dragonboat.Target)
 	peers[2] = store.nh.ID()
@@ -117,7 +118,7 @@ func TestStateMachineCanBeStarted(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := getTestStore(cfg, true)
+	store, err := getTestStore(cfg, true, nil)
 	assert.NoError(t, err)
 	defer func() {
 		assert.NoError(t, store.close())
@@ -129,7 +130,7 @@ func TestReplicaCanBeStopped(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := getTestStore(cfg, true)
+	store, err := getTestStore(cfg, true, nil)
 	assert.NoError(t, err)
 	defer func() {
 		assert.NoError(t, store.close())
@@ -143,7 +144,7 @@ func runStoreTest(t *testing.T, fn func(*testing.T, *store)) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := getTestStore(cfg, true)
+	store, err := getTestStore(cfg, true, nil)
 	assert.NoError(t, err)
 	defer func() {
 		assert.NoError(t, store.close())
@@ -257,7 +258,7 @@ func TestQueryLog(t *testing.T) {
 		assert.Equal(t, 1, len(entries))
 		assert.Equal(t, uint64(4), lsn)
 		assert.Equal(t, entries[0].Data, cmd)
-		// lease holder ID update cmd at entry index 3
+		// leaseholder ID update cmd at entry index 3
 		entries, lsn, err = store.queryLog(ctx, 1, 3, math.MaxUint64)
 		assert.NoError(t, err)
 		assert.Equal(t, 2, len(entries))
@@ -427,7 +428,7 @@ func getTestStores() (*store, *store, error) {
 		GossipSeedAddresses: []string{"127.0.0.1:9011", "127.0.0.1:9012"},
 	}
 	cfg1.Fill()
-	store1, err := newLogStore(cfg1)
+	store1, err := newLogStore(cfg1, nil)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -442,7 +443,7 @@ func getTestStores() (*store, *store, error) {
 		GossipSeedAddresses: []string{"127.0.0.1:9011", "127.0.0.1:9012"},
 	}
 	cfg2.Fill()
-	store2, err := newLogStore(cfg2)
+	store2, err := newLogStore(cfg2, nil)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/pkg/logservice/testclient.go b/pkg/logservice/testclient.go
index a6d2b5941..98687dc11 100644
--- a/pkg/logservice/testclient.go
+++ b/pkg/logservice/testclient.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 )
 
@@ -39,6 +40,7 @@ func NewTestService(fs vfs.FS) (*Service, ClientConfig, error) {
 	cfg.Fill()
 	service, err := NewService(cfg,
 		testutil.NewFS(),
+		taskservice.NewTaskService(taskservice.NewMemTaskStorage()),
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/tests/service/logservice.go b/pkg/tests/service/logservice.go
index 59d7544c7..bceadb568 100644
--- a/pkg/tests/service/logservice.go
+++ b/pkg/tests/service/logservice.go
@@ -28,6 +28,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/logservice"
 	logpb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 )
 
 var (
@@ -138,9 +139,10 @@ type logOptions []logservice.Option
 func newLogService(
 	cfg logservice.Config,
 	fs fileservice.FileService,
+	ts taskservice.TaskService,
 	opts logOptions,
 ) (LogService, error) {
-	svc, err := logservice.NewWrappedService(cfg, fs, opts...)
+	svc, err := logservice.NewWrappedService(cfg, fs, ts, opts...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/tests/service/service.go b/pkg/tests/service/service.go
index f136e5b0f..8303851a4 100644
--- a/pkg/tests/service/service.go
+++ b/pkg/tests/service/service.go
@@ -1200,7 +1200,7 @@ func (c *testCluster) initLogServices() []LogService {
 	for i := 0; i < batch; i++ {
 		cfg := c.log.cfgs[i]
 		opt := c.log.opts[i]
-		ls, err := newLogService(cfg, testutil.NewFS(), opt)
+		ls, err := newLogService(cfg, testutil.NewFS(), testutil.NewTaskService(), opt)
 		require.NoError(c.t, err)
 
 		c.logger.Info(
diff --git a/pkg/testutil/util_new.go b/pkg/testutil/util_new.go
index 92e9cba2a..087c22ce2 100644
--- a/pkg/testutil/util_new.go
+++ b/pkg/testutil/util_new.go
@@ -26,6 +26,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/vm/mheap"
 	"github.com/matrixorigin/matrixone/pkg/vm/mmu/guest"
 	"github.com/matrixorigin/matrixone/pkg/vm/mmu/host"
@@ -85,6 +86,10 @@ func NewFS() *fileservice.FileServices {
 	return fs
 }
 
+func NewTaskService() taskservice.TaskService {
+	return taskservice.NewTaskService(taskservice.NewMemTaskStorage())
+}
+
 func NewBatch(ts []types.Type, random bool, n int, m *mheap.Mheap) *batch.Batch {
 	bat := batch.NewWithSize(len(ts))
 	bat.InitZsOne(n)